Chris Amow
fced18528b
History was only fetched by the startup seed, so a Schwab outage stayed a hole until the next deploy — Sept 9 to 24 after a refresh token expired. A reconnect more than two minutes past the last bar now fetches the gap from Yahoo, fills empty buckets only, refolds the live forming buckets, rebuilds levels without alerting, and resyncs every socket. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
83 lines
3.3 KiB
Python
83 lines
3.3 KiB
Python
from collections import defaultdict, deque
|
|
from typing import Protocol
|
|
|
|
from app.bars.models import Bar, Timeframe
|
|
from app.config import DEFAULT_MAX_BARS_PER_TF
|
|
|
|
|
|
class BarStore(Protocol):
|
|
def put(self, bar: Bar) -> None: ...
|
|
|
|
def get(self, tf: Timeframe, limit: int | None = None) -> list[Bar]: ...
|
|
|
|
|
|
class InMemoryBarStore:
|
|
def __init__(self, max_bars_per_tf: int = DEFAULT_MAX_BARS_PER_TF):
|
|
self._bars: dict[Timeframe, deque[Bar]] = defaultdict(
|
|
lambda: deque(maxlen=max_bars_per_tf)
|
|
)
|
|
|
|
# How far back from the tail a late bar may still land. A closed minute bar
|
|
# arrives a beat after the ticks that opened the next minute, so it is
|
|
# rarely more than a bucket or two behind.
|
|
LATE_BAR_LOOKBACK = 8
|
|
|
|
def put(self, bar: Bar) -> None:
|
|
"""Store a bar, replacing the bucket it belongs to.
|
|
|
|
Matching only the tail was enough while one closed bar arrived per
|
|
minute. With tick-built bars a minute's authoritative bar shows up
|
|
*after* ticks have already opened the next one, so the exchange's own
|
|
figures were being dropped and the approximation left in place forever.
|
|
"""
|
|
bars = self._bars[bar.tf]
|
|
if not bars or bar.t > bars[-1].t:
|
|
bars.append(bar)
|
|
return
|
|
for index in range(len(bars) - 1, max(-1, len(bars) - self.LATE_BAR_LOOKBACK - 1), -1):
|
|
if bars[index].t == bar.t:
|
|
# A provisional bar must never overwrite a settled one: ticks
|
|
# keep arriving for a minute the exchange has already closed.
|
|
if bars[index].closed and not bar.closed:
|
|
return
|
|
bars[index] = bar
|
|
return
|
|
if bars[index].t < bar.t:
|
|
# Buckets are ordered, so nothing further back can match.
|
|
return
|
|
|
|
def fill(self, bars: list[Bar]) -> int:
|
|
"""Insert history into buckets the store has no bar for.
|
|
|
|
``put`` only lands a bar at the tail or a few buckets behind it, so a
|
|
stretch missed while the stream was down cannot reach it — the live
|
|
bars that arrived on reconnect are already newer. Existing bars always
|
|
win: they are the live source's own figures, and the bucket either side
|
|
of the hole is the live aggregator's to finish. Returns how many bars
|
|
were inserted.
|
|
"""
|
|
added = 0
|
|
by_tf: dict[Timeframe, list[Bar]] = defaultdict(list)
|
|
for bar in bars:
|
|
by_tf[bar.tf].append(bar)
|
|
for tf, incoming in by_tf.items():
|
|
held = self._bars[tf]
|
|
merged = {bar.t: bar for bar in held}
|
|
for bar in incoming:
|
|
if bar.t not in merged:
|
|
merged[bar.t] = bar
|
|
added += 1
|
|
if len(merged) == len(held):
|
|
continue
|
|
ordered = [merged[t] for t in sorted(merged)]
|
|
held.clear()
|
|
# maxlen keeps the newest, which is the history the chart shows.
|
|
held.extend(ordered)
|
|
return added
|
|
|
|
def get(self, tf: Timeframe, limit: int | None = None) -> list[Bar]:
|
|
bars = list(self._bars[tf])
|
|
return bars[-limit:] if limit is not None else bars
|
|
|
|
def counts(self) -> dict[str, int]:
|
|
return {tf.value: len(self._bars[tf]) for tf in Timeframe}
|