mirror of
https://github.com/umaiskhan-ops/ApexFX-High-Fidelity-Quant-Ecosystem..git
synced 2026-08-03 05:47:45 +00:00
Umais Khan
1111 lines
54 KiB
Python
1111 lines
54 KiB
Python
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 "")
|