diff --git a/monitoring/real_time_dashboard.py b/monitoring/real_time_dashboard.py
new file mode 100644
index 0000000..bde12bf
--- /dev/null
+++ b/monitoring/real_time_dashboard.py
@@ -0,0 +1,551 @@
+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