From be5a3f63b1076a398ea9156f0b7221b04547ef61 Mon Sep 17 00:00:00 2001 From: Umais Khan Date: Sat, 4 Apr 2026 17:10:33 +0500 Subject: [PATCH] Create state.py --- core/state.py | 1110 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1110 insertions(+) create mode 100644 core/state.py diff --git a/core/state.py b/core/state.py new file mode 100644 index 0000000..c2f8314 --- /dev/null +++ b/core/state.py @@ -0,0 +1,1110 @@ +import asyncio +import os +from typing import Any, Dict, List, Optional + + +class StateManager: + def __init__(self) -> None: + self._lock = asyncio.Lock() + self._latest_tick: Optional[Dict[str, Any]] = None + self._latest_tick_by_symbol: Dict[str, Dict[str, Any]] = {} + self._latest_ohlc_1h: Optional[Dict[str, Any]] = None + self._latest_ohlc_5m: Optional[Dict[str, Any]] = None + self._connection_status: str = "disconnected" + self._tick_listeners: List[asyncio.Queue] = [] + self._status_listeners: List[asyncio.Queue] = [] + self._account_balance: Optional[float] = None + self._account_equity: Optional[float] = None + self._margin: Optional[float] = None + self._free_margin: Optional[float] = None + self._margin_level: Optional[float] = None + self._leverage: Optional[int] = None + self._currency: Optional[str] = None + self._account_listeners: List[asyncio.Queue] = [] + self._account_info_by_symbol: Dict[str, Dict[str, Any]] = {} + self._candle_listeners: List[asyncio.Queue] = [] + self._latest_candles_m1: List[Dict[str, Any]] = [] + self._latest_candles_m5: List[Dict[str, Any]] = [] + self._latest_candles_m15: List[Dict[str, Any]] = [] + self._latest_candles_m30: List[Dict[str, Any]] = [] + self._latest_candles_h1: List[Dict[str, Any]] = [] + self._candles_by_symbol: Dict[str, Dict[str, List[Dict[str, Any]]]] = {} + self._feature_listeners: List[asyncio.Queue] = [] + self._latest_features_m1: List[Dict[str, Any]] = [] + self._latest_features_m5: List[Dict[str, Any]] = [] + self._latest_features_m15: List[Dict[str, Any]] = [] + self._latest_features_m30: List[Dict[str, Any]] = [] + self._latest_features_h1: List[Dict[str, Any]] = [] + self._features_by_symbol: Dict[str, Dict[str, List[Dict[str, Any]]]] = {} + self._regime_listeners: List[asyncio.Queue] = [] + self._latest_market_regime: Optional[Dict[str, Any]] = None + self._latest_regimes: List[Dict[str, Any]] = [] + self._latest_signal_by_symbol: Dict[str, Dict[str, Any]] = {} + self._signal_listeners: List[asyncio.Queue] = [] + self._latest_signal: Optional[Dict[str, Any]] = None + self._latest_decision_by_symbol: Dict[str, Dict[str, Any]] = {} + self._decision_listeners: List[asyncio.Queue] = [] + self._latest_decision: Optional[Dict[str, Any]] = None + self._latest_risk_by_symbol: Dict[str, Dict[str, Any]] = {} + self._risk_listeners: List[asyncio.Queue] = [] + self._latest_risk: Optional[Dict[str, Any]] = None + self._config_listeners: List[asyncio.Queue] = [] + self._trade_listeners: List[asyncio.Queue] = [] + self._latest_trade: Optional[Dict[str, Any]] = None + self._latest_trade_by_symbol: Dict[str, Dict[str, Any]] = {} + self._pnl_listeners: List[asyncio.Queue] = [] + self._latest_pnl_report: Optional[Dict[str, Any]] = None + self._latest_pnl_by_symbol: Dict[str, Dict[str, Any]] = {} + self._trade_close_listeners: List[asyncio.Queue] = [] + self._latest_trade_close: Optional[Dict[str, Any]] = None + self._latest_trade_close_by_symbol: Dict[str, Dict[str, Any]] = {} + self._decision_threshold: float = 0.6 + self._risk_per_trade: float = 0.01 + self._atr_mult: float = 1.5 + self._rr: float = 2.0 + self._dd_limit: float = 0.1 + self._trailing_mult: float = 0.0 + self._latest_market_regime_by_symbol: Dict[str, Dict[str, Any]] = {} + self._broker_params_by_symbol: Dict[str, Dict[str, Any]] = {} + self._retail_signal_listeners: List[asyncio.Queue] = [] + self._latest_retail_signal_by_symbol: Dict[str, Dict[str, Any]] = {} + self._retail_enabled: float = 0.0 + self._retail_rr: float = 1.5 + self._retail_min_conf: float = 0.55 + self._retail_merge_mode: str = "augment" + self._signal_debug_listeners: List[asyncio.Queue] = [] + self._latest_signal_debug_by_symbol: Dict[str, Dict[str, Any]] = {} + self._pipeline_debug_listeners: List[asyncio.Queue] = [] + self._latest_pipeline_debug_by_symbol: Dict[str, Dict[str, Any]] = {} + self._hft_optimizer_enabled: float = 0.0 + self._ga_evaluator_enabled: float = 0.0 + self._correlation_insights_listeners: List[asyncio.Queue] = [] + self._latest_correlation_insights: Optional[Dict[str, Any]] = None + self._latest_correlation_insights_by_symbol: Dict[str, Dict[str, Any]] = {} + self._data_quality_listeners: List[asyncio.Queue] = [] + self._latest_data_quality: Optional[Dict[str, Any]] = None + self._latest_data_quality_by_symbol: Dict[str, Dict[str, Any]] = {} + self._risk_features_by_symbol: Dict[str, Dict[str, Any]] = {} + self._rate_limit_listeners: List[asyncio.Queue] = [] + self._latest_rate_limits: Optional[Dict[str, Any]] = None + self._monte_carlo_insights_listeners: List[asyncio.Queue] = [] + self._latest_monte_carlo_insights: Optional[Dict[str, Any]] = None + self._latest_monte_carlo_insights_by_symbol: Dict[str, Dict[str, Any]] = {} + self._var_insights_listeners: List[asyncio.Queue] = [] + self._latest_var_insights: Optional[Dict[str, Any]] = None + self._latest_var_insights_by_symbol: Dict[str, Dict[str, Any]] = {} + + async def set_connection_status(self, status: str) -> None: + async with self._lock: + self._connection_status = status + for q in list(self._status_listeners): + if not q.full(): + await q.put({"type": "status", "status": status}) + + async def get_connection_status(self) -> str: + async with self._lock: + return self._connection_status + + async def set_latest_tick(self, tick: Dict[str, Any]) -> None: + async with self._lock: + self._latest_tick = tick + for q in list(self._tick_listeners): + if not q.full(): + await q.put({"type": "tick", "tick": tick, "symbol": str(tick.get("symbol") or "")}) + + async def get_latest_tick(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return self._latest_tick + async def set_latest_tick_for(self, symbol: str, tick: Dict[str, Any]) -> None: + async with self._lock: + self._latest_tick_by_symbol[symbol] = dict(tick) + for q in list(self._tick_listeners): + if not q.full(): + await q.put({"type": "tick", "tick": dict(tick), "symbol": symbol}) + async def get_latest_tick_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_tick_by_symbol.get(symbol) + return dict(v) if v else None + + async def set_latest_ohlc_1h(self, ohlc: Dict[str, Any]) -> None: + async with self._lock: + self._latest_ohlc_1h = ohlc + + async def get_latest_ohlc_1h(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return self._latest_ohlc_1h + + async def set_latest_ohlc_5m(self, ohlc: Dict[str, Any]) -> None: + async with self._lock: + self._latest_ohlc_5m = ohlc + + async def get_latest_ohlc_5m(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return self._latest_ohlc_5m + + async def add_tick_listener(self, maxsize: int = 1000) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._tick_listeners.append(q) + if self._latest_tick is not None: + await q.put({"type": "tick", "tick": self._latest_tick}) + return q + + async def remove_tick_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._tick_listeners: + self._tick_listeners.remove(q) + + async def add_status_listener(self, maxsize: int = 100) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._status_listeners.append(q) + await q.put({"type": "status", "status": self._connection_status}) + return q + + async def remove_status_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._status_listeners: + self._status_listeners.remove(q) + + async def set_account_info(self, info: Dict[str, Any]) -> None: + async with self._lock: + self._account_balance = info.get("balance") + self._account_equity = info.get("equity") + self._margin = info.get("margin") + self._free_margin = info.get("free_margin") + self._margin_level = info.get("margin_level") + self._leverage = info.get("leverage") + self._currency = info.get("currency") + payload = { + "type": "account_info", + "balance": self._account_balance, + "equity": self._account_equity, + "margin": self._margin, + "free_margin": self._free_margin, + "margin_level": self._margin_level, + "leverage": self._leverage, + "currency": self._currency, + } + for q in list(self._account_listeners): + if not q.full(): + await q.put(payload) + async def set_account_info_for(self, symbol: str, info: Dict[str, Any]) -> None: + async with self._lock: + self._account_info_by_symbol[symbol] = dict(info) + payload = { + "type": "account_info", + "symbol": symbol, + "balance": info.get("balance"), + "equity": info.get("equity"), + "margin": info.get("margin"), + "free_margin": info.get("free_margin"), + "margin_level": info.get("margin_level"), + "leverage": info.get("leverage"), + "currency": info.get("currency"), + } + for q in list(self._account_listeners): + if not q.full(): + await q.put(payload) + + async def get_account_info(self) -> Optional[Dict[str, Any]]: + async with self._lock: + if self._account_balance is None and self._account_equity is None: + return None + return { + "balance": self._account_balance, + "equity": self._account_equity, + "margin": self._margin, + "free_margin": self._free_margin, + "margin_level": self._margin_level, + "leverage": self._leverage, + "currency": self._currency, + } + + async def add_account_listener(self, maxsize: int = 100) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._account_listeners.append(q) + if self._account_balance is not None or self._account_equity is not None: + await q.put({ + "type": "account_info", + "balance": self._account_balance, + "equity": self._account_equity, + "margin": self._margin, + "free_margin": self._free_margin, + "margin_level": self._margin_level, + "leverage": self._leverage, + "currency": self._currency, + }) + return q + + async def remove_account_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._account_listeners: + self._account_listeners.remove(q) + + async def add_candle_listener(self, maxsize: int = 500) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._candle_listeners.append(q) + return q + + async def remove_candle_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._candle_listeners: + self._candle_listeners.remove(q) + + async def push_candle_update(self, timeframe: str, candle: Dict[str, Any]) -> None: + async with self._lock: + if timeframe == "M1": + self._latest_candles_m1.append(candle) + self._latest_candles_m1 = self._latest_candles_m1[-500:] + elif timeframe == "M5": + self._latest_candles_m5.append(candle) + self._latest_candles_m5 = self._latest_candles_m5[-500:] + elif timeframe == "M15": + self._latest_candles_m15.append(candle) + self._latest_candles_m15 = self._latest_candles_m15[-500:] + elif timeframe == "M30": + self._latest_candles_m30.append(candle) + self._latest_candles_m30 = self._latest_candles_m30[-500:] + elif timeframe == "H1": + self._latest_candles_h1.append(candle) + self._latest_candles_h1 = self._latest_candles_h1[-500:] + msg = { + "type": "candle_update", + "timeframe": timeframe, + "open": candle.get("open"), + "high": candle.get("high"), + "low": candle.get("low"), + "close": candle.get("close"), + "time": candle.get("t"), + } + for q in list(self._candle_listeners): + if not q.full(): + await q.put(msg) + async def push_candle_update_for(self, symbol: str, timeframe: str, candle: Dict[str, Any]) -> None: + async with self._lock: + if symbol not in self._candles_by_symbol: + self._candles_by_symbol[symbol] = {"M1": [], "M5": [], "M15": [], "M30": [], "H1": []} + self._candles_by_symbol[symbol][timeframe].append(candle) + self._candles_by_symbol[symbol][timeframe] = self._candles_by_symbol[symbol][timeframe][-500:] + msg = { + "type": "candle_update", + "symbol": symbol, + "timeframe": timeframe, + "open": candle.get("open"), + "high": candle.get("high"), + "low": candle.get("low"), + "close": candle.get("close"), + "time": candle.get("t"), + } + for q in list(self._candle_listeners): + if not q.full(): + await q.put(msg) + + async def get_latest_candles_m1(self) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._latest_candles_m1) + + async def get_latest_candles_m5(self) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._latest_candles_m5) + + async def get_latest_candles_m15(self) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._latest_candles_m15) + + async def get_latest_candles_m30(self) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._latest_candles_m30) + + async def get_latest_candles_h1(self) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._latest_candles_h1) + async def get_latest_candles_m1_for(self, symbol: str) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._candles_by_symbol.get(symbol, {}).get("M1", [])) + async def get_latest_candles_m5_for(self, symbol: str) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._candles_by_symbol.get(symbol, {}).get("M5", [])) + async def get_latest_candles_m15_for(self, symbol: str) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._candles_by_symbol.get(symbol, {}).get("M15", [])) + async def get_latest_candles_m30_for(self, symbol: str) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._candles_by_symbol.get(symbol, {}).get("M30", [])) + async def get_latest_candles_h1_for(self, symbol: str) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._candles_by_symbol.get(symbol, {}).get("H1", [])) + + async def set_latest_features(self, timeframe: str, features: Dict[str, Any]) -> None: + async with self._lock: + if timeframe == "M1": + self._latest_features_m1.append(features) + self._latest_features_m1 = self._latest_features_m1[-500:] + elif timeframe == "M5": + self._latest_features_m5.append(features) + self._latest_features_m5 = self._latest_features_m5[-500:] + elif timeframe == "M15": + self._latest_features_m15.append(features) + self._latest_features_m15 = self._latest_features_m15[-500:] + elif timeframe == "M30": + self._latest_features_m30.append(features) + self._latest_features_m30 = self._latest_features_m30[-500:] + elif timeframe == "H1": + self._latest_features_h1.append(features) + self._latest_features_h1 = self._latest_features_h1[-500:] + msg = { + "type": "features_update", + "timeframe": timeframe, + "features": features, + } + for q in list(self._feature_listeners): + if not q.full(): + await q.put(msg) + async def set_latest_features_for(self, symbol: str, timeframe: str, features: Dict[str, Any]) -> None: + async with self._lock: + if symbol not in self._features_by_symbol: + self._features_by_symbol[symbol] = {"M1": [], "M5": [], "M15": [], "M30": [], "H1": []} + self._features_by_symbol[symbol][timeframe].append(features) + self._features_by_symbol[symbol][timeframe] = self._features_by_symbol[symbol][timeframe][-500:] + msg = { + "type": "features_update", + "symbol": symbol, + "timeframe": timeframe, + "features": features, + } + for q in list(self._feature_listeners): + if not q.full(): + await q.put(msg) + + async def add_feature_listener(self, maxsize: int = 200) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._feature_listeners.append(q) + return q + + async def remove_feature_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._feature_listeners: + self._feature_listeners.remove(q) + + async def get_latest_features_m1(self) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._latest_features_m1) + + async def get_latest_features_m5(self) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._latest_features_m5) + + async def get_latest_features_m15(self) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._latest_features_m15) + + async def get_latest_features_m30(self) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._latest_features_m30) + + async def get_latest_features_h1(self) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._latest_features_h1) + async def get_latest_features_m1_for(self, symbol: str) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._features_by_symbol.get(symbol, {}).get("M1", [])) + async def get_latest_features_m5_for(self, symbol: str) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._features_by_symbol.get(symbol, {}).get("M5", [])) + async def get_latest_features_m15_for(self, symbol: str) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._features_by_symbol.get(symbol, {}).get("M15", [])) + async def get_latest_features_m30_for(self, symbol: str) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._features_by_symbol.get(symbol, {}).get("M30", [])) + async def get_latest_features_h1_for(self, symbol: str) -> List[Dict[str, Any]]: + async with self._lock: + return list(self._features_by_symbol.get(symbol, {}).get("H1", [])) + + async def set_market_regime(self, regime: str, confidence: float, t: int, **kwargs: Any) -> None: + async with self._lock: + payload = {"regime": regime, "confidence": float(confidence), "time": int(t)} + # Merge optional extended fields + for k, v in (kwargs or {}).items(): + payload[k] = v + self._latest_market_regime = dict(payload) + self._latest_regimes.append(dict(payload)) + self._latest_regimes = self._latest_regimes[-500:] + msg = {"type": "regime_update", **payload} + for q in list(self._regime_listeners): + if not q.full(): + await q.put(msg) + + async def get_latest_market_regime(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_market_regime) if self._latest_market_regime else None + + async def get_latest_regime(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_market_regime) if self._latest_market_regime else None + + async def set_latest_regime(self, payload: Dict[str, Any]) -> None: + """ + Alias to set_market_regime for compatibility with expanded regime payloads. + """ + async with self._lock: + self._latest_market_regime = dict(payload) + self._latest_regimes.append(dict(payload)) + self._latest_regimes = self._latest_regimes[-500:] + msg = {"type": "regime_update", **payload} + for q in list(self._regime_listeners): + if not q.full(): + await q.put(msg) + + async def add_regime_listener(self, maxsize: int = 200) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._regime_listeners.append(q) + if self._latest_market_regime is not None: + await q.put({"type": "regime_update", **self._latest_market_regime}) + return q + + async def remove_regime_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._regime_listeners: + self._regime_listeners.remove(q) + + async def set_latest_signal(self, signal: Dict[str, Any]) -> None: + async with self._lock: + self._latest_signal = signal + msg = {"type": "signal_update", **signal, "time": signal.get("timestamp")} + for q in list(self._signal_listeners): + if not q.full(): + await q.put(msg) + async def set_latest_signal_for(self, symbol: str, signal: Dict[str, Any]) -> None: + async with self._lock: + self._latest_signal_by_symbol[symbol] = dict(signal) + s = dict(signal) + s["symbol"] = symbol + msg = {"type": "signal_update", **s, "time": s.get("timestamp")} + for q in list(self._signal_listeners): + if not q.full(): + await q.put(msg) + + async def get_latest_signal(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_signal) if self._latest_signal else None + async def get_latest_signal_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_signal_by_symbol.get(symbol) + return dict(v) if v else None + + async def add_signal_listener(self, maxsize: int = 200) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._signal_listeners.append(q) + if self._latest_signal is not None: + await q.put({"type": "signal_update", **self._latest_signal, "time": self._latest_signal.get("timestamp")}) + return q + + async def remove_signal_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._signal_listeners: + self._signal_listeners.remove(q) + + async def set_latest_decision(self, decision: Dict[str, Any]) -> None: + async with self._lock: + self._latest_decision = decision + msg = {"type": "decision_update", "decision": decision.get("signal"), "approved": decision.get("approved"), "confidence": decision.get("confidence"), "time": decision.get("time")} + for q in list(self._decision_listeners): + if not q.full(): + await q.put(msg) + async def set_latest_decision_for(self, symbol: str, decision: Dict[str, Any]) -> None: + async with self._lock: + self._latest_decision_by_symbol[symbol] = dict(decision) + msg = {"type": "decision_update", "symbol": symbol, "decision": decision.get("signal"), "approved": decision.get("approved"), "confidence": decision.get("confidence"), "time": decision.get("time")} + for q in list(self._decision_listeners): + if not q.full(): + await q.put(msg) + + async def get_latest_decision(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_decision) if self._latest_decision else None + async def get_latest_decision_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_decision_by_symbol.get(symbol) + return dict(v) if v else None + + async def add_decision_listener(self, maxsize: int = 200) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._decision_listeners.append(q) + if self._latest_decision is not None: + await q.put({"type": "decision_update", "decision": self._latest_decision.get("signal"), "approved": self._latest_decision.get("approved"), "confidence": self._latest_decision.get("confidence"), "time": self._latest_decision.get("time")}) + return q + + async def remove_decision_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._decision_listeners: + self._decision_listeners.remove(q) + + async def set_latest_risk(self, risk: Dict[str, Any]) -> None: + async with self._lock: + self._latest_risk = risk + msg = {"type": "risk_update", "risk": risk, "time": risk.get("time")} + for q in list(self._risk_listeners): + if not q.full(): + await q.put(msg) + async def set_latest_risk_for(self, symbol: str, risk: Dict[str, Any]) -> None: + async with self._lock: + self._latest_risk_by_symbol[symbol] = dict(risk) + msg = {"type": "risk_update", "symbol": symbol, "risk": dict(risk), "time": risk.get("time")} + for q in list(self._risk_listeners): + if not q.full(): + await q.put(msg) + + async def get_latest_risk(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_risk) if self._latest_risk else None + async def get_latest_risk_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_risk_by_symbol.get(symbol) + return dict(v) if v else None + + async def add_risk_listener(self, maxsize: int = 200) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._risk_listeners.append(q) + if self._latest_risk is not None: + await q.put({"type": "risk_update", "risk": self._latest_risk, "time": self._latest_risk.get("time")}) + return q + + async def remove_risk_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._risk_listeners: + self._risk_listeners.remove(q) + + async def set_latest_trade(self, trade: Dict[str, Any]) -> None: + async with self._lock: + self._latest_trade = dict(trade) + msg = {"type": "trade_update", "trade": self._latest_trade} + for q in list(self._trade_listeners): + if not q.full(): + await q.put(msg) + async def set_latest_trade_for(self, symbol: str, trade: Dict[str, Any]) -> None: + async with self._lock: + self._latest_trade_by_symbol[symbol] = dict(trade) + msg = {"type": "trade_update", "symbol": symbol, "trade": dict(trade)} + for q in list(self._trade_listeners): + if not q.full(): + await q.put(msg) + + async def get_latest_trade(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_trade) if self._latest_trade else None + async def get_latest_trade_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_trade_by_symbol.get(symbol) + return dict(v) if v else None + + async def add_trade_listener(self, maxsize: int = 200) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._trade_listeners.append(q) + if self._latest_trade is not None: + await q.put({"type": "trade_update", "trade": self._latest_trade}) + return q + + async def remove_trade_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._trade_listeners: + self._trade_listeners.remove(q) + + async def set_latest_trade_close(self, trade: Dict[str, Any]) -> None: + async with self._lock: + self._latest_trade_close = dict(trade) + msg = {"type": "trade_close", "trade": self._latest_trade_close} + for q in list(self._trade_close_listeners): + if not q.full(): + await q.put(msg) + async def set_latest_trade_close_for(self, symbol: str, trade: Dict[str, Any]) -> None: + async with self._lock: + self._latest_trade_close_by_symbol[symbol] = dict(trade) + msg = {"type": "trade_close", "symbol": symbol, "trade": dict(trade)} + for q in list(self._trade_close_listeners): + if not q.full(): + await q.put(msg) + + async def get_latest_trade_close(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_trade_close) if self._latest_trade_close else None + async def get_latest_trade_close_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_trade_close_by_symbol.get(symbol) + return dict(v) if v else None + + async def add_trade_close_listener(self, maxsize: int = 200) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._trade_close_listeners.append(q) + if self._latest_trade_close is not None: + await q.put({"type": "trade_close", "trade": self._latest_trade_close}) + return q + + async def remove_trade_close_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._trade_close_listeners: + self._trade_close_listeners.remove(q) + + async def set_latest_pnl_report(self, report: Dict[str, Any]) -> None: + async with self._lock: + self._latest_pnl_report = dict(report) + msg = {"type": "pnl_report", "report": self._latest_pnl_report} + for q in list(self._pnl_listeners): + if not q.full(): + await q.put(msg) + async def set_latest_pnl_report_for(self, symbol: str, report: Dict[str, Any]) -> None: + async with self._lock: + self._latest_pnl_by_symbol[symbol] = dict(report) + msg = {"type": "pnl_report", "symbol": symbol, "report": dict(report)} + for q in list(self._pnl_listeners): + if not q.full(): + await q.put(msg) + + async def get_latest_pnl_report(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_pnl_report) if self._latest_pnl_report else None + async def get_latest_pnl_report_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_pnl_by_symbol.get(symbol) + return dict(v) if v else None + + async def add_pnl_listener(self, maxsize: int = 100) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._pnl_listeners.append(q) + if self._latest_pnl_report is not None: + await q.put({"type": "pnl_report", "report": self._latest_pnl_report}) + return q + + async def remove_pnl_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._pnl_listeners: + self._pnl_listeners.remove(q) + async def add_config_listener(self, maxsize: int = 100) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._config_listeners.append(q) + await q.put({"type": "config_update", "parameter": "threshold", "value": self._decision_threshold}) + await q.put({"type": "config_update", "parameter": "risk_per_trade", "value": self._risk_per_trade}) + await q.put({"type": "config_update", "parameter": "atr_multiplier", "value": self._atr_mult}) + await q.put({"type": "config_update", "parameter": "rr", "value": self._rr}) + await q.put({"type": "config_update", "parameter": "drawdown_limit", "value": self._dd_limit}) + await q.put({"type": "config_update", "parameter": "trailing_multiplier", "value": self._trailing_mult}) + await q.put({"type": "config_update", "parameter": "retail_strategies_enabled", "value": self._retail_enabled}) + await q.put({"type": "config_update", "parameter": "retail_rr", "value": self._retail_rr}) + await q.put({"type": "config_update", "parameter": "retail_min_conf", "value": self._retail_min_conf}) + await q.put({"type": "config_update", "parameter": "retail_merge_mode", "value": self._retail_merge_mode}) + await q.put({"type": "config_update", "parameter": "hft_optimizer_enabled", "value": self._hft_optimizer_enabled}) + await q.put({"type": "config_update", "parameter": "ga_evaluator_enabled", "value": self._ga_evaluator_enabled}) + return q + + async def remove_config_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._config_listeners: + self._config_listeners.remove(q) + + async def set_config_param(self, parameter: str, value: float) -> None: + async with self._lock: + p = parameter.lower() + if p == "threshold": + self._decision_threshold = float(value) + elif p == "risk_per_trade": + self._risk_per_trade = float(value) + elif p in ("atr_multiplier", "atr_mult"): + self._atr_mult = float(value) + elif p in ("reward_risk", "rr"): + self._rr = float(value) + elif p in ("drawdown_limit", "dd_limit"): + self._dd_limit = float(value) + elif p in ("trailing_multiplier", "trail_mult"): + self._trailing_mult = float(value) + elif p in ("retail_strategies_enabled", "retail_enabled"): + self._retail_enabled = float(value) + elif p in ("retail_rr",): + self._retail_rr = float(value) + elif p in ("retail_min_conf",): + self._retail_min_conf = float(value) + elif p in ("retail_merge_mode",): + try: + self._retail_merge_mode = str(value) + except Exception: + self._retail_merge_mode = "augment" + elif p in ("hft_optimizer_enabled",): + self._hft_optimizer_enabled = float(value) + elif p in ("ga_evaluator_enabled",): + self._ga_evaluator_enabled = float(value) + msg = {"type": "config_update", "parameter": p if p != "atr_mult" else "atr_multiplier", "value": float(value) if p != "retail_merge_mode" else self._retail_merge_mode} + for q in list(self._config_listeners): + if not q.full(): + await q.put(msg) + + async def get_decision_threshold(self) -> float: + async with self._lock: + return self._decision_threshold + + async def get_risk_params(self) -> Dict[str, float]: + async with self._lock: + return { + "risk_per_trade": self._risk_per_trade, + "atr_multiplier": self._atr_mult, + "rr": self._rr, + "drawdown_limit": self._dd_limit, + "trailing_multiplier": self._trailing_mult, + } + + async def get_hft_optimizer_enabled(self) -> bool: + async with self._lock: + return bool(self._hft_optimizer_enabled > 0.5) + + async def get_ga_evaluator_enabled(self) -> bool: + async with self._lock: + return bool(self._ga_evaluator_enabled > 0.5) + + async def set_latest_regime_for(self, symbol: str, payload: Dict[str, Any]) -> None: + async with self._lock: + p = dict(payload) + p["symbol"] = symbol + self._latest_market_regime_by_symbol[symbol] = p + msg = {"type": "regime_update", **p} + for q in list(self._regime_listeners): + if not q.full(): + await q.put(msg) + + async def get_latest_market_regime_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_market_regime_by_symbol.get(symbol) + return dict(v) if v else None + + async def set_broker_params(self, symbol: str, params: Dict[str, Any]) -> None: + async with self._lock: + self._broker_params_by_symbol[symbol] = dict(params) + + async def get_broker_params(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._broker_params_by_symbol.get(symbol) + return dict(v) if v else None + + async def set_latest_retail_signal_for(self, symbol: str, signal: Dict[str, Any]) -> None: + async with self._lock: + self._latest_retail_signal_by_symbol[symbol] = dict(signal) + s = dict(signal) + s["symbol"] = symbol + msg = {"type": "retail_signal_update", **s, "time": s.get("timestamp")} + for q in list(self._retail_signal_listeners): + if not q.full(): + await q.put(msg) + + async def get_latest_retail_signal_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_retail_signal_by_symbol.get(symbol) + return dict(v) if v else None + + async def add_retail_signal_listener(self, maxsize: int = 200) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._retail_signal_listeners.append(q) + return q + + async def remove_retail_signal_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._retail_signal_listeners: + self._retail_signal_listeners.remove(q) + + async def set_latest_signal_debug_for(self, symbol: str, payload: Dict[str, Any]) -> None: + async with self._lock: + p = dict(payload) + p["symbol"] = symbol + self._latest_signal_debug_by_symbol[symbol] = p + msg = {"type": "signal_debug", **p} + for q in list(self._signal_debug_listeners): + if not q.full(): + await q.put(msg) + + async def add_signal_debug_listener(self, maxsize: int = 500) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._signal_debug_listeners.append(q) + return q + + async def remove_signal_debug_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._signal_debug_listeners: + self._signal_debug_listeners.remove(q) + + async def set_latest_pipeline_debug_for(self, symbol: str, payload: Dict[str, Any]) -> None: + async with self._lock: + p = dict(payload) + p["symbol"] = symbol + self._latest_pipeline_debug_by_symbol[symbol] = p + msg = {"type": "pipeline_debug", **p} + for q in list(self._pipeline_debug_listeners): + if not q.full(): + await q.put(msg) + + async def add_pipeline_debug_listener(self, maxsize: int = 1000) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._pipeline_debug_listeners.append(q) + return q + + async def remove_pipeline_debug_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._pipeline_debug_listeners: + self._pipeline_debug_listeners.remove(q) + + # Correlation insights channel + async def set_correlation_insights(self, insights: Dict[str, Any]) -> None: + async with self._lock: + self._latest_correlation_insights = dict(insights) + msg = {"type": "correlation_insights", "insights": dict(insights)} + for q in list(self._correlation_insights_listeners): + if not q.full(): + await q.put(msg) + async def get_correlation_insights(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_correlation_insights) if self._latest_correlation_insights else None + async def set_latest_correlation_insights_for(self, symbol: str, insights: Dict[str, Any]) -> None: + async with self._lock: + p = dict(insights) + p["symbol"] = symbol + self._latest_correlation_insights_by_symbol[symbol] = p + msg = {"type": "correlation_insights", **p} + for q in list(self._correlation_insights_listeners): + if not q.full(): + await q.put(msg) + async def get_latest_correlation_insights_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_correlation_insights_by_symbol.get(symbol) + return dict(v) if v else None + async def add_correlation_insights_listener(self, maxsize: int = 500) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._correlation_insights_listeners.append(q) + if self._latest_correlation_insights is not None: + await q.put({"type": "correlation_insights", "insights": dict(self._latest_correlation_insights)}) + return q + async def remove_correlation_insights_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._correlation_insights_listeners: + self._correlation_insights_listeners.remove(q) + # Data quality channel + async def set_data_quality_report(self, report: Dict[str, Any]) -> None: + async with self._lock: + self._latest_data_quality = dict(report) + msg = {"type": "data_quality", "report": dict(report)} + for q in list(self._data_quality_listeners): + if not q.full(): + await q.put(msg) + async def get_data_quality_report(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_data_quality) if self._latest_data_quality else None + async def set_latest_data_quality_for(self, symbol: str, report: Dict[str, Any]) -> None: + async with self._lock: + p = dict(report) + p["symbol"] = symbol + self._latest_data_quality_by_symbol[symbol] = p + msg = {"type": "data_quality", **p} + for q in list(self._data_quality_listeners): + if not q.full(): + await q.put(msg) + async def get_latest_data_quality_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_data_quality_by_symbol.get(symbol) + return dict(v) if v else None + async def add_data_quality_listener(self, maxsize: int = 500) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._data_quality_listeners.append(q) + if self._latest_data_quality is not None: + await q.put({"type": "data_quality", "report": dict(self._latest_data_quality)}) + return q + async def remove_data_quality_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._data_quality_listeners: + self._data_quality_listeners.remove(q) + # Risk features storage + async def set_latest_risk_features_for(self, symbol: str, features: Dict[str, Any]) -> None: + async with self._lock: + self._risk_features_by_symbol[symbol] = dict(features) + msg = {"type": "risk_features", "symbol": symbol, "features": dict(features)} + for q in list(self._pipeline_debug_listeners): + if not q.full(): + await q.put(msg) + async def get_latest_risk_features_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._risk_features_by_symbol.get(symbol) + return dict(v) if v else None + # Rate limit health channel + async def set_rate_limit_status(self, status: Dict[str, Any]) -> None: + async with self._lock: + self._latest_rate_limits = dict(status) + msg = {"type": "rate_limit", "status": dict(status)} + for q in list(self._rate_limit_listeners): + if not q.full(): + await q.put(msg) + async def get_rate_limit_status(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_rate_limits) if self._latest_rate_limits else None + async def add_rate_limit_listener(self, maxsize: int = 500) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._rate_limit_listeners.append(q) + if self._latest_rate_limits is not None: + await q.put({"type": "rate_limit", "status": dict(self._latest_rate_limits)}) + return q + async def remove_rate_limit_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._rate_limit_listeners: + self._rate_limit_listeners.remove(q) + # Monte Carlo insights channel + async def set_monte_carlo_insights(self, insights: Dict[str, Any]) -> None: + async with self._lock: + self._latest_monte_carlo_insights = dict(insights) + msg = {"type": "monte_carlo_insights", "insights": dict(insights)} + for q in list(self._monte_carlo_insights_listeners): + if not q.full(): + await q.put(msg) + async def get_monte_carlo_insights(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_monte_carlo_insights) if self._latest_monte_carlo_insights else None + async def set_latest_monte_carlo_insights_for(self, symbol: str, insights: Dict[str, Any]) -> None: + async with self._lock: + p = dict(insights) + p["symbol"] = symbol + self._latest_monte_carlo_insights_by_symbol[symbol] = p + msg = {"type": "monte_carlo_insights", **p} + for q in list(self._monte_carlo_insights_listeners): + if not q.full(): + await q.put(msg) + async def get_latest_monte_carlo_insights_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_monte_carlo_insights_by_symbol.get(symbol) + return dict(v) if v else None + async def add_monte_carlo_insights_listener(self, maxsize: int = 500) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._monte_carlo_insights_listeners.append(q) + if self._latest_monte_carlo_insights is not None: + await q.put({"type": "monte_carlo_insights", "insights": dict(self._latest_monte_carlo_insights)}) + return q + async def remove_monte_carlo_insights_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._monte_carlo_insights_listeners: + self._monte_carlo_insights_listeners.remove(q) + # VaR insights channel + async def set_var_insights(self, insights: Dict[str, Any]) -> None: + async with self._lock: + self._latest_var_insights = dict(insights) + msg = {"type": "var_insights", "insights": dict(insights)} + for q in list(self._var_insights_listeners): + if not q.full(): + await q.put(msg) + async def get_var_insights(self) -> Optional[Dict[str, Any]]: + async with self._lock: + return dict(self._latest_var_insights) if self._latest_var_insights else None + async def set_latest_var_insights_for(self, symbol: str, insights: Dict[str, Any]) -> None: + async with self._lock: + p = dict(insights) + p["symbol"] = symbol + self._latest_var_insights_by_symbol[symbol] = p + msg = {"type": "var_insights", **p} + for q in list(self._var_insights_listeners): + if not q.full(): + await q.put(msg) + async def get_latest_var_insights_for(self, symbol: str) -> Optional[Dict[str, Any]]: + async with self._lock: + v = self._latest_var_insights_by_symbol.get(symbol) + return dict(v) if v else None + async def add_var_insights_listener(self, maxsize: int = 500) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + self._var_insights_listeners.append(q) + if self._latest_var_insights is not None: + await q.put({"type": "var_insights", "insights": dict(self._latest_var_insights)}) + return q + async def remove_var_insights_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if q in self._var_insights_listeners: + self._var_insights_listeners.remove(q) + + # Portfolio targets and health + async def set_portfolio_targets(self, weights: Dict[str, float]) -> None: + async with self._lock: + try: + self._portfolio_targets = dict(weights) + except Exception: + self._portfolio_targets = dict(weights or {}) + msg = {"type": "portfolio_update", "weights": dict(self._portfolio_targets)} + for q in list(getattr(self, "_portfolio_listeners", [])): + if not q.full(): + await q.put(msg) + async def get_portfolio_targets(self) -> Dict[str, float]: + async with self._lock: + return dict(getattr(self, "_portfolio_targets", {}) or {}) + async def add_portfolio_listener(self, maxsize: int = 100) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + async with self._lock: + if not hasattr(self, "_portfolio_listeners"): + self._portfolio_listeners = [] + self._portfolio_listeners.append(q) + if hasattr(self, "_portfolio_targets"): + await q.put({"type": "portfolio_update", "weights": dict(getattr(self, "_portfolio_targets", {}) or {})}) + return q + async def remove_portfolio_listener(self, q: asyncio.Queue) -> None: + async with self._lock: + if hasattr(self, "_portfolio_listeners") and q in self._portfolio_listeners: + self._portfolio_listeners.remove(q) + async def set_portfolio_health(self, health: Dict[str, Any]) -> None: + async with self._lock: + try: + self._portfolio_health = dict(health) + except Exception: + self._portfolio_health = dict(health or {}) + msg = {"type": "portfolio_health", "health": dict(self._portfolio_health)} + for q in list(getattr(self, "_portfolio_listeners", [])): + if not q.full(): + await q.put(msg) + async def get_portfolio_health(self) -> Dict[str, Any]: + async with self._lock: + return dict(getattr(self, "_portfolio_health", {}) or {}) + async def set_selected_method(self, method: str) -> None: + async with self._lock: + self._selected_method = str(method or "") + try: + os.makedirs("data", exist_ok=True) + with open(os.path.join("data", "portfolio_selection.json"), "w", encoding="utf-8") as f: + f.write('{"selected_method": "' + self._selected_method + '"}') + except Exception: + pass + async def get_selected_method(self) -> str: + async with self._lock: + if getattr(self, "_selected_method", None): + return str(self._selected_method) + try: + with open(os.path.join("data", "portfolio_selection.json"), "r", encoding="utf-8") as f: + c = f.read() + if c: + v = c.strip() + if v.startswith("{"): + m = v + try: + import json as _json + d = _json.loads(m) + self._selected_method = str(d.get("selected_method") or "") + except Exception: + self._selected_method = "" + else: + self._selected_method = "" + except Exception: + self._selected_method = "" + return str(getattr(self, "_selected_method", "") or "")