""" ========================================= CV6 AI Trading OS Market Stream Engine Broker-Independent Live Data Feed ========================================= """ import asyncio import random import time from dataclasses import dataclass, asdict from typing import Dict, Optional, Callable from loguru import logger @dataclass class TickData: """ Normalized tick — broker independent. Fields: LTP, Volume, Bid, Ask, OHLC, Timestamp """ symbol: str ltp: float volume: int bid: float ask: float open: float high: float low: float close: float timestamp: float change: float = 0.0 change_pct: float = 0.0 broker: str = "simulation" def to_ws(self) -> dict: return {"type": "tick", **asdict(self)} class MarketStreamEngine: """ Broker-independent streaming engine. Register broker feeds via register_feed(). Falls back to simulation if no feed registered. """ # Seed prices for Indian market symbols _SEEDS: Dict[str, float] = { "NIFTY": 23500.0, "BANKNIFTY": 49000.0, "FINNIFTY": 23000.0, "RELIANCE": 2850.0, "TCS": 4200.0, "INFY": 1750.0, "HDFCBANK": 1680.0, "ICICIBANK": 1280.0, "SBIN": 830.0, "TATAMOTORS": 975.0, "WIPRO": 560.0, "ADANIENT": 2450.0, } def __init__(self, connection_manager): self.mgr = connection_manager self._tasks: Dict[str, asyncio.Task] = {} self._feeds: Dict[str, Callable] = {} self._last: Dict[str, TickData] = {} self._sim_price: Dict[str, float] = {} # PHASE-2 FIX (Goal 7): generic live-data provider, checked for ANY # symbol before falling back to simulation. register_feed() above # only overrides a single symbol and nothing ever called it — clients # can request arbitrary symbols dynamically, so a per-symbol-only # registration could never cover them. set_live_provider() is called # once (from main.py, after the real broker feed connects) and from # then on every symbol tries the real feed first. self._live_provider: Optional[Callable] = None def register_feed(self, symbol: str, feed_fn: Callable) -> None: """Plug in a broker feed for one specific symbol: async def feed(symbol) -> TickData""" self._feeds[symbol.upper()] = feed_fn logger.info(f"[Stream] Broker feed registered: {symbol}") def set_live_provider(self, feed_fn: Optional[Callable]) -> None: """ Plug in a broker feed used for ANY symbol not otherwise overridden by register_feed(). feed_fn: async def feed(symbol) -> TickData, and it must raise/return falsy if it has no fresh real data for that symbol so _fetch() can fall back to simulation. """ self._live_provider = feed_fn logger.info(f"[Stream] Live data provider {'registered' if feed_fn else 'cleared'}") async def start(self, symbol: str, interval: float = 1.0) -> None: symbol = symbol.upper() if symbol in self._tasks and not self._tasks[symbol].done(): return self._tasks[symbol] = asyncio.create_task( self._loop(symbol, interval) ) logger.info(f"[Stream] Started: {symbol}") async def stop(self, symbol: str) -> None: symbol = symbol.upper() t = self._tasks.pop(symbol, None) if t and not t.done(): t.cancel() async def stop_all(self) -> None: for sym in list(self._tasks): await self.stop(sym) async def _loop(self, symbol: str, interval: float) -> None: while True: try: tick = await self._fetch(symbol) self._last[symbol] = tick await self.mgr.broadcast_symbol(symbol, tick.to_ws()) except asyncio.CancelledError: return except Exception as e: logger.error(f"[Stream] {symbol} error: {e}") await asyncio.sleep(interval) async def _fetch(self, symbol: str) -> TickData: if symbol in self._feeds: try: return await self._feeds[symbol](symbol) except Exception as e: logger.warning(f"[Stream] Per-symbol feed error {symbol}: {e} → trying live provider") if self._live_provider is not None: try: tick = await self._live_provider(symbol) if tick is not None: return tick except Exception as e: logger.debug(f"[Stream] Live provider has no real data for {symbol}: {e} → using sim") return self._simulate(symbol) def _simulate(self, symbol: str) -> TickData: base = self._sim_price.get(symbol) or self._SEEDS.get(symbol, 1000.0) chg_pct = random.uniform(-0.002, 0.002) ltp = round(base * (1 + chg_pct), 2) self._sim_price[symbol] = ltp spread = round(ltp * 0.0003, 2) open_p = round(base * random.uniform(0.998, 1.002), 2) return TickData( symbol=symbol, ltp=ltp, volume=random.randint(50000, 5000000), bid=round(ltp - spread, 2), ask=round(ltp + spread, 2), open=open_p, high=round(max(open_p, ltp) * 1.001, 2), low=round(min(open_p, ltp) * 0.999, 2), close=base, timestamp=time.time(), change=round(ltp - base, 2), change_pct=round(chg_pct * 100, 4), broker="simulation", ) def last_tick(self, symbol: str) -> Optional[dict]: t = self._last.get(symbol.upper()) return t.to_ws() if t else None def streaming(self) -> list: return [s for s, t in self._tasks.items() if not t.done()] # ── PHASE-2.5 (Goal 3): explicit LIVE vs SIMULATED status ─────────────── def status(self) -> dict: """ Per-symbol data source for every symbol currently streaming, plus whether a live provider is registered at all — lets a client (or the /ws/status REST endpoint) tell definitively whether it's getting real Angel One ticks or a simulated fallback, instead of having to infer it from the `broker` field buried in each tick payload. """ sources = {sym: (t.broker if t else "unknown") for sym, t in self._last.items()} return { "live_provider_registered": self._live_provider is not None, "streaming_symbols": self.streaming(), "symbol_sources": sources, # e.g. {"RELIANCE": "angelone", "XYZ": "simulation"} "live_count": sum(1 for v in sources.values() if v not in ("simulation", "unknown")), "simulated_count": sum(1 for v in sources.values() if v == "simulation"), } # Singletons — imported by router from app.websocket.websocket_manager import manager as _mgr stream_engine = MarketStreamEngine(_mgr) # ── PHASE-2.5 (Goal 3) ─────────────────────────────────────────────────────── def connect_angel_one_feed() -> bool: """ Bridge the real Angel One tick feed (app.market.live_feed) into this engine's generic live-data provider, so every /ws/live, /ws/market, /ws/ticks client gets real ticks instead of simulation. PHASE-2 originally wired this ONLY from main.py's startup auto-connect thread — a broker connected later via POST /broker/connect (manual reconnect) never got the bridge wired at all. This function is now the single reusable entry point, called from both main.py's startup path AND app/broker/broker_manager.py's manual connect() path. Safe to call multiple times (idempotent — just re-registers the same provider function). """ try: from app.market.live_feed import live_feed async def _angel_live_provider(symbol: str) -> Optional[TickData]: tick = live_feed.get_tick(symbol) if not tick or not tick.is_fresh or tick.ltp <= 0: return None spread = round(tick.ltp * 0.0005, 2) or 0.01 return TickData( symbol=symbol, ltp=tick.ltp, volume=tick.volume, bid=round(tick.ltp - spread, 2), ask=round(tick.ltp + spread, 2), open=tick.open, high=tick.high, low=tick.low, close=tick.close, timestamp=tick.timestamp, change=round(tick.ltp - tick.close, 2) if tick.close else 0.0, change_pct=tick.change_pct, broker="angelone", ) stream_engine.set_live_provider(_angel_live_provider) logger.info("[Stream] WebSocket stream bridged to real Angel One feed ✅") return True except Exception as e: logger.error(f"[Stream] Failed to bridge live feed to WebSocket stream: {e}") return False