runtime.py

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"],
        }