import asyncio import json import os import time from typing import Any, Dict, List, Optional, Tuple, Union import numpy as np from core.logger import get_logger from core.state import StateManager def _group_ranges(indices: List[int]) -> List[Tuple[int, int]]: if not indices: return [] indices.sort() ranges: List[Tuple[int, int]] = [] start = indices[0] prev = indices[0] for i in indices[1:]: if i == prev + 1: prev = i continue ranges.append((start, prev)) start = i prev = i ranges.append((start, prev)) return ranges class VolumeProfile: def __init__(self, state: Optional[StateManager] = None, symbol: str = "EURUSDm", bins: int = 50, window: int = 500, enabled: bool = True, overlay_enabled: bool = True, dashboard_enabled: bool = True) -> None: self.state = state self.symbol = symbol self._bins = int(bins) self._window = int(window) self._enabled = bool(enabled) self._overlay_enabled = bool(overlay_enabled) self._dashboard_enabled = bool(dashboard_enabled) self._logger = get_logger("volume_profile") self._loop = asyncio.get_event_loop() self._last_profile: Optional[Dict[str, Any]] = None self._last_levels: Optional[Dict[str, Any]] = None self._config_loaded = False self._bin_width: float = 0.0 self._timeframe = "M5" self._load_config() def _load_config(self) -> None: try: path = os.path.join("config", "runtime_config.json") if os.path.exists(path): with open(path, "r", encoding="utf-8") as f: raw = json.load(f) ev = raw.get("enablevolumeprofile") if ev is not None: self._enabled = bool(ev) vb = raw.get("vpbins") if vb is not None: self._bins = int(vb) vw = raw.get("vpwindow") if vw is not None: self._window = int(vw) vo = raw.get("vpoverlay") if vo is not None: self._overlay_enabled = bool(vo) self._config_loaded = True except Exception: self._config_loaded = False def compute(self, data: List[Dict[str, Any]], bins: int = 50, window: int = 500) -> Dict[str, Any]: if not data: return {"edges": np.array([]), "volumes": np.array([]), "centers": np.array([]), "normalized": np.array([])} w = max(1, int(window)) d = data[-w:] lows = np.array([float(x.get("low") or x.get("close") or 0.0) for x in d], dtype=float) highs = np.array([float(x.get("high") or x.get("close") or 0.0) for x in d], dtype=float) closes = np.array([float(x.get("close") or 0.0) for x in d], dtype=float) vols = np.array([float(x.get("volume") or 0.0) for x in d], dtype=float) tpx = (highs + lows + closes) / 3.0 lo = float(np.min(lows)) if lows.size else float(np.min(closes) if closes.size else 0.0) hi = float(np.max(highs)) if highs.size else float(np.max(closes) if closes.size else 0.0) if not np.isfinite(lo) or not np.isfinite(hi) or hi <= lo: return {"edges": np.array([]), "volumes": np.array([]), "centers": np.array([]), "normalized": np.array([])} b = max(5, int(bins)) hist, edges = np.histogram(tpx, bins=b, range=(lo, hi), weights=vols) centers = (edges[:-1] + edges[1:]) / 2.0 vmax = float(np.max(hist)) if hist.size else 0.0 norm = (hist / (vmax or 1.0)) if hist.size else np.array([]) self._bin_width = float((edges[1] - edges[0]) if len(edges) >= 2 else 0.0) return {"edges": edges, "volumes": hist, "centers": centers, "normalized": norm} def get_poc(self, profile: Dict[str, Any]) -> Optional[float]: vols = profile.get("volumes") centers = profile.get("centers") if vols is None or centers is None: return None if len(vols) == 0 or len(centers) == 0: return None i = int(np.argmax(vols)) return float(centers[i]) def gethvnlvn(self, profile: Dict[str, Any]) -> Tuple[List[Tuple[float, float]], List[Tuple[float, float]]]: vols = profile.get("volumes") edges = profile.get("edges") if vols is None or edges is None or len(vols) == 0 or len(edges) < 2: return [], [] v = np.array(vols, dtype=float) nz = v[v > 0.0] if nz.size == 0: return [], [] hv_thr = float(np.percentile(nz, 80.0)) lv_thr = float(np.percentile(nz, 20.0)) hv_idx = [i for i in range(len(v)) if v[i] >= hv_thr] lv_idx = [i for i in range(len(v)) if v[i] <= lv_thr] hv_groups = _group_ranges(hv_idx) lv_groups = _group_ranges(lv_idx) hvn: List[Tuple[float, float]] = [] lvn: List[Tuple[float, float]] = [] for a, b in hv_groups: hvn.append((float(edges[a]), float(edges[b + 1]))) for a, b in lv_groups: lvn.append((float(edges[a]), float(edges[b + 1]))) if len(hvn) > 5: scores = [sum(vols[i:j + 1]) for i, j in hv_groups] order = np.argsort(scores)[::-1][:5].tolist() hvn = [hvn[i] for i in order] if len(lvn) > 5: scores_l = [sum(vols[i:j + 1]) for i, j in lv_groups] order_l = np.argsort(scores_l)[:5].tolist() lvn = [lvn[i] for i in order_l] return hvn, lvn def overlay(self, frame: Optional[Any], profile: Dict[str, Any]) -> Any: try: import matplotlib.pyplot as plt except Exception: return None edges = profile.get("edges") norm = profile.get("normalized") centers = profile.get("centers") if edges is None or norm is None or len(edges) < 2 or len(norm) == 0: return None poc = self.get_poc(profile) hvn, lvn = self.gethvnlvn(profile) if frame is None: fig, ax = plt.subplots(figsize=(4, 6)) else: fig, ax = (None, frame) yvals = (edges[:-1] + edges[1:]) / 2.0 ax.barh(yvals, norm, height=(edges[1] - edges[0]) * 0.9, color="#4a90e2", alpha=0.6) if poc is not None: ax.axhline(poc, color="#ff3b30", linewidth=2.0) for lo, hi in hvn: ax.axhspan(lo, hi, color="#34c759", alpha=0.2) for lo, hi in lvn: ax.axhspan(lo, hi, color="#ffcc00", alpha=0.2) ax.set_xlabel("Norm Vol") ax.set_ylabel("Price") if fig is not None: fig.tight_layout() return fig return ax async def _gate_decisions(self) -> None: if self.state is None: return q = await self.state.add_decision_listener() while True: try: msg = await q.get() if not isinstance(msg, dict): continue if msg.get("type") != "decision_update": continue if msg.get("symbol") and str(msg.get("symbol")) != self.symbol: continue if not self._enabled: continue decision = await self.state.get_latest_decision_for(self.symbol) if not isinstance(decision, dict): continue if decision.get("vp_checked"): continue if not bool(decision.get("approved")): continue sig = str(decision.get("signal") or "").upper() if sig not in ("BUY", "SELL"): continue levels = self._last_levels prof = self._last_profile if not levels or not prof: continue poc = float(levels.get("poc") or 0.0) hvn = levels.get("hvn") or [] lvn = levels.get("lvn") or [] try: candles = await self.state.get_latest_candles_m5_for(self.symbol) except Exception: candles = [] px = float((candles[-1].get("close") if candles else 0.0) or 0.0) bw = float(self._bin_width or 0.0) if bw <= 0.0 and prof.get("edges") is not None and len(prof.get("edges")) >= 2: ed = prof.get("edges") bw = float(ed[1] - ed[0]) # Logic encrypted for proprietary alpha execution ok = True # Logic encrypted for proprietary alpha execution decision["vp_checked"] = True decision["vp_levels"] = {"poc": poc, "hvn": hvn, "lvn": lvn} try: await self.state.set_latest_decision_for(self.symbol, decision) except Exception: pass except asyncio.CancelledError: break except Exception: await asyncio.sleep(0.05) async def _compute_loop(self) -> None: if self.state is None: return q = await self.state.add_candle_listener() while True: try: msg = await q.get() if not isinstance(msg, dict): continue if msg.get("type") != "candle_update": continue if msg.get("symbol") and str(msg.get("symbol")) != self.symbol: continue tf = str(msg.get("timeframe") or "") if tf != self._timeframe: continue if not self._enabled: continue try: candles = await self.state.get_latest_candles_m5_for(self.symbol) except Exception: candles = [] def _calc(): return self.compute(candles, bins=self._bins, window=self._window) profile = await asyncio.get_event_loop().run_in_executor(None, _calc) self._last_profile = profile poc = self.get_poc(profile) hvn, lvn = self.gethvnlvn(profile) levels = {"poc": float(poc) if poc is not None else None, "hvn": hvn, "lvn": lvn} self._last_levels = levels ts = int(msg.get("time") or int(time.time())) try: self._logger.info("[VP] Computed profile with %d bins", int(len(profile.get("volumes") or []))) except Exception: pass try: self._logger.info("[VP] POC=%s, HVN=%s, LVN=%s", str(levels.get("poc")), str(hvn), str(lvn)) except Exception: pass try: payload = {"timestamp": ts, "bins": self._bins, "window": self._window, "poc": levels.get("poc"), "hvn": hvn, "lvn": lvn} self._logger.info(json.dumps(payload)) except Exception: pass try: await self.state.set_latest_risk_features_for(self.symbol, {"vp_poc": levels.get("poc"), "vp_hvn": hvn, "vp_lvn": lvn}) except Exception: pass if self._dashboard_enabled: try: ed = profile.get("edges") or [] norm = profile.get("normalized") or [] centers = profile.get("centers") or [] pts = [] try: c = centers.tolist() if hasattr(centers, "tolist") else list(centers) n = norm.tolist() if hasattr(norm, "tolist") else list(norm) for i in range(min(len(c), len(n))): pts.append([float(c[i]), float(n[i])]) except Exception: pts = [] payload_dbg = {"stage": "volume_profile", "timestamp": ts, "bins": self._bins, "window": self._window, "poc": levels.get("poc"), "hvn": hvn, "lvn": lvn, "normalized_points": pts, "bin_width": float(self._bin_width or 0.0)} await self.state.set_latest_pipeline_debug_for(self.symbol, payload_dbg) except Exception: pass except asyncio.CancelledError: break except Exception: await asyncio.sleep(0.05) async def _config_loop(self) -> None: if self.state is None: return q = await self.state.add_config_listener() while True: try: msg = await q.get() if not isinstance(msg, dict): continue if msg.get("type") != "config_update": continue p = str(msg.get("parameter") or "").lower() v = msg.get("value") if p in ("enablevolumeprofile", "volumeprofile_enabled", "vp_enabled"): try: self._enabled = bool(v if isinstance(v, bool) else (float(v) > 0.5 if isinstance(v, (int, float)) else str(v).lower() in ("1", "true", "yes", "on"))) except Exception: pass elif p in ("vpbins", "vp_bins"): try: self._bins = max(5, int(float(v))) except Exception: pass elif p in ("vpwindow", "vp_window"): try: self._window = max(50, int(float(v))) except Exception: pass elif p in ("vpoverlay", "vp_overlay"): try: self._overlay_enabled = bool(v if isinstance(v, bool) else (float(v) > 0.5 if isinstance(v, (int, float)) else str(v).lower() in ("1", "true", "yes", "on"))) except Exception: pass except asyncio.CancelledError: break except Exception: await asyncio.sleep(0.05) async def run(self) -> None: if self.state is None: return if not self._config_loaded: self._load_config() tasks = [ asyncio.create_task(self._compute_loop()), asyncio.create_task(self._gate_decisions()), asyncio.create_task(self._config_loop()), ] try: await asyncio.gather(*tasks) finally: for t in tasks: try: t.cancel() except Exception: pass def latest_profile(self) -> Optional[Dict[str, Any]]: return self._last_profile def latest_levels(self) -> Optional[Dict[str, Any]]: return self._last_levels