src/telegram_bot/controller.py

"""Bridge between the async Telegram bot and the synchronous polling core.

The platform loop (build_runtime/run_once) runs in a daemon thread; the bot
reads state snapshots and flips thread-safe flags (pause, mute, interval).
Consistent with the rest of the POC, subscriber chats and pause state are
in-memory only and reset on restart.
"""

import logging
import threading
import time
from collections import deque
from typing import Dict, List, Optional

from src.core.config import POLL_INTERVAL_SEC, RISK_ENVELOPE_POLICY
from src.core.main import Runtime, build_runtime, run_once
from src.core.topics import TOPICS


logger = logging.getLogger(__name__)

MIN_POLL_INTERVAL_SEC = 15
MAX_POLL_INTERVAL_SEC = 3600


class PlatformController:
    def __init__(self) -> None:
        self.runtime: Optional[Runtime] = None
        self.poll_interval = POLL_INTERVAL_SEC
        self.started_at = time.time()
        self.startup_error: Optional[str] = None
        self.last_cycle_ts: Optional[float] = None
        self.last_cycle_duration: Optional[float] = None

        self.recent_risk_events: deque = deque(maxlen=50)
        self.recent_regime_changes: deque = deque(maxlen=20)

        self._subscribers: set[str] = set()
        self._subscribers_lock = threading.Lock()

        self._paused = threading.Event()
        self._stop = threading.Event()
        self._thread: Optional[threading.Thread] = None

    # -- lifecycle ---------------------------------------------------------

    def start(self) -> None:
        if self._thread is not None:
            return
        self._thread = threading.Thread(target=self._run_loop, name="platform-loop", daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()

    def _run_loop(self) -> None:
        try:
            # Universe load hits GitHub/Wikipedia/Binance; may take a while
            runtime = build_runtime()
        except Exception as exc:
            logger.exception("platform runtime failed to build")
            self.startup_error = str(exc)
            return

        runtime.pubsub.subscribe(TOPICS["risk_event"], self.recent_risk_events.append)
        runtime.pubsub.subscribe(TOPICS["risk_regime_change"], self.recent_regime_changes.append)
        runtime.notifier.chat_ids_provider = self.subscribers
        self.runtime = runtime
        logger.info("platform runtime ready, entering poll loop")

        while not self._stop.is_set():
            if not self._paused.is_set():
                started = time.time()
                try:
                    run_once(runtime)
                except Exception:
                    logger.exception("poll cycle failed")
                self.last_cycle_ts = time.time()
                self.last_cycle_duration = self.last_cycle_ts - started
            # Sleep in small steps so pause/interval changes apply quickly
            deadline = time.time() + self.poll_interval
            while time.time() < deadline and not self._stop.is_set():
                time.sleep(1)

    # -- control commands --------------------------------------------------

    def pause(self) -> bool:
        """Returns False if already paused."""
        if self._paused.is_set():
            return False
        self._paused.set()
        return True

    def resume(self) -> bool:
        if not self._paused.is_set():
            return False
        self._paused.clear()
        return True

    @property
    def paused(self) -> bool:
        return self._paused.is_set()

    def set_poll_interval(self, seconds: int) -> int:
        seconds = max(MIN_POLL_INTERVAL_SEC, min(MAX_POLL_INTERVAL_SEC, seconds))
        self.poll_interval = seconds
        return seconds

    def set_muted(self, muted: bool) -> bool:
        """Returns False if the runtime is not ready yet."""
        if self.runtime is None:
            return False
        self.runtime.notifier.muted = muted
        return True

    @property
    def muted(self) -> bool:
        return bool(self.runtime and self.runtime.notifier.muted)

    # -- alert subscriptions ----------------------------------------------

    def subscribe(self, chat_id: int | str) -> bool:
        with self._subscribers_lock:
            before = len(self._subscribers)
            self._subscribers.add(str(chat_id))
            return len(self._subscribers) > before

    def unsubscribe(self, chat_id: int | str) -> bool:
        with self._subscribers_lock:
            try:
                self._subscribers.remove(str(chat_id))
                return True
            except KeyError:
                return False

    def subscribers(self) -> List[str]:
        with self._subscribers_lock:
            return list(self._subscribers)

    # -- state snapshots ---------------------------------------------------

    def status(self) -> Dict:
        if self.startup_error:
            state = "failed"
        elif self.runtime is None:
            state = "starting"
        elif self.paused:
            state = "paused"
        else:
            state = "running"

        info: Dict = {
            "state": state,
            "startup_error": self.startup_error,
            "uptime_sec": int(time.time() - self.started_at),
            "poll_interval_sec": self.poll_interval,
            "muted": self.muted,
            "subscribers": len(self.subscribers()),
        }
        if self.runtime is not None:
            info.update(
                {
                    "cycle": self.runtime.cycle,
                    "regime": self.runtime.risk.regime,
                    "universe": {venue: len(tickers) for venue, tickers in self.runtime.universe.items()},
                    "last_cycle_duration_sec": (
                        round(self.last_cycle_duration, 1) if self.last_cycle_duration else None
                    ),
                }
            )
        return info

    def regime_info(self) -> Optional[Dict]:
        if self.runtime is None:
            return None
        regime = self.runtime.risk.regime
        envelope = self.runtime.envelope_store.get()
        policy = RISK_ENVELOPE_POLICY.get(regime, {})
        return {"regime": regime, "envelope": envelope, "policy": policy}

    def universe(self) -> Optional[Dict[str, List[str]]]:
        if self.runtime is None:
            return None
        return self.runtime.universe

    def telemetry(self) -> Optional[Dict[str, int]]:
        if self.runtime is None:
            return None
        return self.runtime.telemetry.snapshot()

    def events(self, limit: int = 10) -> List[Dict]:
        return list(self.recent_risk_events)[-limit:]

    def regime_changes(self, limit: int = 5) -> List[Dict]:
        return list(self.recent_regime_changes)[-limit:]