from __future__ import annotations import asyncio import json import math import os import random import time import importlib.util from collections import deque from dataclasses import dataclass, asdict, field from pathlib import Path from typing import Any, Deque, Dict, List, Optional, Tuple import websockets TIMEFRAMES = [30, 60, 120, 180, 300] def clamp(value: float, lo: float = 0.0, hi: float = 1.0) -> float: if not math.isfinite(value): return lo return max(lo, min(hi, value)) def safe_float(value: Any, default: float = 0.0) -> float: try: f = float(value) return f if math.isfinite(f) else default except Exception: return default def now_utc() -> float: return time.time() def tf_label(tf_seconds: int) -> str: return {30: "30s", 60: "1m", 120: "2m", 180: "3m", 300: "5m"}.get(int(tf_seconds), f"{tf_seconds}s") def detect_market_type(symbol: str) -> str: s = (symbol or "").lower() if s.startswith("frx") or "forex" in s: return "forex" if s.startswith("cry") or "crypto" in s: return "crypto" return "unknown" def ensure_repo_engine_path() -> Path: candidates = [ Path(__file__).resolve().parent / "maythos_patched.py", Path("/mnt/data/maythos_patched.py"), ] for p in candidates: if p.exists(): return p raise FileNotFoundError("maythos_patched.py not found in repo root or /mnt/data") def load_maythos_module(): path = ensure_repo_engine_path() spec = importlib.util.spec_from_file_location("maythos_patched", str(path)) if spec is None or spec.loader is None: raise RuntimeError(f"Unable to load MAYTHOS module from {path}") module = importlib.util.module_from_spec(spec) import sys sys.modules[spec.name] = module spec.loader.exec_module(module) # type: ignore[arg-type] return module _ENGINE_MODULE = None def get_engine_module(): global _ENGINE_MODULE if _ENGINE_MODULE is None: _ENGINE_MODULE = load_maythos_module() return _ENGINE_MODULE @dataclass class CandleBar: timeframe: int start_ts: float end_ts: float open: float high: float low: float close: float volume: float = 0.0 spread: float = 0.0 bid: float = 0.0 ask: float = 0.0 source_id: str = "deriv" closed: bool = False def update(self, price: float, ts: float, volume: float = 0.0, spread: float = 0.0, bid: float = 0.0, ask: float = 0.0) -> None: p = safe_float(price, self.close) self.high = max(self.high, p) self.low = min(self.low, p) self.close = p self.end_ts = max(self.end_ts, ts) if volume: self.volume += max(0.0, volume) if spread: self.spread = spread if bid: self.bid = bid if ask: self.ask = ask def finalize(self, end_ts: Optional[float] = None) -> None: if end_ts is not None: self.end_ts = end_ts self.closed = True def to_engine_candle(self, engine_candle_cls, timestamp_override: Optional[float] = None): ts = timestamp_override if timestamp_override is not None else self.end_ts return engine_candle_cls( timestamp=ts, open=self.open, high=self.high, low=self.low, close=self.close, volume=self.volume, spread=self.spread, bid=self.bid, ask=self.ask, source_id=self.source_id, session_label="unknown", is_closed=self.closed, ) def to_dict(self) -> Dict[str, Any]: return { "timeframe": self.timeframe, "timeframe_label": tf_label(self.timeframe), "start_ts": self.start_ts, "end_ts": self.end_ts, "open": self.open, "high": self.high, "low": self.low, "close": self.close, "volume": self.volume, "spread": self.spread, "bid": self.bid, "ask": self.ask, "source_id": self.source_id, "closed": self.closed, } class TimeframeAggregator: def __init__(self, timeframes: List[int] = TIMEFRAMES, maxlen: int = 180) -> None: self.timeframes = list(timeframes) self.current: Dict[int, Optional[CandleBar]] = {tf: None for tf in self.timeframes} self.history: Dict[int, Deque[CandleBar]] = {tf: deque(maxlen=maxlen) for tf in self.timeframes} def reset(self) -> None: self.current = {tf: None for tf in self.timeframes} self.history = {tf: deque(maxlen=self.history[tf].maxlen) for tf in self.timeframes} def update_tick(self, price: float, ts: float, volume: float = 0.0, spread: float = 0.0, bid: float = 0.0, ask: float = 0.0, source_id: str = "deriv") -> Dict[int, List[CandleBar]]: finalized: Dict[int, List[CandleBar]] = {tf: [] for tf in self.timeframes} p = safe_float(price) t = safe_float(ts, now_utc()) for tf in self.timeframes: bucket_start = math.floor(t / tf) * tf bucket_end = bucket_start + tf cur = self.current[tf] if cur is None: self.current[tf] = CandleBar( timeframe=tf, start_ts=bucket_start, end_ts=t, open=p, high=p, low=p, close=p, volume=max(0.0, volume), spread=spread, bid=bid, ask=ask, source_id=source_id, closed=False, ) continue if bucket_start > cur.start_ts: cur.finalize(end_ts=min(bucket_start, bucket_end)) self.history[tf].append(cur) finalized[tf].append(cur) self.current[tf] = CandleBar( timeframe=tf, start_ts=bucket_start, end_ts=t, open=p, high=p, low=p, close=p, volume=max(0.0, volume), spread=spread, bid=bid, ask=ask, source_id=source_id, closed=False, ) else: cur.update(p, t, volume=volume, spread=spread, bid=bid, ask=ask) return finalized def force_close_all(self) -> Dict[int, List[CandleBar]]: finalized: Dict[int, List[CandleBar]] = {tf: [] for tf in self.timeframes} for tf, cur in self.current.items(): if cur is not None and not cur.closed: cur.finalize() self.history[tf].append(cur) finalized[tf].append(cur) return finalized def snapshot(self, include_current: bool = True) -> Dict[str, List[Dict[str, Any]]]: out: Dict[str, List[Dict[str, Any]]] = {} for tf in self.timeframes: bars = list(self.history[tf]) if include_current and self.current[tf] is not None: bars = bars + [self.current[tf]] out[str(tf)] = [b.to_dict() for b in bars] return out def latest_candle(self, tf: int) -> Optional[CandleBar]: cur = self.current.get(tf) if cur is not None: return cur hist = self.history.get(tf) if hist: return hist[-1] return None @dataclass class SignalEvent: signal_id: int direction: str generated_at: float confirmed_at: Optional[float] expires_at: Optional[float] expiry_bucket: str lifecycle_state: str confidence: float raw_output: Dict[str, Any] = field(default_factory=dict) active: bool = True stale: bool = False invalidated: bool = False def to_dict(self, now_ts: Optional[float] = None) -> Dict[str, Any]: now_ts = now_ts if now_ts is not None else now_utc() countdown = None age = max(0.0, now_ts - self.generated_at) if self.expires_at is not None: countdown = max(0.0, self.expires_at - now_ts) return { "signal_id": self.signal_id, "direction": self.direction, "generated_at": self.generated_at, "confirmed_at": self.confirmed_at, "expires_at": self.expires_at, "countdown": countdown, "age": age, "expiry_bucket": self.expiry_bucket, "lifecycle_state": self.lifecycle_state, "confidence": self.confidence, "active": self.active, "stale": self.stale, "invalidated": self.invalidated, } class SignalTimeline: def __init__(self) -> None: self.history: Deque[SignalEvent] = deque(maxlen=50) self.active_event_id: Optional[int] = None self._next_id = 1 @staticmethod def _stale_ticks(tf_seconds: float) -> int: tf = tf_seconds if tf_seconds > 0 else 60.0 return max(12, min(40, int(900.0 / tf))) @staticmethod def _cooling_ticks(tf_seconds: float) -> int: tf = tf_seconds if tf_seconds > 0 else 60.0 return max(5, min(15, int(300.0 / tf))) def update(self, output: Dict[str, Any], candle_ts: float, tf_seconds: float, now_ts: Optional[float] = None) -> None: now_ts = now_ts if now_ts is not None else now_utc() state = output.get("lifecycle_state", "idle") direction = output.get("direction", "BUY") confidence = safe_float(output.get("confidence_final", output.get("confidence", 0.0))) expiry_bucket = output.get("expiry_bucket", "short") stale_flag = bool(output.get("stale_signal_flag", False)) cooling = bool(output.get("cooling_flag", False)) active_lifecycle = state in {"forming", "candidate", "confirmed", "cooling"} existing = self._get_active() if active_lifecycle: if existing is None or existing.direction != direction or (existing.lifecycle_state in {"expired", "invalidated"}): generated_at = candle_ts confirmed_at = candle_ts if state == "confirmed" else None expires_at = candle_ts + self._stale_ticks(tf_seconds) * max(1.0, tf_seconds) ev = SignalEvent( signal_id=self._next_id, direction=direction, generated_at=generated_at, confirmed_at=confirmed_at, expires_at=expires_at, expiry_bucket=expiry_bucket, lifecycle_state=state, confidence=confidence, raw_output=output, active=True, stale=stale_flag, invalidated=False, ) self._next_id += 1 self.history.append(ev) self.active_event_id = ev.signal_id else: existing.lifecycle_state = state existing.confidence = confidence existing.raw_output = output existing.expiry_bucket = expiry_bucket existing.stale = stale_flag existing.active = True if state == "confirmed" and existing.confirmed_at is None: existing.confirmed_at = candle_ts if existing.generated_at > candle_ts: existing.generated_at = candle_ts if existing.expires_at is None: existing.expires_at = candle_ts + self._stale_ticks(tf_seconds) * max(1.0, tf_seconds) self.active_event_id = existing.signal_id else: if existing is not None: if state in {"expired", "invalidated"}: existing.lifecycle_state = state existing.active = False existing.invalidated = state == "invalidated" existing.stale = state == "expired" or stale_flag elif cooling: existing.lifecycle_state = "cooling" existing.active = False else: existing.active = False self.active_event_id = None if state in {"idle", "expired", "invalidated"} else self.active_event_id # Expire old events based on clock for ev in self.history: if ev.expires_at is not None and now_ts >= ev.expires_at: ev.active = False if ev.lifecycle_state not in {"expired", "invalidated"}: ev.lifecycle_state = "expired" ev.stale = True def _get_active(self) -> Optional[SignalEvent]: if self.active_event_id is None: return None for ev in reversed(self.history): if ev.signal_id == self.active_event_id: return ev return None def active_signals(self, now_ts: Optional[float] = None) -> List[Dict[str, Any]]: now_ts = now_ts if now_ts is not None else now_utc() active = [] for ev in self.history: if ev.active or (ev.expires_at is not None and now_ts < ev.expires_at and ev.lifecycle_state not in {"expired", "invalidated"}): active.append(ev.to_dict(now_ts=now_ts)) return active def current(self, now_ts: Optional[float] = None) -> Optional[Dict[str, Any]]: now_ts = now_ts if now_ts is not None else now_utc() ev = self._get_active() if ev is None: for candidate in reversed(self.history): if candidate.expires_at is not None and now_ts < candidate.expires_at and candidate.lifecycle_state not in {"expired", "invalidated"}: ev = candidate break return ev.to_dict(now_ts=now_ts) if ev else None def to_dict(self, now_ts: Optional[float] = None) -> Dict[str, Any]: now_ts = now_ts if now_ts is not None else now_utc() return { "current": self.current(now_ts), "active_signals": self.active_signals(now_ts), "history": [ev.to_dict(now_ts=now_ts) for ev in self.history], } class MarketRuntime: def __init__(self, default_symbol: str = "frxEURUSD", base_timeframe: int = 30, debug_mode: bool = True) -> None: module = get_engine_module() self.Engine = module.MAYTHOS self.Candle = module.Candle self.engine = self.Engine(debug_mode=debug_mode) self.debug_mode = debug_mode self.selected_symbol = default_symbol self.base_timeframe = base_timeframe self.selected_display_tf = base_timeframe self.market_type = detect_market_type(default_symbol) self.aggregator = TimeframeAggregator(TIMEFRAMES, maxlen=180) self.signals = SignalTimeline() self.connected = False self.live_mode = False self.demo_mode = False self.ws_status = "idle" self.ws_url = os.getenv("DERIV_WS_URL", "wss://api.derivws.com/trading/v1/options/ws/public") self.app_id = os.getenv("DERIV_APP_ID", "").strip() self.source_id = "deriv_public" self.last_error = "" self.last_error_at: Optional[float] = None self.reconnects = 0 self.connection_attempts = 0 self.last_ping_at: Optional[float] = None self.last_pong_at: Optional[float] = None self.last_server_time: Optional[float] = None self.last_server_sync_at: Optional[float] = None self.last_tick: Optional[Dict[str, Any]] = None self.last_snapshot: Dict[str, Any] = {} self.last_engine_output: Dict[str, Any] = {} self.last_debug_trace: Optional[Dict[str, Any]] = None self.validation_errors: List[str] = [] self.health_snapshot: Dict[str, Any] = {} self.logs: Deque[Dict[str, Any]] = deque(maxlen=200) self.tick_stream: Deque[Dict[str, Any]] = deque(maxlen=240) self.base_candle_closes: Deque[Dict[str, Any]] = deque(maxlen=240) self.latest_price: Optional[float] = None self.latest_tick_ts: Optional[float] = None self.latest_candle_ts: Optional[float] = None self.latest_tf_seconds: float = float(base_timeframe) self.tick_counter = 0 self.candle_counter = 0 self._lock = asyncio.Lock() self._stop = asyncio.Event() self._restart = asyncio.Event() self._stream_task: Optional[asyncio.Task] = None self._demo_rng = random.Random(7) self._demo_price = 1.0 self._demo_anchor = 1.0 self._symbol_options = [ "frxEURUSD", "frxGBPUSD", "frxUSDJPY", "cryBTCUSD", "cryETHUSD", ] async def start(self) -> None: if self._stream_task is None or self._stream_task.done(): self._stop.clear() self._restart.clear() self._stream_task = asyncio.create_task(self._stream_loop()) async def stop(self) -> None: self._stop.set() self._restart.set() if self._stream_task is not None: self._stream_task.cancel() try: await self._stream_task except Exception: pass self._stream_task = None async def set_symbol(self, symbol: str) -> None: symbol = (symbol or "").strip() if not symbol: return async with self._lock: self.selected_symbol = symbol self.market_type = detect_market_type(symbol) self.source_id = f"deriv_{self.market_type or 'unknown'}" self._restart.set() async def set_display_timeframe(self, tf_seconds: int) -> None: tf_seconds = int(tf_seconds) if tf_seconds in TIMEFRAMES: async with self._lock: self.selected_display_tf = tf_seconds def symbol_options(self) -> List[str]: return list(self._symbol_options) async def _stream_loop(self) -> None: while not self._stop.is_set(): symbol = self.selected_symbol try: self.connection_attempts += 1 await self._run_live_stream(symbol) except asyncio.CancelledError: raise except Exception as exc: self._set_error(f"{type(exc).__name__}: {exc}") self.reconnects += 1 await self._run_demo_stream(symbol) finally: if self._restart.is_set(): self._restart.clear() if self._stop.is_set(): break await asyncio.sleep(min(5.0, 1.0 + self.reconnects * 0.5)) async def _run_live_stream(self, symbol: str) -> None: url = self._build_ws_url() self.ws_status = "connecting" self.connected = False self.live_mode = False self.demo_mode = False async with websockets.connect(url, ping_interval=None, close_timeout=5, open_timeout=10, max_queue=256) as ws: self.ws_status = "connected" self.connected = True self.live_mode = True self.demo_mode = False self.last_error = "" self.last_error_at = None # bootstrap system time / server sync await self._send(ws, {"time": 1, "req_id": 1}) await self._send(ws, {"ping": 1, "req_id": 2}) # historical seed + live subscription await self._send(ws, { "ticks_history": symbol, "end": "latest", "style": "ticks", "count": 1000, "subscribe": 0, "req_id": 3, }) await self._send(ws, { "ticks": symbol, "subscribe": 1, "req_id": 4, }) heartbeat = asyncio.create_task(self._heartbeat(ws)) received_any_message = False received_tick = False try: while not self._stop.is_set() and not self._restart.is_set(): timeout = 15.0 if not received_any_message else 60.0 try: raw = await asyncio.wait_for(ws.recv(), timeout=timeout) except asyncio.TimeoutError: if not received_tick: raise TimeoutError( f"No market data received for {symbol} from Deriv public stream" ) # Keep the connection alive and keep waiting for ticks. await self._send(ws, {"ping": 1, "req_id": int(now_utc()) % 1_000_000}) self.last_ping_at = now_utc() continue received_any_message = True msg = json.loads(raw) before_tick_count = self.tick_counter await self._handle_message(msg, symbol) if msg.get("msg_type") in {"tick", "history"}: received_tick = received_tick or (self.tick_counter > before_tick_count) finally: heartbeat.cancel() try: await heartbeat except BaseException: pass self.connected = False async def _heartbeat(self, ws) -> None: counter = 100 while not self._stop.is_set() and not self._restart.is_set(): await asyncio.sleep(30) try: await self._send(ws, {"ping": 1, "req_id": counter}) self.last_ping_at = now_utc() counter += 1 except Exception as exc: self._set_error(f"heartbeat:{type(exc).__name__}: {exc}") return async def _run_demo_stream(self, symbol: str) -> None: self.ws_status = "demo" self.connected = False self.live_mode = False self.demo_mode = True if self.latest_price is not None and self.latest_price > 0: self._demo_price = self.latest_price else: self._demo_price = 1.0 if detect_market_type(symbol) == "forex" else 30000.0 self._demo_anchor = self._demo_price start = now_utc() while not self._stop.is_set() and not self._restart.is_set(): await self._generate_demo_tick(symbol) await asyncio.sleep(1.0) # Demo stays alive until live data becomes available or the app stops. if now_utc() - start > 45: start = now_utc() self.ws_status = "idle" async def _send(self, ws, payload: Dict[str, Any]) -> None: if self.app_id and "app_id" not in payload and self.ws_url.endswith("/public"): # No hardcoded secrets. Optional app_id is only appended when supplied. pass await ws.send(json.dumps(payload)) def _build_ws_url(self) -> str: url = self.ws_url.strip() app_id = self.app_id if app_id and "app_id=" not in url: joiner = "&" if "?" in url else "?" url = f"{url}{joiner}app_id={app_id}" return url async def _handle_message(self, msg: Dict[str, Any], symbol: str) -> None: msg_type = msg.get("msg_type") if msg_type == "time": server_time = msg.get("time") if server_time is not None: self.last_server_time = safe_float(server_time) self.last_server_sync_at = now_utc() return if msg_type == "ping": self.last_pong_at = now_utc() return if msg_type == "history": history = msg.get("history", {}) prices = history.get("prices") or [] times = history.get("times") or [] await self._seed_history(times, prices, symbol) return if msg_type == "tick": tick = msg.get("tick", {}) await self._handle_tick_message(tick, symbol) return if msg_type == "active_symbols": return if "error" in msg: self._set_error(str(msg["error"])) return async def _seed_history(self, times: List[Any], prices: List[Any], symbol: str) -> None: for ts, price in zip(times, prices): await self._process_tick( tick_ts=safe_float(ts), price=safe_float(price), symbol=symbol, bid=0.0, ask=0.0, volume=0.0, source="history", is_history_seed=True, ) async def _handle_tick_message(self, tick: Dict[str, Any], symbol: str) -> None: tick_ts = safe_float(tick.get("epoch"), now_utc()) quote = tick.get("quote", tick.get("price", tick.get("last_price", 0.0))) bid = safe_float(tick.get("bid"), 0.0) ask = safe_float(tick.get("ask"), 0.0) volume = safe_float(tick.get("volume"), 0.0) await self._process_tick( tick_ts=tick_ts, price=safe_float(quote), symbol=symbol, bid=bid, ask=ask, volume=volume, source="live", is_history_seed=False, ) async def _generate_demo_tick(self, symbol: str) -> None: market_type = detect_market_type(symbol) drift = 0.00005 if market_type == "forex" else 1.5 vol = 0.0005 if market_type == "forex" else 40.0 shock = self._demo_rng.gauss(0.0, vol) if market_type == "forex": self._demo_price = max(0.0001, self._demo_price + shock + drift * (1 if self._demo_rng.random() > 0.5 else -0.5)) else: self._demo_price = max(1.0, self._demo_price + shock + drift * (1 if self._demo_rng.random() > 0.5 else -0.5)) bid = self._demo_price - (0.0001 if market_type == "forex" else 0.5) ask = self._demo_price + (0.0001 if market_type == "forex" else 0.5) await self._process_tick( tick_ts=now_utc(), price=self._demo_price, symbol=symbol, bid=bid, ask=ask, volume=0.0, source="demo", is_history_seed=False, ) async def _process_tick(self, tick_ts: float, price: float, symbol: str, bid: float, ask: float, volume: float, source: str, is_history_seed: bool) -> None: async with self._lock: self.tick_counter += 1 self.latest_price = price self.latest_tick_ts = tick_ts self.market_type = detect_market_type(symbol) spread = abs(ask - bid) if (ask and bid and ask > bid) else 0.0 self.last_tick = { "ts": tick_ts, "price": price, "bid": bid, "ask": ask, "spread": spread, "volume": volume, "symbol": symbol, "source": source, "is_history_seed": is_history_seed, } self.tick_stream.append(self.last_tick) finalized = self.aggregator.update_tick( price=price, ts=tick_ts, volume=volume, spread=spread, bid=bid, ask=ask, source_id=source, ) if finalized[self.base_timeframe]: for candle in finalized[self.base_timeframe]: await self._run_engine_on_candle(candle) # Update latest snapshot frequently even between 30s closes self.last_snapshot = self._build_snapshot_unlocked() async def _run_engine_on_candle(self, candle: CandleBar) -> None: module = get_engine_module() engine_candle = candle.to_engine_candle(self.Candle, timestamp_override=candle.end_ts) receive_time = now_utc() output = self.engine.tick(engine_candle, receive_time=receive_time, debug=True) self.last_engine_output = output self.last_debug_trace = output.get("debug_trace") self.validation_errors = self.engine.validate(output) self.health_snapshot = self.engine.engine_health() self.candle_counter += 1 self.latest_candle_ts = candle.end_ts self.latest_tf_seconds = float(self.base_timeframe) self.base_candle_closes.append({ "ts": candle.end_ts, "timeframe": candle.timeframe, "open": candle.open, "high": candle.high, "low": candle.low, "close": candle.close, "closed": candle.closed, "direction": output.get("direction"), "confidence": output.get("confidence"), "lifecycle_state": output.get("lifecycle_state"), }) self.logs.append({ "ts": candle.end_ts, "iso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(candle.end_ts)), "price": candle.close, "direction": output.get("direction"), "confidence": output.get("confidence"), "execution_suitability": output.get("execution_suitability"), "market_state": output.get("market_state"), "lifecycle_state": output.get("lifecycle_state"), "reason_summary": output.get("reason_summary"), "signal_freshness": output.get("signal_freshness"), }) self.signals.update(output, candle.end_ts, tf_seconds=self.base_timeframe, now_ts=now_utc()) def _set_error(self, message: str) -> None: self.last_error = message[:400] self.last_error_at = now_utc() self.ws_status = "error" def _build_snapshot_unlocked(self) -> Dict[str, Any]: now_ts = now_utc() base_output = self.last_engine_output or {} current_signal = self.signals.current(now_ts) active_signals = self.signals.active_signals(now_ts) tf_snapshot = self.aggregator.snapshot(include_current=True) selected = self.selected_display_tf latest_selected = self.aggregator.latest_candle(selected) latest_base = self.aggregator.latest_candle(self.base_timeframe) signal_generated_at = None signal_expires_at = None signal_countdown = None signal_age = None if current_signal: signal_generated_at = current_signal.get("generated_at") signal_expires_at = current_signal.get("expires_at") if signal_expires_at is not None: signal_countdown = max(0.0, signal_expires_at - now_ts) if signal_generated_at is not None: signal_age = max(0.0, now_ts - signal_generated_at) timeline = { "current": current_signal, "active_signals": active_signals, "history": self.signals.to_dict(now_ts)["history"], "signal_generated_at": signal_generated_at, "signal_expires_at": signal_expires_at, "signal_countdown": signal_countdown, "signal_age": signal_age, "lifecycle_state": current_signal.get("lifecycle_state") if current_signal else base_output.get("lifecycle_state"), } engine_block = { "tick_count": self.engine.tick_count, "is_warm": self.engine.is_warm, "debug_log_size": len(self.engine.debug_log), "raw_output": base_output, "validation_errors": list(self.validation_errors), "health": dict(self.health_snapshot or self.engine.engine_health()), "debug_trace": self.last_debug_trace, } connection_block = { "selected_symbol": self.selected_symbol, "market_type": self.market_type, "selected_display_tf": selected, "selected_display_tf_label": tf_label(selected), "base_timeframe": self.base_timeframe, "base_timeframe_label": tf_label(self.base_timeframe), "status": self.ws_status, "connected": self.connected, "live_mode": self.live_mode, "demo_mode": self.demo_mode, "connection_attempts": self.connection_attempts, "reconnects": self.reconnects, "last_error": self.last_error, "last_error_at": self.last_error_at, "last_ping_at": self.last_ping_at, "last_pong_at": self.last_pong_at, "last_server_time": self.last_server_time, "last_server_sync_at": self.last_server_sync_at, } clock = { "utc_now": now_ts, "latest_tick_ts": self.latest_tick_ts, "latest_candle_ts": self.latest_candle_ts, "latest_tick_age_sec": None if self.latest_tick_ts is None else max(0.0, now_ts - self.latest_tick_ts), "latest_candle_age_sec": None if self.latest_candle_ts is None else max(0.0, now_ts - self.latest_candle_ts), "server_time": self.last_server_time, "server_sync_offset": None if (self.last_server_time is None or self.last_server_sync_at is None) else self.last_server_time - self.last_server_sync_at, } charts = { "selected_timeframe": selected, "selected_timeframe_label": tf_label(selected), "timeframes": tf_snapshot, "latest_selected_candle": None if latest_selected is None else latest_selected.to_dict(), "latest_base_candle": None if latest_base is None else latest_base.to_dict(), "price_stream": list(self.tick_stream), } snapshot = { "clock": clock, "connection": connection_block, "engine": engine_block, "timeline": timeline, "charts": charts, "logs": list(self.logs), "symbol_options": self.symbol_options(), "base_ref": { "timeframe_seconds": self.base_timeframe, "timeframe_label": tf_label(self.base_timeframe), "reference_source": "Deriv tick epoch", }, "source": { "market_type": self.market_type, "engine_file": "maythos_patched.py", }, } return snapshot async def snapshot(self) -> Dict[str, Any]: async with self._lock: return self._build_snapshot_unlocked() async def health(self) -> Dict[str, Any]: snap = await self.snapshot() engine_health = snap["engine"]["health"] return { "ok": snap["connection"]["status"] in {"connected", "demo", "idle", "error"}, "status": snap["connection"]["status"], "selected_symbol": snap["connection"]["selected_symbol"], "market_type": snap["connection"]["market_type"], "live_mode": snap["connection"]["live_mode"], "demo_mode": snap["connection"]["demo_mode"], "tick_count": snap["engine"]["tick_count"], "is_warm": snap["engine"]["is_warm"], "validation_errors": snap["engine"]["validation_errors"], "engine_health": engine_health, "last_error": snap["connection"]["last_error"], "last_tick_age_sec": snap["clock"]["latest_tick_age_sec"], "latest_candle_age_sec": snap["clock"]["latest_candle_age_sec"], "signal": snap["timeline"]["current"], }