mirror of
https://github.com/umaiskhan-ops/ApexFX-High-Fidelity-Quant-Ecosystem..git
synced 2026-08-06 15:27:47 +00:00
Umais Khan
552 lines
29 KiB
Python
552 lines
29 KiB
Python
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
|