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>
89 lines
3.3 KiB
Python
89 lines
3.3 KiB
Python
import asyncio
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
|
|
from app.bars.models import Bar, Timeframe
|
|
from app.market.base import MarketDataSource
|
|
|
|
logger = logging.getLogger(__name__)
|
|
BarHandler = Callable[[Bar], Awaitable[None]]
|
|
|
|
|
|
class StreamService:
|
|
def __init__(self, source: MarketDataSource, symbol: str):
|
|
self.source = source
|
|
self.symbol = symbol
|
|
self.status = "disconnected"
|
|
self.last_error: str | None = None
|
|
self.last_bar_t: int | None = None
|
|
self._handlers: list[BarHandler] = []
|
|
self._stop = asyncio.Event()
|
|
self.on_drop = None
|
|
# Called with (last bar before the outage, first bar after it) when a
|
|
# new connection opens further past the last bar than this. Reconnect
|
|
# alone only resumes the present; nothing else fetches what was missed.
|
|
self.on_resume: Callable[[int, int], None] | None = None
|
|
self.resume_gap_seconds = 120
|
|
self.reconnect_seconds = 5.0
|
|
|
|
def add_handler(self, handler: BarHandler) -> None:
|
|
self._handlers.append(handler)
|
|
|
|
async def seed(
|
|
self,
|
|
source: MarketDataSource | None,
|
|
tf: Timeframe,
|
|
range_: str,
|
|
symbol: str | None = None,
|
|
) -> None:
|
|
# The seed source names the instrument differently from the live one:
|
|
# Yahoo says ES=F where Schwab says /ES.
|
|
if source is None or not source.supports_history():
|
|
return
|
|
bars = await source.history(symbol or self.symbol, tf, None, None, range_=range_)
|
|
for bar in bars:
|
|
await self._emit(bar)
|
|
|
|
async def _emit(self, bar: Bar) -> None:
|
|
self.last_bar_t = max(self.last_bar_t or bar.t, bar.t)
|
|
for handler in self._handlers:
|
|
await handler(bar)
|
|
|
|
async def run(self) -> None:
|
|
while not self._stop.is_set():
|
|
first = True
|
|
try:
|
|
async for bar in self.source.stream(self.symbol):
|
|
self.status = "replay" if self.source.name == "replay" else "connected"
|
|
self.last_error = None
|
|
before = self.last_bar_t
|
|
await self._emit(bar)
|
|
if first:
|
|
first = False
|
|
if (
|
|
self.on_resume is not None
|
|
and before is not None
|
|
and bar.t - before > self.resume_gap_seconds
|
|
):
|
|
self.on_resume(before, bar.t)
|
|
if self._stop.is_set():
|
|
break
|
|
if self.source.name == "replay":
|
|
return
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc:
|
|
was_up = self.status == "connected"
|
|
self.last_error = str(exc)
|
|
logger.exception("Market stream failed; reconnecting")
|
|
if was_up and self.on_drop:
|
|
self.on_drop(str(exc))
|
|
self.status = "disconnected"
|
|
try:
|
|
await asyncio.wait_for(self._stop.wait(), timeout=self.reconnect_seconds)
|
|
except TimeoutError:
|
|
pass
|
|
|
|
def stop(self) -> None:
|
|
self._stop.set()
|
|
self.status = "disconnected"
|