import asyncio from dataclasses import dataclass, field from app.bars.models import Bar, Timeframe from app.bars.aggregator import Aggregator from app.analysis.levels import Level from app.analysis.moving_averages import build_ma_levels from app.bars.store import InMemoryBarStore from app.config import Settings from app.market.factory import live_source, seed_source from app.market.stream import StreamService @dataclass class Runtime: settings: Settings store: InMemoryBarStore = field(init=False) stream: StreamService = field(init=False) subscribers: set[asyncio.Queue[dict]] = field(default_factory=set) aggregator: Aggregator = field(init=False) levels: list[Level] = field(default_factory=list) def __post_init__(self) -> None: self.store = InMemoryBarStore(self.settings.max_bars_per_tf) self.aggregator = Aggregator(self.settings.enabled_timeframes) self.stream = StreamService(live_source(self.settings), self.settings.yahoo_symbol) self.stream.add_handler(self.on_bar) async def on_bar(self, bar: Bar) -> None: for aggregated in self.aggregator.update(bar): self.store.put(aggregated) self.broadcast({"type": "bar", "bar": aggregated}) if self.settings.ma_sets.get(aggregated.tf): self.rebuild_levels() def broadcast(self, event: dict) -> None: for queue in self.subscribers.copy(): if queue.full(): queue.get_nowait() queue.put_nowait(event) def rebuild_levels(self) -> None: self.levels = build_ma_levels( {tf: self.store.get(tf) for tf in self.settings.ma_sets}, self.settings.ma_sets, ) self.broadcast({"type": "levels", "levels": self.levels}) async def start(self) -> asyncio.Task: try: source = seed_source(self.settings) await self.stream.seed(source, Timeframe.H1, self.settings.seed_1h_range) await self.stream.seed(source, Timeframe.M1, self.settings.seed_1m_range) except Exception: # A transient seed failure must not prevent the live stream or UI starting. pass return asyncio.create_task(self.stream.run(), name="market-stream")