""" ========================================= CV6 AI Trading OS Dashboard — Aggregation Service Broker-independent, dependency-injected. ========================================= """ import os import sys import time from datetime import datetime from typing import Dict, Any, List, Optional from sqlalchemy.orm import Session from loguru import logger from app.dashboard.dashboard_schemas import ( DashboardResponse, FundsInfo, PositionInfo, OrderInfo, PnLInfo, RiskInfo, AIStatusInfo, BrokerStatusInfo, MarketStatusInfo, SystemHealthInfo, ) # ── Paper engine fallback ──────────────────────────────────────────── # Import lazily to avoid circular imports at module load time. # The live singleton is _engine in paper_manager (used by the router). def _get_paper_engine(): try: from app.paper_trading.paper_manager import _engine return _engine except Exception: return None _START = time.time() class DashboardService: """ Aggregates data from registered brokers, AI engine, WebSocket manager, and trade history into a single snapshot. """ def __init__(self): self._brokers: Dict[str, Any] = {} self._ws_manager = None self._stream = None self._ai_engine = None self._history_svc = None # ─── Dependency Registration ──────────────────────────────────── def register_broker(self, name: str, adapter) -> None: self._brokers[name] = adapter def set_ws_manager(self, mgr) -> None: self._ws_manager = mgr def set_stream_engine(self, eng) -> None: self._stream = eng def set_ai_engine(self, eng) -> None: self._ai_engine = eng def set_history_service(self, svc) -> None: self._history_svc = svc # ─── SOURCE OF TRUTH gate ─────────────────────────────────────────────── # The dashboard must read the TradingStateManager (real, persisted state) # whenever it is active, and only fall back to the in-memory paper engine # in a genuine paper-only context. Previously register_broker() was never # called, so funds/positions/risk/PnL were ALWAYS served from the paper # engine — even during live REAL trading. This gate fixes that. def _tsm(self): from app.state.trading_state_manager import trading_state_manager return trading_state_manager def _tsm_active(self) -> bool: """ Use the TSM (REAL source of truth) when there is real activity, or when the autonomous engine is explicitly in REAL mode with a broker connected. In a genuine paper context (paper mode, no real positions) fall through to the paper engine so paper users see paper results. TSM is REAL-only, so real positions/deployed capital unambiguously mean "real activity present". """ try: snap = self._tsm().snapshot() if not snap.get("initialized"): return False # Any current OR historical real activity ⇒ TSM is authoritative. # realized PnL != 0 covers the case where a real session has closed # out and is now flat: the dashboard must keep showing TSM's real, # persisted realized PnL rather than flipping back to the paper # engine (which would misreport once positions go flat / the broker # connection blips). if (snap["position_count"] > 0 or snap["capital"]["deployed"] > 0 or abs(snap["pnl"]["realized"]) > 1e-9): return True from app.autonomous.autonomous_engine import autonomous_engine if getattr(autonomous_engine.config, "mode", "PAPER") == "REAL": from app.api.broker_api import manager as _bm return bool(getattr(_bm, "connected", False)) return False except Exception: return False # ─── Funds ────────────────────────────────────────────────────── def _funds(self) -> FundsInfo: for name, broker in self._brokers.items(): try: b = broker.get_balance() return FundsInfo( available_cash=float(b.get("available_cash") or 0), used_margin=float(b.get("used_margin") or 0), total_funds=float(b.get("net") or 0), broker=name, ) except Exception as e: logger.warning(f"[Dashboard] Funds {name}: {e}") # ── Fallback: paper engine when no broker connected ────────── pe = _get_paper_engine() if pe: try: acc = pe.account() return FundsInfo( available_cash=float(acc.get("available_capital", 0)), used_margin=float(acc.get("invested_amount", 0)), total_funds=float(acc.get("total_capital", 0)), collateral=0.0, broker="paper", ) except Exception as e: logger.warning(f"[Dashboard] Paper funds fallback: {e}") return FundsInfo() # ─── Positions ────────────────────────────────────────────────── def _positions(self) -> List[PositionInfo]: out = [] for name, broker in self._brokers.items(): try: for p in (broker.get_positions() or []): qty = int(p.get("quantity") or p.get("netqty") or 0) if qty == 0: continue avg = float(p.get("averageprice") or p.get("average_price") or 0) ltp = float(p.get("ltp") or p.get("lastprice") or 0) pnl = float(p.get("pnl") or p.get("unrealised") or 0) out.append(PositionInfo( symbol=str(p.get("tradingsymbol") or p.get("symbol", "")), exchange=str(p.get("exchange", "NSE")), quantity=qty, avg_price=avg, ltp=ltp, pnl=pnl, pnl_pct=round((ltp - avg) / avg * 100, 2) if avg else 0, product=str(p.get("product", "MIS")), trade_type="BUY" if qty > 0 else "SELL", broker=name, )) except Exception as e: logger.warning(f"[Dashboard] Positions {name}: {e}") # ── Fallback: paper engine open positions when broker offline ─ if not out and not self._brokers: pe = _get_paper_engine() if pe: try: acc = pe.account() for pos in acc.get("open_positions", []): qty = int(pos.get("quantity", 0)) avg = float(pos.get("entry_price", 0)) ltp = float(pos.get("ltp", avg)) pnl = float(pos.get("pnl", 0)) pnl_pct = float(pos.get("pnl_pct", 0)) out.append(PositionInfo( symbol=str(pos.get("symbol", "")), exchange="NSE", quantity=qty, avg_price=avg, ltp=ltp, pnl=pnl, pnl_pct=pnl_pct, product="MIS", trade_type=str(pos.get("side", "BUY")), broker="paper", )) except Exception as e: logger.warning(f"[Dashboard] Paper positions fallback: {e}") return out # ─── Orders ───────────────────────────────────────────────────── def _orders(self) -> List[OrderInfo]: out = [] # ── Primary: read from live BrokerManager singleton ──────────── try: from app.api.broker_api import manager as _bm if _bm.is_connected(): broker = _bm.get_broker() bname = _bm.get_broker_name() or "broker" if broker: raw = broker.orders() if isinstance(raw, list): order_list = raw elif isinstance(raw, dict): order_list = raw.get("data") or [] else: order_list = [] for o in order_list[-50:]: out.append(OrderInfo( order_id=str(o.get("orderid") or o.get("order_id", "")), symbol=str(o.get("tradingsymbol") or o.get("symbol", "")), trade_type=str(o.get("transactiontype") or o.get("side", "BUY")), quantity=int(o.get("quantity") or 0), price=float(o.get("price") or 0), status=str(o.get("status", "UNKNOWN")), order_type=str(o.get("ordertype") or o.get("order_type", "MARKET")), broker=bname, )) except Exception as e: logger.warning(f"[Dashboard] Orders from BrokerManager: {e}") # ── Legacy: explicitly registered adapters ────────────────────── for name, broker in self._brokers.items(): if any(o.broker == name for o in out): continue try: for o in (broker.get_orders() or [])[-50:]: out.append(OrderInfo( order_id=str(o.get("orderid") or o.get("order_id", "")), symbol=str(o.get("tradingsymbol") or o.get("symbol", "")), trade_type=str(o.get("transactiontype") or o.get("side", "BUY")), quantity=int(o.get("quantity") or 0), price=float(o.get("price") or 0), status=str(o.get("status", "UNKNOWN")), order_type=str(o.get("ordertype") or o.get("order_type", "MARKET")), broker=name, )) except Exception as e: logger.warning(f"[Dashboard] Orders {name}: {e}") return out # ─── PnL ──────────────────────────────────────────────────────── def _pnl(self, positions: List[PositionInfo], db: Optional[Session]) -> PnLInfo: unrealized = sum(p.pnl for p in positions) realized = charges = 0.0 wins = losses = 0 # Try broker history (SQLite trade_history table) if self._history_svc and db: try: s = self._history_svc.pnl_summary(db) realized = s.total_pnl charges = s.total_charges wins = s.winning_trades losses = s.losing_trades except Exception as e: logger.warning(f"[Dashboard] PnL: {e}") # ── Fallback: paper engine when no broker data ─────────────── # If realized is still 0 and no broker registered (paper mode), # read PnL directly from the in-memory paper engine. if realized == 0.0 and not self._brokers: pe = _get_paper_engine() if pe: try: acc = pe.account() realized = float(acc.get("realized_pnl", 0)) wins = int(acc.get("win_count", 0)) losses = int(acc.get("loss_count", 0)) unrealized = float(acc.get("unrealized_pnl", 0)) except Exception as e: logger.warning(f"[Dashboard] Paper PnL fallback: {e}") total = realized + unrealized n = wins + losses return PnLInfo( realized_pnl=round(realized, 2), unrealized_pnl=round(unrealized, 2), total_pnl=round(total, 2), total_charges=round(charges, 2), net_pnl=round(total - charges, 2), winning_trades=wins, losing_trades=losses, win_rate_pct=round(wins / n * 100, 1) if n else 0.0, ) # ─── Risk ─────────────────────────────────────────────────────── def _risk(self, funds: FundsInfo, positions: List[PositionInfo]) -> RiskInfo: exposure = sum(abs(p.quantity * p.ltp) for p in positions) cap = funds.total_funds or 1 margin_pct = round(funds.used_margin / cap * 100, 1) open_risk = sum(abs(p.pnl) for p in positions if p.pnl < 0) status = "OK" if margin_pct > 80: status = "WARNING" if margin_pct > 95: status = "BREACH" return RiskInfo( current_exposure=round(exposure, 2), margin_utilization_pct=margin_pct, open_risk=round(open_risk, 2), daily_loss_limit=float(os.getenv("DAILY_LOSS_LIMIT", "5000")), daily_loss_used=round(open_risk, 2), risk_status=status, ) # ─── AI Status ────────────────────────────────────────────────── def _ai_status(self) -> AIStatusInfo: if not self._ai_engine: return AIStatusInfo(enabled=False) try: models = [a.name for a in getattr(self._ai_engine, "adapters", [])] return AIStatusInfo( enabled=True, active_models=models, consensus_available=bool(models), ) except Exception: return AIStatusInfo(enabled=False) # ─── Broker Status ────────────────────────────────────────────── def _broker_status(self) -> List[BrokerStatusInfo]: out = [] # ── Primary: read from the live BrokerManager singleton ──────── try: from app.api.broker_api import manager as _bm if _bm.is_connected(): t = time.time() name = _bm.get_broker_name() or "broker" try: _bm.funds() # lightweight ping out.append(BrokerStatusInfo( name=name, connected=True, latency_ms=round((time.time() - t) * 1000, 1), last_ping=time.time(), )) except Exception: # connected but funds call failed — still show connected out.append(BrokerStatusInfo( name=name, connected=True, latency_ms=round((time.time() - t) * 1000, 1), last_ping=time.time(), )) except Exception: pass # ── Legacy: explicitly registered adapters (keep for compat) ─── for name, broker in self._brokers.items(): # skip if already reported from BrokerManager if any(b.name == name for b in out): continue t = time.time() try: fn = getattr(broker, "funds", None) or getattr(broker, "get_balance", None) if fn: fn() out.append(BrokerStatusInfo( name=name, connected=True, latency_ms=round((time.time() - t) * 1000, 1), last_ping=time.time(), )) except Exception as e: out.append(BrokerStatusInfo(name=name, connected=False, error=str(e))) if not out: out.append(BrokerStatusInfo(name="none", connected=False, error="No broker connected")) return out # ─── Market Status ────────────────────────────────────────────── def _market_status(self) -> MarketStatusInfo: now = datetime.now() t = now.hour * 60 + now.minute if t < 540: session = "CLOSED" elif 540 <= t < 555: session = "PRE_OPEN" elif 555 <= t <= 930: session = "OPEN" else: session = "POST_CLOSE" return MarketStatusInfo( is_open=(session == "OPEN"), session=session, exchange="NSE", ) # ─── System Health ────────────────────────────────────────────── def _health(self) -> SystemHealthInfo: cpu = mem = disk = 0.0 status = "HEALTHY" try: import psutil cpu = psutil.cpu_percent(interval=0.1) mem = psutil.virtual_memory().percent import sys, os _disk_path = "C:\\" if sys.platform == "win32" else "/" disk = psutil.disk_usage(_disk_path).percent if cpu > 85 or mem > 90: status = "DEGRADED" if cpu > 95 or mem > 95: status = "CRITICAL" except ImportError: pass ws_clients = active_streams = 0 if self._ws_manager: ws_clients = len(getattr(self._ws_manager, "active", {})) if self._stream: active_streams = len(self._stream.streaming()) return SystemHealthInfo( cpu_pct=cpu, memory_pct=mem, disk_pct=disk, uptime_seconds=round(time.time() - _START, 0), websocket_clients=ws_clients, active_streams=active_streams, db_ok=True, status=status, ) # ─── Full Snapshot ─────────────────────────────────────────────── def snapshot(self, db: Optional[Session] = None) -> DashboardResponse: # ── SOURCE OF TRUTH: read funds/positions/PnL/risk from the # TradingStateManager when it's active; only use the paper-engine # fallback in a genuine paper-only context. Orders/broker status keep # their existing (already-correct) direct BrokerManager read paths. if self._tsm_active(): funds, positions, pnl, risk = self._from_tsm(db) else: funds = self._funds() positions = self._positions() pnl = self._pnl(positions, db) risk = self._risk(funds, positions) orders = self._orders() return DashboardResponse( portfolio=positions, funds=funds, pnl=pnl, risk=risk, open_positions=[p for p in positions if p.quantity != 0], orders=orders, ai_status=self._ai_status(), broker_status=self._broker_status(), market_status=self._market_status(), system_health=self._health(), ) # ─── Build the money views from the single source of truth ────────────── def _from_tsm(self, db: Optional[Session]): snap = self._tsm().snapshot() cap = snap["capital"] pnl_s = snap["pnl"] funds = FundsInfo( available_cash=float(cap["available"]), used_margin=float(cap["deployed"]), total_funds=float(cap["total"]), collateral=0.0, broker="live", ) positions: List[PositionInfo] = [] for p in snap["positions"]: positions.append(PositionInfo( symbol=p["symbol"], exchange=p["exchange"], quantity=p["quantity"], avg_price=p["avg_price"], ltp=p["ltp"], pnl=p["pnl"], pnl_pct=p["pnl_pct"], product="MIS" if p["style"] in ("INTRADAY", "SCALPING") else "NRML", trade_type=p["side"], broker=p["broker"] or "live", )) # Win/loss come from the journal (trade_history) which TSM now writes. wins = losses = 0 charges = 0.0 if self._history_svc and db is not None: try: s = self._history_svc.pnl_summary(db) wins = s.winning_trades losses = s.losing_trades charges = s.total_charges except Exception as e: logger.warning(f"[Dashboard] journal summary: {e}") n = wins + losses pnl = PnLInfo( realized_pnl=round(pnl_s["realized"], 2), unrealized_pnl=round(pnl_s["unrealized"], 2), total_pnl=round(pnl_s["total"], 2), total_charges=round(charges, 2), net_pnl=round(pnl_s["total"] - charges, 2), winning_trades=wins, losing_trades=losses, win_rate_pct=round(wins / n * 100, 1) if n else 0.0, ) # Risk derived from the SAME authoritative positions/capital, plus the # real risk_guard status (not a separate ad hoc calc on paper data). exposure = sum(abs(p.quantity * p.ltp) for p in positions) total_cap = cap["total"] or 1 margin_pct = round(cap["deployed"] / total_cap * 100, 1) open_risk = sum(abs(p.pnl) for p in positions if p.pnl < 0) status = "OK" daily_limit = 0.0 daily_used = round(open_risk, 2) try: from app.risk_guard.risk_guard import risk_guard rg = risk_guard.get_status() if isinstance(rg, dict): if rg.get("kill_switch"): status = "BREACH" daily_limit = float((rg.get("limits") or {}).get("daily", 0) or 0) today = float((rg.get("pnl") or {}).get("today_pnl", 0) or 0) if today < 0: daily_used = round(abs(today), 2) except Exception: pass if status == "OK": if margin_pct > 80: status = "WARNING" if margin_pct > 95: status = "BREACH" risk = RiskInfo( current_exposure=round(exposure, 2), margin_utilization_pct=margin_pct, open_risk=round(open_risk, 2), daily_loss_limit=daily_limit, daily_loss_used=daily_used, risk_status=status, ) return funds, positions, pnl, risk # Singleton dashboard_service = DashboardService()