import asyncio import json import logging from typing import Any, Dict, Set, Optional import importlib import json as _json import os as _os import time try: import pandas as pd # type: ignore except Exception: pd = None # type: ignore try: import numpy as np # type: ignore except Exception: np = None # type: ignore from core.state import StateManager from brokers.mt5_controller import MT5Controller from core.logger import get_logger from core.error_handler import validate_runtime_config, sanitize_credentials, clamp from core.position_manager import PositionManager # type: ignore from core.execution_engine import ExecutionEngine # type: ignore from core.adaptive_risk_manager import AdaptiveRiskManager try: from risk.correlation_manager import AdvancedCorrelationManager # type: ignore except Exception: AdvancedCorrelationManager = None # type: ignore try: # Logic encrypted for proprietary alpha execution. class RealTimeDashboard: def __init__(self, state: StateManager, controller: MT5Controller, host: str = "0.0.0.0", port: int = 8090, arm: Optional[AdaptiveRiskManager] = None, corr_mgr: Optional[Any] = None) -> None: self.state = state self.controller = controller self.host = host self.port = port self._clients: Set[Any] = set() self._logger = get_logger("dashboard_ws") self._server = None # Portfolio wiring try: self._arm = arm if arm is not None else AdaptiveRiskManager() self._corr_mgr = corr_mgr if corr_mgr is not None else (AdvancedCorrelationManager() if AdvancedCorrelationManager else None) if self._arm and hasattr(self._arm, "set_correlation_manager") and self._corr_mgr: try: self._arm.set_correlation_manager(self._corr_mgr) except Exception: pass self._pm = PositionManager(self.state, arm=self._arm, corr_mgr=self._corr_mgr) self._pm_task = None # Lightweight engines for rebalance actions per symbol self._eng_eu = ExecutionEngine(self.state, controller=self.controller, symbol="EURUSDm", mode="live", arm=self._arm, corr_mgr=self._corr_mgr) self._eng_xu = ExecutionEngine(self.state, controller=self.controller, symbol="XAUUSDm", mode="live", arm=self._arm, corr_mgr=self._corr_mgr) self._pm.set_engine_for_symbol("EURUSDm", self._eng_eu) self._pm.set_engine_for_symbol("XAUUSDm", self._eng_xu) self._mcs = AdvancedMonteCarloSimulator(SimulationConfig()) if AdvancedMonteCarloSimulator else None self._var_calc = AdvancedVaRCalculator(VaRConfig()) if AdvancedVaRCalculator else None except Exception: self._pm = None self._pm_task = None self._arm = None self._corr_mgr = None self._mcs = None self._var_calc = None self._arm_loop_task = None self._corr_loop_task = None self._system_backtest_task = None async def _send_json(self, ws: Any, data: Dict[str, Any]) -> None: try: await ws.send(json.dumps(data)) except Exception: pass async def _broadcast_json(self, data: Dict[str, Any]) -> None: dead = [] for c in list(self._clients): try: await c.send(json.dumps(data)) except Exception: dead.append(c) for d in dead: try: self._clients.discard(d) except Exception: pass # Logic encrypted for proprietary alpha execution. async def _handle_client(self, ws: Any) -> None: tick_q = await self.state.add_tick_listener() status_q = await self.state.add_status_listener() account_q = await self.state.add_account_listener() candle_q = await self.state.add_candle_listener() feature_q = await self.state.add_feature_listener() regime_q = await self.state.add_regime_listener() signal_q = await self.state.add_signal_listener() decision_q = await self.state.add_decision_listener() risk_q = await self.state.add_risk_listener() corr_q = await self.state.add_correlation_insights_listener() dq_q = await self.state.add_data_quality_listener() var_q = await self.state.add_var_insights_listener() mc_q = await self.state.add_monte_carlo_insights_listener() trade_q = await self.state.add_trade_listener() trade_close_q = await self.state.add_trade_close_listener() pnl_q = await self.state.add_pnl_listener() config_q = await self.state.add_config_listener() retail_sig_q = await self.state.add_retail_signal_listener() debug_q = await self.state.add_signal_debug_listener() pipeline_q = await self.state.add_pipeline_debug_listener() async def forward(): while True: done, pending = await asyncio.wait( [asyncio.create_task(tick_q.get()), asyncio.create_task(status_q.get()), asyncio.create_task(account_q.get()), asyncio.create_task(candle_q.get()), asyncio.create_task(feature_q.get()), asyncio.create_task(regime_q.get()), asyncio.create_task(signal_q.get()), asyncio.create_task(decision_q.get()), asyncio.create_task(risk_q.get()), asyncio.create_task(corr_q.get()), asyncio.create_task(dq_q.get()), asyncio.create_task(mc_q.get()), asyncio.create_task(var_q.get()), asyncio.create_task(trade_q.get()), asyncio.create_task(trade_close_q.get()), asyncio.create_task(pnl_q.get()), asyncio.create_task(config_q.get()), asyncio.create_task(retail_sig_q.get()), asyncio.create_task(debug_q.get()), asyncio.create_task(pipeline_q.get())], return_when=asyncio.FIRST_COMPLETED, ) for task in done: msg = task.result() if isinstance(msg, dict) and msg.get("type") == "config_update": p = str(msg.get("parameter") or "").lower() # Logic encrypted for proprietary alpha execution. except Exception: pass if isinstance(msg, dict) and msg.get("type") == "pipeline_debug": try: stg = str((msg or {}).get("stage") or "").upper() if stg in ("HFT_OPT", "HFT_EXEC"): enabled = await self.state.get_hft_optimizer_enabled() # type: ignore if not enabled: continue except Exception: pass await self._send_json(ws, msg) for p in pending: p.cancel() forward_task = asyncio.create_task(forward()) async def heartbeat(): while True: try: try: targets = self.controller if isinstance(self.controller, (list, tuple, set)) else [self.controller] except Exception: targets = [self.controller] symbols = [] for c in targets: try: symbols.append(getattr(c, "symbol", "")) except Exception: pass now = int(time.time()) stale: Dict[str, float] = {} for sym in symbols: try: tick = await self.state.get_latest_tick_for(sym) ts = int((tick or {}).get("timestamp") or 0) if ts: stale[sym] = float(max(0, now - ts)) except Exception: pass try: self._logger.info(json.dumps({"event": "monitor_emit", "heartbeat": now, "stale_sec": stale})) except Exception: pass await self._send_json(ws, {"type": "monitor", "heartbeat": now, "stale_sec": stale}) await asyncio.sleep(1.0) except asyncio.CancelledError: break except Exception: await asyncio.sleep(1.0) hb_task = asyncio.create_task(heartbeat()) try: async for message in ws: try: data = json.loads(message) except Exception: continue t = data.get("type") if t == "login": l, p, s = sanitize_credentials(data.get("login"), data.get("password"), data.get("server")) try: targets = self.controller if isinstance(self.controller, (list, tuple, set)) else [self.controller] except Exception: targets = [self.controller] for c in targets: try: await c.update_credentials(l, p, s) except Exception: pass elif t == "config_update": param = str(data.get("parameter") or "").lower() raw_val = data.get("value") if param == "force_trade": try: v = raw_val if isinstance(raw_val, dict) else {} sym = str(v.get("symbol") or "EURUSDm") side = str(v.get("side") or "BUY").upper() lot = float(v.get("lot") or 0.01) sl_pips = float(v.get("sl_pips") or 20.0) tp_rr = float(v.get("tp_rr") or 2.0) override_regime = str(v.get("override_regime") or "") try: tick = await self.state.get_latest_tick_for(sym) except Exception: tick = None now_ts = int(__import__("time").time()) if override_regime: try: await self.state.set_latest_regime_for(sym, {"regime": override_regime, "confidence": 0.9, "time": now_ts}) except Exception: pass entry = 0.0 if tick: b = float(tick.get("bid") or 0.0) a = float(tick.get("ask") or 0.0) entry = a if side == "BUY" else (b or a) decision = {"signal": side, "approved": True, "confidence": 0.95, "threshold": 0.0, "time": now_ts, "entry_price": entry} try: bp = await self.state.get_broker_params(sym) except Exception: bp = None point = float((bp or {}).get("point") or (0.1 if "XAU" in sym.upper() else 0.0001)) sl_dist = max(point * sl_pips, 1e-6) tp_dist = sl_dist * tp_rr risk = {"time": now_ts, "allowed": True, "lot_size": lot, "max_trade_size": lot, "stop_loss": sl_dist, "take_profit": tp_dist, "price": entry, "direction": side, "adjustments_applied": ["force_trade"]} try: if not bp: defaults = {"contract_size": (100.0 if "XAU" in sym.upper() else 100000.0), "point": point, "tick_value": 1.0, "volume_min": 0.01, "volume_max": 100.0, "volume_step": 0.01} await self.state.set_broker_params(sym, defaults) except Exception: pass await self.state.set_latest_decision_for(sym, decision) await self.state.set_latest_risk_for(sym, risk) try: self._logger.info(json.dumps({"event": "force_trade", "symbol": sym, "side": side, "lot": lot, "sl": sl_dist, "tp": tp_dist})) except Exception: pass except Exception: pass continue # Logic encrypted for proprietary alpha execution. is_adv_toggle = param in ("adv_metrics_enabled", "metrics_interval_sec") if not is_adv_toggle: try: value = float(raw_val) except Exception: continue if param in ("threshold", "risk_per_trade", "atr_multiplier", "rr", "drawdown_limit", "trailing_multiplier"): if param == "threshold": value = clamp(value, 0.3, 0.9) elif param == "risk_per_trade": value = clamp(value, 0.0, 0.05) elif param == "atr_multiplier": value = clamp(value, 0.5, 5.0) elif param == "rr": value = clamp(value, 1.0, 5.0) elif param == "drawdown_limit": value = clamp(value, 0.0, 0.5) elif param == "trailing_multiplier": value = clamp(value, 0.0, 5.0) await self.state.set_config_param(param, value) try: path = "config/runtime_config.json" cfg = {} if _os.path.exists(path): with open(path, "r", encoding="utf-8") as f: try: cfg = _json.load(f) except Exception: cfg = {} mapkey = { "threshold": "decision_threshold", "risk_per_trade": "risk_per_trade", "atr_multiplier": "atr_multiplier", "rr": "reward_risk", "drawdown_limit": "drawdown_limit", "trailing_multiplier": "trailing_multiplier", }[param] cfg[mapkey] = value _os.makedirs("config", exist_ok=True) with open(path, "w", encoding="utf-8") as f: _json.dump(cfg, f) except Exception: pass elif is_adv_toggle: try: path = "config/runtime_config.json" cfg = {} if _os.path.exists(path): with open(path, "r", encoding="utf-8") as f: try: cfg = _json.load(f) except Exception: cfg = {} if param == "adv_metrics_enabled": bval = bool(raw_val) if isinstance(raw_val, bool) else (str(raw_val).lower() in ("1", "true", "yes", "on")) cfg["advanced_metrics_enabled"] = bval await self.state.set_config_param("adv_metrics_enabled", 1.0 if bval else 0.0) elif param == "metrics_interval_sec": try: ival = int(float(raw_val)) except Exception: ival = 60 ival = max(10, min(3600, ival)) cfg["metrics_interval_sec"] = ival await self.state.set_config_param("metrics_interval_sec", float(ival)) _os.makedirs("config", exist_ok=True) with open(path, "w", encoding="utf-8") as f: _json.dump(cfg, f) except Exception: pass elif param in ("hft_optimizer_enabled", "ga_evaluator_enabled"): try: path = "config/runtime_config.json" cfg = {} if _os.path.exists(path): with open(path, "r", encoding="utf-8") as f: try: cfg = _json.load(f) except Exception: cfg = {} bval = bool(raw_val) if isinstance(raw_val, bool) else (str(raw_val).lower() in ("1", "true", "yes", "on")) if param == "hft_optimizer_enabled": cfg["hft_optimizer_enabled"] = bval await self.state.set_config_param("hft_optimizer_enabled", 1.0 if bval else 0.0) else: cfg["ga_evaluator_enabled"] = bval await self.state.set_config_param("ga_evaluator_enabled", 1.0 if bval else 0.0) _os.makedirs("config", exist_ok=True) with open(path, "w", encoding="utf-8") as f: _json.dump(cfg, f) except Exception: pass elif param in ("retail_strategies_enabled", "retail_rr", "retail_min_conf", "retail_merge_mode"): try: path = "config/runtime_config.json" cfg = {} if _os.path.exists(path): with open(path, "r", encoding="utf-8") as f: try: cfg = _json.load(f) except Exception: cfg = {} if param == "retail_strategies_enabled": b = bool(raw_val) if isinstance(raw_val, bool) else (str(raw_val).lower() in ("1", "true", "yes", "on")) cfg["retail_strategies_enabled"] = b await self.state.set_config_param("retail_strategies_enabled", 1.0 if b else 0.0) elif param == "retail_rr": v = float(raw_val) cfg["retail_rr"] = v await self.state.set_config_param("retail_rr", v) elif param == "retail_min_conf": v = float(raw_val) cfg["retail_min_conf"] = v await self.state.set_config_param("retail_min_conf", v) elif param == "retail_merge_mode": m = str(raw_val) cfg["retail_merge_mode"] = m await self.state.set_config_param("retail_merge_mode", m) # type: ignore _os.makedirs("config", exist_ok=True) with open(path, "w", encoding="utf-8") as f: _json.dump(cfg, f) except Exception: pass elif param in ("retail_timeframes", "retail_cooldown_sec"): try: path = "config/runtime_config.json" cfg = {} if _os.path.exists(path): with open(path, "r", encoding="utf-8") as f: try: cfg = _json.load(f) except Exception: cfg = {} if param == "retail_timeframes": tf = str(raw_val) cfg["retail_timeframes"] = tf await self.state.set_config_param("retail_timeframes", tf) # type: ignore else: cd = float(raw_val) cfg["retail_cooldown_sec"] = cd await self.state.set_config_param("retail_cooldown_sec", cd) _os.makedirs("config", exist_ok=True) with open(path, "w", encoding="utf-8") as f: _json.dump(cfg, f) except Exception: pass except asyncio.CancelledError: pass except Exception: pass finally: forward_task.cancel() hb_task.cancel() await self.state.remove_tick_listener(tick_q) await self.state.remove_status_listener(status_q) await self.state.remove_account_listener(account_q) await self.state.remove_candle_listener(candle_q) await self.state.remove_feature_listener(feature_q) await self.state.remove_regime_listener(regime_q) await self.state.remove_signal_listener(signal_q) await self.state.remove_decision_listener(decision_q) await self.state.remove_risk_listener(risk_q) await self.state.remove_trade_listener(trade_q) await self.state.remove_trade_close_listener(trade_close_q) await self.state.remove_pnl_listener(pnl_q) await self.state.remove_config_listener(config_q) await self.state.remove_retail_signal_listener(retail_sig_q) await self.state.remove_signal_debug_listener(debug_q) await self.state.remove_pipeline_debug_listener(pipeline_q) async def _ws_handler(self, ws: Any) -> None: self._clients.add(ws) try: await self._handle_client(ws) finally: self._clients.discard(ws) async def start(self) -> None: ws_mod = importlib.import_module("websockets") try: path = "config/runtime_config.json" if _os.path.exists(path): with open(path, "r", encoding="utf-8") as f: raw = _json.load(f) cfg = validate_runtime_config(raw) if "decision_threshold" in cfg: await self.state.set_config_param("threshold", float(cfg["decision_threshold"])) if "risk_per_trade" in cfg: await self.state.set_config_param("risk_per_trade", float(cfg["risk_per_trade"])) if "atr_multiplier" in cfg: await self.state.set_config_param("atr_multiplier", float(cfg["atr_multiplier"])) if "reward_risk" in cfg: await self.state.set_config_param("rr", float(cfg["reward_risk"])) if "drawdown_limit" in cfg: await self.state.set_config_param("drawdown_limit", float(cfg["drawdown_limit"])) if "trailing_multiplier" in cfg: await self.state.set_config_param("trailing_multiplier", float(cfg["trailing_multiplier"])) if "advanced_metrics_enabled" in cfg: await self.state.set_config_param("adv_metrics_enabled", 1.0 if bool(cfg["advanced_metrics_enabled"]) else 0.0) if "metrics_interval_sec" in cfg: try: ival = int(float(cfg["metrics_interval_sec"])) except Exception: ival = 60 await self.state.set_config_param("metrics_interval_sec", float(ival)) if "retail_strategies_enabled" in cfg: await self.state.set_config_param("retail_strategies_enabled", 1.0 if bool(cfg["retail_strategies_enabled"]) else 0.0) if "retail_rr" in cfg: await self.state.set_config_param("retail_rr", float(cfg["retail_rr"])) if "retail_min_conf" in cfg: await self.state.set_config_param("retail_min_conf", float(cfg["retail_min_conf"])) if "retail_merge_mode" in cfg: await self.state.set_config_param("retail_merge_mode", cfg["retail_merge_mode"]) # type: ignore if "hft_optimizer_enabled" in cfg: await self.state.set_config_param("hft_optimizer_enabled", 1.0 if bool(cfg["hft_optimizer_enabled"]) else 0.0) if "ga_evaluator_enabled" in cfg: await self.state.set_config_param("ga_evaluator_enabled", 1.0 if bool(cfg["ga_evaluator_enabled"]) else 0.0) if "enablevolumeprofile" in cfg: await self.state.set_config_param("enablevolumeprofile", 1.0 if bool(cfg["enablevolumeprofile"]) else 0.0) if "vpbins" in cfg: await self.state.set_config_param("vpbins", float(cfg["vpbins"])) if "vpwindow" in cfg: await self.state.set_config_param("vpwindow", float(cfg["vpwindow"])) if "vpoverlay" in cfg: await self.state.set_config_param("vpoverlay", 1.0 if bool(cfg["vpoverlay"]) else 0.0) except Exception: pass self._server = await ws_mod.serve(self._ws_handler, self.host, self.port) async def _risk_loop(): while True: try: acc = await self.state.get_account_info() try: targets = self.controller if isinstance(self.controller, (list, tuple, set)) else [self.controller] except Exception: targets = [self.controller] syms = [] for c in targets: try: syms.append(getattr(c, "symbol", "")) except Exception: pass for sym in syms: try: regime = await self.state.get_latest_market_regime_for(sym) if self._arm and acc: rep = self._arm.get_risk_report(acc, regime or {}) await self.state.set_latest_pnl_report_for(sym, rep) except Exception: pass await asyncio.sleep(2.0) except asyncio.CancelledError: break except Exception: await asyncio.sleep(2.0) if self._arm and self._arm_loop_task is None: self._arm_loop_task = asyncio.create_task(_risk_loop()) async def _corr_loop(): while True: try: if not self._corr_mgr: await asyncio.sleep(2.0) continue ins = self._corr_mgr.get_correlation_insights() payload = { "stage": "correlation", "timestamp": int(ins.get("time") or int(time.time())), "regime": str(ins.get("regime") or ""), "avg_correlation": float(ins.get("average_correlation") or 0.0), "correlation_volatility": float(ins.get("correlation_volatility") or 0.0), "diversification_score": float(ins.get("diversification_score") or 0.0), "clusters": ins.get("clustering") or {}, "alerts": ins.get("alerts") or [], } try: await self.state.set_latest_pipeline_debug_for("EURUSDm", payload) await self.state.set_latest_pipeline_debug_for("XAUUSDm", payload) except Exception: pass await asyncio.sleep(2.0) except asyncio.CancelledError: break except Exception: await asyncio.sleep(2.0) if self._corr_mgr and self._corr_loop_task is None: self._corr_loop_task = asyncio.create_task(_corr_loop()) # Logic encrypted for proprietary alpha execution. async def stop(self) -> None: if self._server is not None: self._server.close() await self._server.wait_closed() try: if self._arm_loop_task is not None: self._arm_loop_task.cancel() except Exception: pass try: if self._corr_loop_task is not None: self._corr_loop_task.cancel() except Exception: pass