[codex] Consolidate backend live snapshot routes

apps/api/gammascope_api/ingestion/collector_state.py

@@ -2,6 +2,7 @@
 
 from copy import deepcopy
 from typing import Any
+from uuid import uuid4
 
 from gammascope_api.contracts.generated.collector_events import (
     CollectorEvents,
@@ -17,6 +18,8 @@ def __init__(self) -> None:
         self.clear()
 
     def clear(self) -> None:
+        self._state_epoch = uuid4().hex
+        self._revision = 0
         self._health_events: dict[str, dict[str, Any]] = {}
         self._contracts: dict[str, dict[str, Any]] = {}
         self._underlying_ticks: dict[str, dict[str, Any]] = {}
@@ -38,10 +41,13 @@ def ingest(self, event: CollectorEvents) -> str:
         elif isinstance(payload, OptionTick):
             self._option_ticks[payload.contract_id] = event_dict
 
+        self._revision += 1
         return event_type
 
     def summary(self) -> dict[str, Any]:
         return {
+            "state_epoch": self._state_epoch,
+            "revision": self._revision,
             "health_events_count": len(self._health_events),
             "contracts_count": len(self._contracts),
             "underlying_ticks_count": len(self._underlying_ticks),
@@ -50,10 +56,13 @@ def summary(self) -> dict[str, Any]:
             "latest_health": self.latest_health(),
         }
 
+    def revision(self) -> int:
+        return self._revision
+
     def latest_health(self) -> dict[str, Any] | None:
         if not self._health_events:
             return None
-        return max(self._health_events.values(), key=lambda event: str(event["received_time"]))
+        return deepcopy(max(self._health_events.values(), key=lambda event: str(event["received_time"])))
 
     def latest_underlying_tick(self, session_id: str | None = None) -> dict[str, Any] | None:
         underlying_ticks = self._underlying_ticks
@@ -65,19 +74,21 @@ def latest_underlying_tick(self, session_id: str | None = None) -> dict[str, Any
             }
         if not underlying_ticks:
             return None
-        return max(underlying_ticks.values(), key=lambda event: str(event["event_time"]))
+        return deepcopy(max(underlying_ticks.values(), key=lambda event: str(event["event_time"])))
 
     def contracts(self) -> list[dict[str, Any]]:
-        return list(self._contracts.values())
+        return deepcopy(list(self._contracts.values()))
 
     def option_ticks(self) -> dict[str, dict[str, Any]]:
-        return dict(self._option_ticks)
+        return deepcopy(self._option_ticks)
 
     def last_event_time(self) -> str | None:
         return self._last_event_time
 
     def snapshot(self) -> dict[str, Any]:
         return {
+            "state_epoch": self._state_epoch,
+            "revision": self._revision,
             "health_events": deepcopy(self._health_events),
             "contracts": deepcopy(self._contracts),
             "underlying_ticks": deepcopy(self._underlying_ticks),
@@ -93,6 +104,8 @@ def from_snapshot(cls, snapshot: dict[str, Any]) -> CollectorState:
         state._underlying_ticks = deepcopy(snapshot.get("underlying_ticks", {}))
         state._option_ticks = deepcopy(snapshot.get("option_ticks", {}))
         state._last_event_time = snapshot.get("last_event_time")
+        state._state_epoch = str(snapshot.get("state_epoch") or state._state_epoch)
+        state._revision = int(snapshot.get("revision") or 0)
         return state
 
 

apps/api/gammascope_api/ingestion/live_snapshot_service.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from dataclasses import dataclass
+from functools import lru_cache
+from typing import Any, Callable, Literal
+
+from gammascope_api.ingestion.collector_state import CollectorState
+from gammascope_api.ingestion.latest_state_cache import cached_or_memory_collector_state
+from gammascope_api.ingestion.live_snapshot import _freshness_ms, build_live_snapshot, build_spx_dashboard_live_snapshot
+
+HeatmapSymbol = Literal["SPX", "SPY", "QQQ", "NDX", "IWM"]
+
+MOOMOO_LIVE_REPLAY_SESSION_IDS: dict[HeatmapSymbol, str] = {
+    "SPX": "moomoo-spx-0dte-live",
+    "SPY": "moomoo-spy-0dte-live",
+    "QQQ": "moomoo-qqq-0dte-live",
+    "NDX": "moomoo-ndx-0dte-live",
+    "IWM": "moomoo-iwm-0dte-live",
+}
+
+_DASHBOARD_KEY = "__dashboard__"
+StateIdentity = tuple[str | None, int, str | None, int, int, int, int]
+
+
+@dataclass(frozen=True)
+class _CachedSnapshot:
+    state_identity: StateIdentity
+    snapshot: dict[str, Any] | None
+
+
+class LiveSnapshotService:
+    def __init__(self, state_provider: Callable[[], CollectorState] = cached_or_memory_collector_state) -> None:
+        self._state_provider = state_provider
+        self._cache: dict[str, _CachedSnapshot] = {}
+
+    def dashboard_snapshot(self) -> dict[str, Any] | None:
+        return self._cached_snapshot(
+            _DASHBOARD_KEY,
+            lambda state: build_spx_dashboard_live_snapshot(state),
+        )
+
+    def session_snapshot(self, session_id: str) -> dict[str, Any] | None:
+        return self._cached_snapshot(
+            session_id,
+            lambda state: build_live_snapshot(state, session_id=session_id),
+        )
+
+    def symbol_snapshot(self, symbol: HeatmapSymbol) -> dict[str, Any] | None:
+        return self.session_snapshot(MOOMOO_LIVE_REPLAY_SESSION_IDS[symbol])
+
+    def _cached_snapshot(
+        self,
+        cache_key: str,
+        builder: Callable[[CollectorState], dict[str, Any] | None],
+    ) -> dict[str, Any] | None:
+        state = self._state_provider()
+        state_identity = _state_identity(state)
+        cached = self._cache.get(cache_key)
+
+        if cached is None or cached.state_identity != state_identity:
+            cached = _CachedSnapshot(state_identity=state_identity, snapshot=builder(state))
+            self._cache[cache_key] = cached
+
+        return _snapshot_response(cached.snapshot)
+
+
+_service_override: LiveSnapshotService | None = None
+
+
+def get_live_snapshot_service() -> LiveSnapshotService:
+    if _service_override is not None:
+        return _service_override
+    return _default_live_snapshot_service()
+
+
+def set_live_snapshot_service_override(service: LiveSnapshotService) -> None:
+    global _service_override
+    _service_override = service
+
+
+def reset_live_snapshot_service_override() -> None:
+    global _service_override
+    _service_override = None
+    _default_live_snapshot_service.cache_clear()
+
+
+@lru_cache(maxsize=1)
+def _default_live_snapshot_service() -> LiveSnapshotService:
+    return LiveSnapshotService()
+
+
+def _state_identity(state: CollectorState) -> StateIdentity:
+    summary = state.summary()
+    return (
+        _string_or_none(summary.get("state_epoch")),
+        _int_value(summary.get("revision")),
+        _string_or_none(summary.get("last_event_time")),
+        _int_value(summary.get("health_events_count")),
+        _int_value(summary.get("contracts_count")),
+        _int_value(summary.get("underlying_ticks_count")),
+        _int_value(summary.get("option_ticks_count")),
+    )
+
+
+def _snapshot_response(snapshot: dict[str, Any] | None) -> dict[str, Any] | None:
+    if snapshot is None:
+        return None
+
+    response = deepcopy(snapshot)
+    snapshot_time = response.get("snapshot_time")
+    if isinstance(snapshot_time, str):
+        try:
+            response["freshness_ms"] = _freshness_ms(snapshot_time)
+        except ValueError:
+            pass
+    return response
+
+
+def _int_value(value: Any) -> int:
+    try:
+        return int(value or 0)
+    except (TypeError, ValueError):
+        return 0
+
+
+def _string_or_none(value: Any) -> str | None:
+    if value is None:
+        return None
+    return str(value)

apps/api/gammascope_api/routes/experimental.py

@@ -6,8 +6,7 @@
 from gammascope_api.contracts.generated.experimental_analytics import ExperimentalAnalytics
 from gammascope_api.experimental.service import build_experimental_payload, validate_experimental_payload
 from gammascope_api.fixtures import load_json_fixture
-from gammascope_api.ingestion.latest_state_cache import cached_or_memory_collector_state
-from gammascope_api.ingestion.live_snapshot import build_spx_dashboard_live_snapshot
+from gammascope_api.ingestion.live_snapshot_service import get_live_snapshot_service
 from gammascope_api.routes import replay as replay_routes
 
 
@@ -17,7 +16,7 @@
 @router.get("/api/spx/0dte/experimental/latest", response_model=ExperimentalAnalytics)
 def get_latest_experimental(x_gammascope_admin_token: str | None = Header(default=None)) -> dict:
     if can_read_live_state(x_gammascope_admin_token):
-        live_snapshot = build_spx_dashboard_live_snapshot(cached_or_memory_collector_state())
+        live_snapshot = get_live_snapshot_service().dashboard_snapshot()
         if live_snapshot is not None:
             return build_experimental_payload(live_snapshot, "latest")
     return validate_experimental_payload(load_json_fixture("experimental-analytics.seed.json"))

apps/api/gammascope_api/routes/heatmap.py

@@ -9,21 +9,16 @@
 from gammascope_api.fixtures import load_json_fixture
 from gammascope_api.heatmap.dependencies import get_heatmap_repository
 from gammascope_api.heatmap.service import build_heatmap_payload
-from gammascope_api.ingestion.latest_state_cache import cached_or_memory_collector_state
-from gammascope_api.ingestion.live_snapshot import build_live_snapshot
+from gammascope_api.ingestion.live_snapshot_service import (
+    MOOMOO_LIVE_REPLAY_SESSION_IDS,
+    HeatmapSymbol,
+    get_live_snapshot_service,
+)
 from gammascope_api.replay.dependencies import get_replay_repository
 
 
 router = APIRouter()
 HeatmapMetric = Literal["gex", "vex"]
-HeatmapSymbol = Literal["SPX", "SPY", "QQQ", "NDX", "IWM"]
-MOOMOO_LIVE_REPLAY_SESSION_IDS = {
-    "SPX": "moomoo-spx-0dte-live",
-    "SPY": "moomoo-spy-0dte-live",
-    "QQQ": "moomoo-qqq-0dte-live",
-    "NDX": "moomoo-ndx-0dte-live",
-    "IWM": "moomoo-iwm-0dte-live",
-}
 
 
 @router.get("/api/spx/0dte/heatmap/latest")
@@ -33,10 +28,7 @@ def get_latest_heatmap(
     x_gammascope_admin_token: str | None = Header(default=None),
 ) -> dict:
     if can_read_live_state(x_gammascope_admin_token):
-        live_snapshot = build_live_snapshot(
-            cached_or_memory_collector_state(),
-            session_id=MOOMOO_LIVE_REPLAY_SESSION_IDS[symbol],
-        )
+        live_snapshot = get_live_snapshot_service().symbol_snapshot(symbol)
         if live_snapshot is not None:
             return build_heatmap_payload(live_snapshot, metric, get_heatmap_repository())
         replay_snapshot = _latest_moomoo_live_replay_snapshot(symbol)

apps/api/gammascope_api/routes/scenario.py

@@ -5,8 +5,7 @@
 from gammascope_api.auth import can_read_live_state
 from gammascope_api.analytics.scenario import create_scenario_snapshot
 from gammascope_api.fixtures import load_json_fixture
-from gammascope_api.ingestion.latest_state_cache import cached_or_memory_collector_state
-from gammascope_api.ingestion.live_snapshot import build_spx_dashboard_live_snapshot
+from gammascope_api.ingestion.live_snapshot_service import get_live_snapshot_service
 
 
 router = APIRouter()
@@ -18,7 +17,7 @@ def create_scenario(
     x_gammascope_admin_token: str | None = Header(default=None),
 ) -> dict:
     if can_read_live_state(x_gammascope_admin_token):
-        live_snapshot = build_spx_dashboard_live_snapshot(cached_or_memory_collector_state())
+        live_snapshot = get_live_snapshot_service().dashboard_snapshot()
         if live_snapshot is not None and live_snapshot["session_id"] == payload.get("session_id"):
             return create_scenario_snapshot(live_snapshot, payload)
 

apps/api/gammascope_api/routes/snapshot.py

@@ -2,8 +2,7 @@
 
 from gammascope_api.auth import can_read_live_state
 from gammascope_api.fixtures import load_json_fixture
-from gammascope_api.ingestion.latest_state_cache import cached_or_memory_collector_state
-from gammascope_api.ingestion.live_snapshot import build_spx_dashboard_live_snapshot
+from gammascope_api.ingestion.live_snapshot_service import get_live_snapshot_service
 
 
 router = APIRouter()
@@ -12,7 +11,7 @@
 @router.get("/api/spx/0dte/snapshot/latest")
 def get_latest_snapshot(x_gammascope_admin_token: str | None = Header(default=None)) -> dict:
     if can_read_live_state(x_gammascope_admin_token):
-        live_snapshot = build_spx_dashboard_live_snapshot(cached_or_memory_collector_state())
+        live_snapshot = get_live_snapshot_service().dashboard_snapshot()
         if live_snapshot is not None:
             return live_snapshot
     return load_json_fixture("analytics-snapshot.seed.json")

apps/api/gammascope_api/routes/stream.py

@@ -5,8 +5,7 @@
 
 from gammascope_api.auth import is_valid_admin_token, private_mode_enabled, websocket_admin_token
 from gammascope_api.fixtures import load_json_fixture
-from gammascope_api.ingestion.latest_state_cache import cached_or_memory_collector_state
-from gammascope_api.ingestion.live_snapshot import build_spx_dashboard_live_snapshot
+from gammascope_api.ingestion.live_snapshot_service import get_live_snapshot_service
 from gammascope_api.routes.replay import replay_stream_snapshots, seed_replay_snapshots
 
 
@@ -66,7 +65,7 @@ async def stream_spx_0dte_replay(
 
 
 def _current_snapshot() -> dict[str, Any]:
-    live_snapshot = build_spx_dashboard_live_snapshot(cached_or_memory_collector_state())
+    live_snapshot = get_live_snapshot_service().dashboard_snapshot()
     if live_snapshot is not None:
         return live_snapshot
     return load_json_fixture("analytics-snapshot.seed.json")

apps/api/tests/test_contract_endpoints.py

@@ -9,6 +9,7 @@
     set_latest_state_cache_override,
 )
 from gammascope_api.ingestion.live_snapshot import reset_live_snapshot_memory
+from gammascope_api.ingestion.live_snapshot_service import reset_live_snapshot_service_override
 from gammascope_api.main import app
 from gammascope_api.replay.capture import reset_replay_capture_circuit
 from gammascope_api.replay.dependencies import set_replay_repository_override
@@ -26,13 +27,15 @@
 def setup_function() -> None:
     collector_state.clear()
     reset_live_snapshot_memory()
+    reset_live_snapshot_service_override()
     set_latest_state_cache_override(InMemoryLatestStateCache())
     set_replay_repository_override(NullReplayRepository())
     set_saved_view_repository_override(InMemorySavedViewRepository())
     reset_replay_capture_circuit()
 
 
 def teardown_function() -> None:
+    reset_live_snapshot_service_override()
     reset_latest_state_cache_override()
     reset_saved_view_repository_override()
 

apps/api/tests/test_experimental_routes.py

@@ -12,6 +12,7 @@
     set_latest_state_cache_override,
 )
 from gammascope_api.ingestion.live_snapshot import reset_live_snapshot_memory
+from gammascope_api.ingestion.live_snapshot_service import reset_live_snapshot_service_override
 from gammascope_api.main import app
 from gammascope_api.routes import experimental as experimental_routes
 
@@ -22,10 +23,12 @@
 def setup_function() -> None:
     collector_state.clear()
     reset_live_snapshot_memory()
+    reset_live_snapshot_service_override()
     set_latest_state_cache_override(InMemoryLatestStateCache())
 
 
 def teardown_function() -> None:
+    reset_live_snapshot_service_override()
     reset_latest_state_cache_override()
 
 

apps/api/tests/test_heatmap_route.py

@@ -11,6 +11,7 @@
     set_latest_state_cache_override,
 )
 from gammascope_api.ingestion.live_snapshot import reset_live_snapshot_memory
+from gammascope_api.ingestion.live_snapshot_service import reset_live_snapshot_service_override
 from gammascope_api.main import app
 from gammascope_api.replay.capture import reset_replay_capture_circuit
 from gammascope_api.replay.dependencies import reset_replay_repository_override, set_replay_repository_override
@@ -23,13 +24,15 @@
 def setup_function() -> None:
     collector_state.clear()
     reset_live_snapshot_memory()
+    reset_live_snapshot_service_override()
     set_latest_state_cache_override(InMemoryLatestStateCache())
     set_replay_repository_override(NullReplayRepository())
     set_heatmap_repository_override(InMemoryHeatmapRepository())
     reset_replay_capture_circuit()
 
 
 def teardown_function() -> None:
+    reset_live_snapshot_service_override()
     reset_latest_state_cache_override()
     reset_replay_repository_override()
     reset_heatmap_repository_override()