File size: 6,779 Bytes
820c4b2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 | """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:]
|