From de7e4b61122a28ebf4dd812f8c313042427d39d3 Mon Sep 17 00:00:00 2001 From: Sakura <32687351+zifanzhou1024@users.noreply.github.com> Date: Thu, 30 Apr 2026 02:25:59 -0700 Subject: [PATCH 1/2] Consolidate backend live snapshot routes --- .../ingestion/collector_state.py | 21 +- .../ingestion/live_snapshot_service.py | 130 ++++ .../api/gammascope_api/routes/experimental.py | 5 +- apps/api/gammascope_api/routes/heatmap.py | 20 +- apps/api/gammascope_api/routes/scenario.py | 5 +- apps/api/gammascope_api/routes/snapshot.py | 5 +- apps/api/gammascope_api/routes/stream.py | 5 +- apps/api/tests/test_contract_endpoints.py | 3 + apps/api/tests/test_experimental_routes.py | 3 + apps/api/tests/test_heatmap_route.py | 3 + apps/api/tests/test_live_snapshot_service.py | 240 ++++++ apps/api/tests/test_private_mode.py | 3 + apps/api/tests/test_stream_endpoint.py | 3 + .../app/api/spx/0dte/snapshot/latest/route.ts | 8 +- apps/web/app/page.tsx | 6 +- apps/web/app/replay/page.tsx | 5 +- apps/web/lib/serverBackendFetch.ts | 46 ++ .../lib/serverExperimentalAnalyticsSource.ts | 27 +- apps/web/lib/serverHeatmapSource.ts | 27 +- apps/web/lib/serverSnapshotSource.ts | 39 + apps/web/lib/snapshotSource.ts | 43 -- apps/web/tests/HeatmapPage.test.tsx | 15 +- .../LiveDashboardAdminHydration.test.tsx | 4 +- apps/web/tests/LivePage.test.tsx | 12 +- apps/web/tests/ReplayPage.test.tsx | 14 +- .../serverExperimentalAnalyticsSource.test.ts | 27 +- apps/web/tests/serverHeatmapSource.test.ts | 32 +- apps/web/tests/snapshotRoute.test.ts | 21 +- apps/web/tests/snapshotSource.test.ts | 30 +- ...-gammascope-backend-route-consolidation.md | 729 ++++++++++++++++++ 30 files changed, 1379 insertions(+), 152 deletions(-) create mode 100644 apps/api/gammascope_api/ingestion/live_snapshot_service.py create mode 100644 apps/api/tests/test_live_snapshot_service.py create mode 100644 apps/web/lib/serverBackendFetch.ts create mode 100644 apps/web/lib/serverSnapshotSource.ts create mode 100644 docs/superpowers/plans/2026-04-30-gammascope-backend-route-consolidation.md diff --git a/apps/api/gammascope_api/ingestion/collector_state.py b/apps/api/gammascope_api/ingestion/collector_state.py index 4db3ee3..0fac831 100644 --- a/apps/api/gammascope_api/ingestion/collector_state.py +++ b/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 diff --git a/apps/api/gammascope_api/ingestion/live_snapshot_service.py b/apps/api/gammascope_api/ingestion/live_snapshot_service.py new file mode 100644 index 0000000..178dd42 --- /dev/null +++ b/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) diff --git a/apps/api/gammascope_api/routes/experimental.py b/apps/api/gammascope_api/routes/experimental.py index 78c5cbb..f454cf0 100644 --- a/apps/api/gammascope_api/routes/experimental.py +++ b/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")) diff --git a/apps/api/gammascope_api/routes/heatmap.py b/apps/api/gammascope_api/routes/heatmap.py index 23e2fd3..8f9677d 100644 --- a/apps/api/gammascope_api/routes/heatmap.py +++ b/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) diff --git a/apps/api/gammascope_api/routes/scenario.py b/apps/api/gammascope_api/routes/scenario.py index 1df6a11..93f65b7 100644 --- a/apps/api/gammascope_api/routes/scenario.py +++ b/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) diff --git a/apps/api/gammascope_api/routes/snapshot.py b/apps/api/gammascope_api/routes/snapshot.py index a43f3a2..f378ba4 100644 --- a/apps/api/gammascope_api/routes/snapshot.py +++ b/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") diff --git a/apps/api/gammascope_api/routes/stream.py b/apps/api/gammascope_api/routes/stream.py index fe87648..01ce202 100644 --- a/apps/api/gammascope_api/routes/stream.py +++ b/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") diff --git a/apps/api/tests/test_contract_endpoints.py b/apps/api/tests/test_contract_endpoints.py index bf7e6b2..ab0cc1c 100644 --- a/apps/api/tests/test_contract_endpoints.py +++ b/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,6 +27,7 @@ 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()) @@ -33,6 +35,7 @@ def setup_function() -> None: def teardown_function() -> None: + reset_live_snapshot_service_override() reset_latest_state_cache_override() reset_saved_view_repository_override() diff --git a/apps/api/tests/test_experimental_routes.py b/apps/api/tests/test_experimental_routes.py index d38ca69..e043989 100644 --- a/apps/api/tests/test_experimental_routes.py +++ b/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() diff --git a/apps/api/tests/test_heatmap_route.py b/apps/api/tests/test_heatmap_route.py index 330f033..541c98c 100644 --- a/apps/api/tests/test_heatmap_route.py +++ b/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,6 +24,7 @@ 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()) @@ -30,6 +32,7 @@ def setup_function() -> None: def teardown_function() -> None: + reset_live_snapshot_service_override() reset_latest_state_cache_override() reset_replay_repository_override() reset_heatmap_repository_override() diff --git a/apps/api/tests/test_live_snapshot_service.py b/apps/api/tests/test_live_snapshot_service.py new file mode 100644 index 0000000..7e15c97 --- /dev/null +++ b/apps/api/tests/test_live_snapshot_service.py @@ -0,0 +1,240 @@ +from __future__ import annotations + +from gammascope_api.contracts.generated.collector_events import CollectorEvents +from gammascope_api.ingestion.collector_state import CollectorState +from gammascope_api.ingestion.live_snapshot_service import LiveSnapshotService + + +def test_collector_state_revision_increments_on_ingest() -> None: + state = CollectorState() + assert state.revision() == 0 + + state.ingest(CollectorEvents.model_validate(_health_event("2026-04-24T15:30:00Z"))) + + assert state.revision() == 1 + assert state.summary()["revision"] == 1 + assert state.snapshot()["revision"] == 1 + assert CollectorState.from_snapshot(state.snapshot()).revision() == 1 + assert CollectorState.from_snapshot(state.snapshot()).snapshot()["state_epoch"] == state.snapshot()["state_epoch"] + + +def test_collector_state_accessors_return_defensive_copies() -> None: + state = CollectorState() + for event in _spx_events(spot=5200.0, event_time="2026-04-24T15:30:01Z"): + state.ingest(CollectorEvents.model_validate(event)) + + state.latest_health()["status"] = "mutated" + state.latest_underlying_tick()["spot"] = 1 + state.contracts()[0]["strike"] = 1 + option_ticks = state.option_ticks() + option_ticks["SPX-2026-04-24-C-5200"]["bid"] = 1 + + assert state.latest_health()["status"] == "connected" + assert state.latest_underlying_tick()["spot"] == 5200.0 + assert state.contracts()[0]["strike"] == 5200 + assert state.option_ticks()["SPX-2026-04-24-C-5200"]["bid"] == 9.9 + + +def test_live_snapshot_service_caches_dashboard_snapshot_until_state_revision_changes(monkeypatch) -> None: + state = CollectorState() + for event in _spx_events(spot=5200.0, event_time="2026-04-24T15:30:01Z"): + state.ingest(CollectorEvents.model_validate(event)) + calls = 0 + + def fake_builder(input_state): + nonlocal calls + calls += 1 + return { + "session_id": "moomoo-spx-0dte-live", + "spot": input_state.latest_underlying_tick()["spot"], + } + + monkeypatch.setattr( + "gammascope_api.ingestion.live_snapshot_service.build_spx_dashboard_live_snapshot", + fake_builder, + ) + + service = LiveSnapshotService(lambda: state) + + assert service.dashboard_snapshot()["spot"] == 5200.0 + assert service.dashboard_snapshot()["spot"] == 5200.0 + assert calls == 1 + + for event in _spx_events(spot=5210.0, event_time="2026-04-24T15:30:02Z"): + state.ingest(CollectorEvents.model_validate(event)) + + assert service.dashboard_snapshot()["spot"] == 5210.0 + assert calls == 2 + + +def test_live_snapshot_service_refreshes_freshness_without_rebuilding(monkeypatch) -> None: + state = CollectorState() + state.ingest(CollectorEvents.model_validate(_health_event("2026-04-24T15:30:00Z"))) + calls = 0 + + def fake_builder(_input_state): + nonlocal calls + calls += 1 + return { + "session_id": "moomoo-spx-0dte-live", + "snapshot_time": "2024-01-01T00:00:00Z", + "freshness_ms": 0, + } + + monkeypatch.setattr( + "gammascope_api.ingestion.live_snapshot_service.build_spx_dashboard_live_snapshot", + fake_builder, + ) + + service = LiveSnapshotService(lambda: state) + + assert service.dashboard_snapshot()["freshness_ms"] > 0 + assert service.dashboard_snapshot()["freshness_ms"] > 0 + assert calls == 1 + + +def test_live_snapshot_service_invalidates_when_state_identity_changes_with_same_revision(monkeypatch) -> None: + first_state = CollectorState() + for event in _spx_events(spot=5200.0, event_time="2026-04-24T15:30:01Z"): + first_state.ingest(CollectorEvents.model_validate(event)) + second_state = CollectorState() + for event in _spx_events(spot=5210.0, event_time="2026-04-24T15:30:01Z"): + second_state.ingest(CollectorEvents.model_validate(event)) + states = iter([first_state, second_state]) + + def fake_builder(input_state): + return { + "session_id": "moomoo-spx-0dte-live", + "spot": input_state.latest_underlying_tick()["spot"], + } + + monkeypatch.setattr( + "gammascope_api.ingestion.live_snapshot_service.build_spx_dashboard_live_snapshot", + fake_builder, + ) + + service = LiveSnapshotService(lambda: next(states)) + + assert first_state.revision() == second_state.revision() + assert service.dashboard_snapshot()["spot"] == 5200.0 + assert service.dashboard_snapshot()["spot"] == 5210.0 + + +def test_live_snapshot_service_returns_defensive_copies(monkeypatch) -> None: + state = CollectorState() + state.ingest(CollectorEvents.model_validate(_health_event("2026-04-24T15:30:00Z"))) + + monkeypatch.setattr( + "gammascope_api.ingestion.live_snapshot_service.build_spx_dashboard_live_snapshot", + lambda _: {"session_id": "moomoo-spx-0dte-live", "rows": []}, + ) + + service = LiveSnapshotService(lambda: state) + first = service.dashboard_snapshot() + first["rows"].append({"mutated": True}) + + assert service.dashboard_snapshot()["rows"] == [] + + +def test_live_snapshot_service_maps_heatmap_symbols_to_live_sessions(monkeypatch) -> None: + state = CollectorState() + state.ingest(CollectorEvents.model_validate(_health_event("2026-04-24T15:30:00Z"))) + requested = [] + + def fake_build_live_snapshot(_state, *, session_id=None): + requested.append(session_id) + return {"session_id": session_id} + + monkeypatch.setattr( + "gammascope_api.ingestion.live_snapshot_service.build_live_snapshot", + fake_build_live_snapshot, + ) + + service = LiveSnapshotService(lambda: state) + + assert service.symbol_snapshot("SPY") == {"session_id": "moomoo-spy-0dte-live"} + assert requested == ["moomoo-spy-0dte-live"] + + +def _spx_events(*, spot: float, event_time: str) -> list[dict[str, object]]: + return [ + _health_event(event_time), + _underlying_event("moomoo-spx-0dte-live", "SPX", spot, event_time), + _contract_event("moomoo-spx-0dte-live", "SPX-2026-04-24-C-5200", "call", event_time), + _option_event("moomoo-spx-0dte-live", "SPX-2026-04-24-C-5200", event_time), + ] + + +def _health_event(event_time: str) -> dict[str, object]: + return { + "schema_version": "1.0.0", + "source": "ibkr", + "collector_id": "local-dev", + "status": "connected", + "ibkr_account_mode": "paper", + "message": "live", + "event_time": event_time, + "received_time": event_time, + } + + +def _underlying_event(session_id: str, symbol: str, spot: float, event_time: str) -> dict[str, object]: + return { + "schema_version": "1.0.0", + "source": "ibkr", + "session_id": session_id, + "symbol": symbol, + "spot": spot, + "bid": spot - 0.5, + "ask": spot + 0.5, + "last": spot, + "mark": spot, + "event_time": event_time, + "quote_status": "valid", + } + + +def _contract_event( + session_id: str, + contract_id: str, + right: str, + event_time: str, +) -> dict[str, object]: + return { + "schema_version": "1.0.0", + "source": "ibkr", + "session_id": session_id, + "contract_id": contract_id, + "ibkr_con_id": abs(hash(contract_id)) % 1_000_000, + "symbol": "SPX", + "expiry": "2026-04-24", + "right": right, + "strike": 5200, + "multiplier": 100, + "exchange": "CBOE", + "currency": "USD", + "event_time": event_time, + } + + +def _option_event(session_id: str, contract_id: str, event_time: str) -> dict[str, object]: + return { + "schema_version": "1.0.0", + "source": "ibkr", + "session_id": session_id, + "contract_id": contract_id, + "bid": 9.9, + "ask": 10.1, + "last": 10.0, + "bid_size": 10, + "ask_size": 11, + "volume": 380, + "open_interest": 2200, + "ibkr_iv": 0.2, + "ibkr_delta": 0.49, + "ibkr_gamma": 0.017, + "ibkr_vega": 0.9, + "ibkr_theta": -1.0, + "event_time": event_time, + "quote_status": "valid", + } diff --git a/apps/api/tests/test_private_mode.py b/apps/api/tests/test_private_mode.py index afddeca..8d4de30 100644 --- a/apps/api/tests/test_private_mode.py +++ b/apps/api/tests/test_private_mode.py @@ -8,6 +8,7 @@ reset_latest_state_cache_override, set_latest_state_cache_override, ) +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 @@ -24,6 +25,7 @@ def setup_function() -> None: collector_state.clear() + reset_live_snapshot_service_override() set_latest_state_cache_override(InMemoryLatestStateCache()) set_replay_repository_override(NullReplayRepository()) set_saved_view_repository_override(InMemorySavedViewRepository()) @@ -31,6 +33,7 @@ def setup_function() -> None: def teardown_function() -> None: + reset_live_snapshot_service_override() reset_latest_state_cache_override() reset_replay_repository_override() reset_saved_view_repository_override() diff --git a/apps/api/tests/test_stream_endpoint.py b/apps/api/tests/test_stream_endpoint.py index 357961d..d0aa818 100644 --- a/apps/api/tests/test_stream_endpoint.py +++ b/apps/api/tests/test_stream_endpoint.py @@ -11,6 +11,7 @@ reset_latest_state_cache_override, set_latest_state_cache_override, ) +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 @@ -24,12 +25,14 @@ def setup_function() -> None: collector_state.clear() + reset_live_snapshot_service_override() set_latest_state_cache_override(InMemoryLatestStateCache()) set_replay_repository_override(NullReplayRepository()) reset_replay_capture_circuit() def teardown_function() -> None: + reset_live_snapshot_service_override() reset_latest_state_cache_override() diff --git a/apps/web/app/api/spx/0dte/snapshot/latest/route.ts b/apps/web/app/api/spx/0dte/snapshot/latest/route.ts index 075754b..0d596ba 100644 --- a/apps/web/app/api/spx/0dte/snapshot/latest/route.ts +++ b/apps/web/app/api/spx/0dte/snapshot/latest/route.ts @@ -1,8 +1,10 @@ import { NextResponse } from "next/server"; -import { loadDashboardSnapshot } from "../../../../../../lib/snapshotSource"; +import { loadDashboardSnapshot } from "../../../../../../lib/serverSnapshotSource"; -export async function GET() { - const response = NextResponse.json(await loadDashboardSnapshot()); +export async function GET(request: Request) { + const response = NextResponse.json(await loadDashboardSnapshot({ + requestHeaders: request.headers + })); response.headers.set("Cache-Control", "no-store"); return response; } diff --git a/apps/web/app/page.tsx b/apps/web/app/page.tsx index 19adced..713a4f7 100644 --- a/apps/web/app/page.tsx +++ b/apps/web/app/page.tsx @@ -1,10 +1,10 @@ -import { cookies } from "next/headers"; +import { cookies, headers } from "next/headers"; import { LiveDashboard } from "../components/LiveDashboard"; import { ADMIN_COOKIE_NAME, adminLoginAvailable, parseAdminSessionValue } from "../lib/adminSession"; -import { loadDashboardSnapshot } from "../lib/snapshotSource"; +import { loadDashboardSnapshot } from "../lib/serverSnapshotSource"; export default async function Home() { - const snapshot = await loadDashboardSnapshot(); + const snapshot = await loadDashboardSnapshot({ requestHeaders: await headers() }); const initialAdminSession = await loadInitialAdminSession(); return ; diff --git a/apps/web/app/replay/page.tsx b/apps/web/app/replay/page.tsx index d809afc..9f6ec70 100644 --- a/apps/web/app/replay/page.tsx +++ b/apps/web/app/replay/page.tsx @@ -1,5 +1,6 @@ +import { headers } from "next/headers"; import { ReplayDashboard } from "../../components/ReplayDashboard"; -import { loadDashboardSnapshot } from "../../lib/snapshotSource"; +import { loadDashboardSnapshot } from "../../lib/serverSnapshotSource"; interface ReplayPageProps { searchParams?: Promise<{ @@ -8,7 +9,7 @@ interface ReplayPageProps { } export default async function ReplayPage({ searchParams }: ReplayPageProps) { - const snapshot = await loadDashboardSnapshot(); + const snapshot = await loadDashboardSnapshot({ requestHeaders: await headers() }); const params = await searchParams; const requestedSessionId = firstSearchParamValue(params?.session_id); diff --git a/apps/web/lib/serverBackendFetch.ts b/apps/web/lib/serverBackendFetch.ts new file mode 100644 index 0000000..9c347dc --- /dev/null +++ b/apps/web/lib/serverBackendFetch.ts @@ -0,0 +1,46 @@ +import { verifyAdminRequest } from "./adminSession"; + +export const DEFAULT_API_BASE_URL = "http://127.0.0.1:8000"; +export const ADMIN_TOKEN_HEADER = "X-GammaScope-Admin-Token"; + +export function backendApiUrl( + path: string, + searchParams?: URLSearchParams, + apiBaseUrl = process.env.GAMMASCOPE_API_BASE_URL ?? DEFAULT_API_BASE_URL +): string { + const base = apiBaseUrl.replace(/\/+$/, ""); + const normalizedPath = path.startsWith("/") ? path : `/${path}`; + const query = searchParams?.toString(); + + return query ? `${base}${normalizedPath}?${query}` : `${base}${normalizedPath}`; +} + +export function backendJsonHeaders(requestHeaders?: Pick): HeadersInit { + const headers: Record = { + Accept: "application/json" + }; + const adminToken = process.env.GAMMASCOPE_ADMIN_TOKEN?.trim(); + + if (adminToken && requestHeaders && requestHasValidAdminSession(requestHeaders)) { + headers[ADMIN_TOKEN_HEADER] = adminToken; + } + + return headers; +} + +function requestHasValidAdminSession(requestHeaders: Pick): boolean { + const cookie = requestHeaders.get("cookie"); + if (!cookie) { + return false; + } + + const host = requestHeaders.get("x-forwarded-host") ?? requestHeaders.get("host") ?? "localhost:3000"; + const protocol = requestHeaders.get("x-forwarded-proto") ?? "http"; + const request = new Request(`${protocol}://${host}/__gammascope_backend_fetch_auth`, { + headers: { + cookie + } + }); + + return verifyAdminRequest(request, { csrf: false }).ok; +} diff --git a/apps/web/lib/serverExperimentalAnalyticsSource.ts b/apps/web/lib/serverExperimentalAnalyticsSource.ts index 3e2c2e6..968d9ab 100644 --- a/apps/web/lib/serverExperimentalAnalyticsSource.ts +++ b/apps/web/lib/serverExperimentalAnalyticsSource.ts @@ -1,8 +1,9 @@ import seed from "../../../packages/contracts/fixtures/experimental-analytics.seed.json"; import { isExperimentalAnalytics } from "./clientExperimentalAnalyticsSource"; import type { ExperimentalAnalytics } from "./contracts"; +import { backendApiUrl, backendJsonHeaders } from "./serverBackendFetch"; -const EXPERIMENTAL_LATEST_PROXY_PATH = "/api/spx/0dte/experimental/latest"; +const EXPERIMENTAL_LATEST_PATH = "/api/spx/0dte/experimental/latest"; const seedExperimentalAnalytics = seed as ExperimentalAnalytics; @@ -11,9 +12,9 @@ export async function loadLatestExperimentalAnalytics( requestHeaders?: Pick ): Promise { try { - const response = await fetcher(sameOriginProxyUrl(requestHeaders), { + const response = await fetcher(backendApiUrl(EXPERIMENTAL_LATEST_PATH), { cache: "no-store", - headers: proxyRequestHeaders(requestHeaders) + headers: backendJsonHeaders(requestHeaders) }); if (!response.ok) { @@ -26,23 +27,3 @@ export async function loadLatestExperimentalAnalytics( return seedExperimentalAnalytics; } } - -function sameOriginProxyUrl(requestHeaders?: Pick): string { - const host = requestHeaders?.get("x-forwarded-host") ?? requestHeaders?.get("host") ?? "localhost:3000"; - const protocol = requestHeaders?.get("x-forwarded-proto") ?? "http"; - - return `${protocol}://${host}${EXPERIMENTAL_LATEST_PROXY_PATH}`; -} - -function proxyRequestHeaders(requestHeaders?: Pick): HeadersInit { - const headers: Record = { - Accept: "application/json" - }; - const cookie = requestHeaders?.get("cookie"); - - if (cookie) { - headers.Cookie = cookie; - } - - return headers; -} diff --git a/apps/web/lib/serverHeatmapSource.ts b/apps/web/lib/serverHeatmapSource.ts index 8a1087a..d26a39b 100644 --- a/apps/web/lib/serverHeatmapSource.ts +++ b/apps/web/lib/serverHeatmapSource.ts @@ -1,6 +1,7 @@ import { HEATMAP_SYMBOLS, isHeatmapPayload, type HeatmapPayload, type HeatmapSymbol } from "./clientHeatmapSource"; +import { backendApiUrl, backendJsonHeaders } from "./serverBackendFetch"; -const HEATMAP_PROXY_PATH = "/api/spx/0dte/heatmap/latest"; +const HEATMAP_PATH = "/api/spx/0dte/heatmap/latest"; export async function loadLatestHeatmap( fetcher: typeof fetch = fetch, @@ -28,9 +29,9 @@ async function loadLatestHeatmapForSymbol( const params = new URLSearchParams({ metric: "gex", symbol }); try { - const response = await fetcher(`${sameOriginProxyUrl(requestHeaders)}?${params.toString()}`, { + const response = await fetcher(backendApiUrl(HEATMAP_PATH, params), { cache: "no-store", - headers: proxyRequestHeaders(requestHeaders) + headers: backendJsonHeaders(requestHeaders) }); if (!response.ok) { @@ -70,23 +71,3 @@ function unavailableHeatmap(symbol: HeatmapSymbol): HeatmapPayload { } }; } - -function sameOriginProxyUrl(requestHeaders?: Pick): string { - const host = requestHeaders?.get("x-forwarded-host") ?? requestHeaders?.get("host") ?? "localhost:3000"; - const protocol = requestHeaders?.get("x-forwarded-proto") ?? "http"; - - return `${protocol}://${host}${HEATMAP_PROXY_PATH}`; -} - -function proxyRequestHeaders(requestHeaders?: Pick): HeadersInit { - const headers: Record = { - Accept: "application/json" - }; - const cookie = requestHeaders?.get("cookie"); - - if (cookie) { - headers.Cookie = cookie; - } - - return headers; -} diff --git a/apps/web/lib/serverSnapshotSource.ts b/apps/web/lib/serverSnapshotSource.ts new file mode 100644 index 0000000..3326726 --- /dev/null +++ b/apps/web/lib/serverSnapshotSource.ts @@ -0,0 +1,39 @@ +import type { AnalyticsSnapshot } from "./contracts"; +import { isAnalyticsSnapshot } from "./snapshotSource"; +import { seedSnapshot } from "./seedSnapshot"; +import { backendApiUrl, backendJsonHeaders } from "./serverBackendFetch"; + +const SNAPSHOT_PATH = "/api/spx/0dte/snapshot/latest"; + +type SnapshotFetcher = (input: string, init: RequestInit) => Promise; + +type LoadDashboardSnapshotOptions = { + apiBaseUrl?: string; + fetcher?: SnapshotFetcher; + requestHeaders?: Pick; +}; + +export async function loadDashboardSnapshot(options: LoadDashboardSnapshotOptions = {}): Promise { + const fetcher = options.fetcher ?? fetch; + + try { + const response = await fetcher(backendApiUrl(SNAPSHOT_PATH, undefined, options.apiBaseUrl), { + cache: "no-store", + headers: backendJsonHeaders(options.requestHeaders) + }); + + if (!response.ok) { + return seedSnapshot; + } + + const payload = await response.json(); + + if (!isAnalyticsSnapshot(payload)) { + return seedSnapshot; + } + + return payload; + } catch { + return seedSnapshot; + } +} diff --git a/apps/web/lib/snapshotSource.ts b/apps/web/lib/snapshotSource.ts index e4473da..34a6df7 100644 --- a/apps/web/lib/snapshotSource.ts +++ b/apps/web/lib/snapshotSource.ts @@ -1,15 +1,4 @@ import type { AnalyticsSnapshot } from "./contracts"; -import { seedSnapshot } from "./seedSnapshot"; - -const DEFAULT_API_BASE_URL = "http://127.0.0.1:8000"; -const SNAPSHOT_PATH = "/api/spx/0dte/snapshot/latest"; - -type SnapshotFetcher = (input: string, init: RequestInit) => Promise; - -type LoadDashboardSnapshotOptions = { - apiBaseUrl?: string; - fetcher?: SnapshotFetcher; -}; type Validator = (value: unknown) => boolean; @@ -62,10 +51,6 @@ const ROW_FIELDS: Record = { comparison_status: (value) => isOneOf(value, ["ok", "missing", "stale", "outside_tolerance", "not_supported"]) }; -function snapshotUrl(apiBaseUrl: string): string { - return `${apiBaseUrl.replace(/\/+$/, "")}${SNAPSHOT_PATH}`; -} - export function isAnalyticsSnapshot(payload: unknown): payload is AnalyticsSnapshot { if (!isRecord(payload)) { return false; @@ -129,31 +114,3 @@ function isScenarioParams(value: unknown): value is AnalyticsSnapshot["scenario_ function isOneOf(value: unknown, allowedValues: readonly T[]): value is T { return typeof value === "string" && allowedValues.includes(value as T); } - -export async function loadDashboardSnapshot(options: LoadDashboardSnapshotOptions = {}): Promise { - const apiBaseUrl = options.apiBaseUrl ?? process.env.GAMMASCOPE_API_BASE_URL ?? DEFAULT_API_BASE_URL; - const fetcher = options.fetcher ?? fetch; - - try { - const response = await fetcher(snapshotUrl(apiBaseUrl), { - cache: "no-store", - headers: { - Accept: "application/json" - } - }); - - if (!response.ok) { - return seedSnapshot; - } - - const payload = await response.json(); - - if (!isAnalyticsSnapshot(payload)) { - return seedSnapshot; - } - - return payload; - } catch { - return seedSnapshot; - } -} diff --git a/apps/web/tests/HeatmapPage.test.tsx b/apps/web/tests/HeatmapPage.test.tsx index 79bb943..a3c2c0b 100644 --- a/apps/web/tests/HeatmapPage.test.tsx +++ b/apps/web/tests/HeatmapPage.test.tsx @@ -21,6 +21,7 @@ vi.mock("next/headers", () => ({ describe("HeatmapPage", () => { afterEach(() => { vi.unstubAllGlobals(); + vi.unstubAllEnvs(); vi.resetModules(); mocks.heatmapProps.mockReset(); mocks.requestHeaders.mockReset(); @@ -39,23 +40,23 @@ describe("HeatmapPage", () => { "Content-Type": "application/json" } }))); + vi.stubEnv("GAMMASCOPE_API_BASE_URL", "http://fastapi.test/"); vi.stubGlobal("React", React); const { default: HeatmapPage } = await import("../app/heatmap/page"); const page = await HeatmapPage(); expect(renderToStaticMarkup(page)).toContain("Heatmap page shell"); - expect(fetch).toHaveBeenCalledWith("https://gammascope.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=SPX", { + expect(fetch).toHaveBeenCalledWith("http://fastapi.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=SPX", { cache: "no-store", headers: { - Accept: "application/json", - Cookie: "gammascope_admin=signed-session" + Accept: "application/json" } }); - expect(fetch).toHaveBeenCalledWith("https://gammascope.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=SPY", expect.any(Object)); - expect(fetch).toHaveBeenCalledWith("https://gammascope.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=QQQ", expect.any(Object)); - expect(fetch).toHaveBeenCalledWith("https://gammascope.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=NDX", expect.any(Object)); - expect(fetch).toHaveBeenCalledWith("https://gammascope.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=IWM", expect.any(Object)); + expect(fetch).toHaveBeenCalledWith("http://fastapi.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=SPY", expect.any(Object)); + expect(fetch).toHaveBeenCalledWith("http://fastapi.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=QQQ", expect.any(Object)); + expect(fetch).toHaveBeenCalledWith("http://fastapi.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=NDX", expect.any(Object)); + expect(fetch).toHaveBeenCalledWith("http://fastapi.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=IWM", expect.any(Object)); expect(mocks.heatmapProps).toHaveBeenCalledWith({ initialPayloads: [ heatmapPayload("SPX", "SPXW"), diff --git a/apps/web/tests/LiveDashboardAdminHydration.test.tsx b/apps/web/tests/LiveDashboardAdminHydration.test.tsx index 49e93a9..2ba2441 100644 --- a/apps/web/tests/LiveDashboardAdminHydration.test.tsx +++ b/apps/web/tests/LiveDashboardAdminHydration.test.tsx @@ -11,8 +11,9 @@ const ADMIN_ENV = { } as const; const cookieValue = vi.fn(() => undefined as string | undefined); +const requestHeaders = new Headers({ host: "gamma.test" }); -vi.mock("../lib/snapshotSource", () => ({ +vi.mock("../lib/serverSnapshotSource", () => ({ loadDashboardSnapshot: vi.fn(async () => seedSnapshot) })); @@ -21,6 +22,7 @@ vi.mock("../components/DashboardChart", () => ({ })); vi.mock("next/headers", () => ({ + headers: vi.fn(async () => requestHeaders), cookies: vi.fn(async () => ({ get: vi.fn((name: string) => { const value = cookieValue(); diff --git a/apps/web/tests/LivePage.test.tsx b/apps/web/tests/LivePage.test.tsx index 8e2a64d..d1748e3 100644 --- a/apps/web/tests/LivePage.test.tsx +++ b/apps/web/tests/LivePage.test.tsx @@ -12,12 +12,15 @@ const ADMIN_ENV = { const cookieValue = vi.fn(() => undefined as string | undefined); const liveDashboardProps = vi.fn(); +const requestHeaders = new Headers({ host: "gamma.test" }); +const loadDashboardSnapshot = vi.fn(async () => seedSnapshot); -vi.mock("../lib/snapshotSource", () => ({ - loadDashboardSnapshot: vi.fn(async () => seedSnapshot) +vi.mock("../lib/serverSnapshotSource", () => ({ + loadDashboardSnapshot })); vi.mock("next/headers", () => ({ + headers: vi.fn(async () => requestHeaders), cookies: vi.fn(async () => ({ get: vi.fn((name: string) => { const value = cookieValue(); @@ -45,6 +48,8 @@ describe("Home page", () => { vi.unstubAllEnvs(); vi.resetModules(); cookieValue.mockReset(); + loadDashboardSnapshot.mockReset(); + loadDashboardSnapshot.mockResolvedValue(seedSnapshot); liveDashboardProps.mockReset(); }); @@ -60,6 +65,9 @@ describe("Home page", () => { const page = await Home(); expect(renderToStaticMarkup(page)).toContain("Live dashboard"); + expect(loadDashboardSnapshot).toHaveBeenCalledWith({ + requestHeaders + }); expect(liveDashboardProps).toHaveBeenCalledWith({ initialSnapshot: seedSnapshot, initialAdminSession: { diff --git a/apps/web/tests/ReplayPage.test.tsx b/apps/web/tests/ReplayPage.test.tsx index 35944d5..0afdd19 100644 --- a/apps/web/tests/ReplayPage.test.tsx +++ b/apps/web/tests/ReplayPage.test.tsx @@ -3,8 +3,15 @@ import { describe, expect, it, vi } from "vitest"; import { renderToStaticMarkup } from "react-dom/server"; import { seedSnapshot } from "../lib/seedSnapshot"; -vi.mock("../lib/snapshotSource", () => ({ - loadDashboardSnapshot: vi.fn(async () => seedSnapshot) +const requestHeaders = new Headers({ host: "gamma.test" }); +const loadDashboardSnapshot = vi.fn(async () => seedSnapshot); + +vi.mock("../lib/serverSnapshotSource", () => ({ + loadDashboardSnapshot +})); + +vi.mock("next/headers", () => ({ + headers: vi.fn(async () => requestHeaders) })); vi.mock("../components/ReplayDashboard", () => ({ @@ -23,6 +30,9 @@ describe("ReplayPage", () => { }) }); + expect(loadDashboardSnapshot).toHaveBeenCalledWith({ + requestHeaders + }); expect(renderToStaticMarkup(page)).toContain("data-requested-session-id=\"import-session-ready\""); }); }); diff --git a/apps/web/tests/serverExperimentalAnalyticsSource.test.ts b/apps/web/tests/serverExperimentalAnalyticsSource.test.ts index d99df31..544508e 100644 --- a/apps/web/tests/serverExperimentalAnalyticsSource.test.ts +++ b/apps/web/tests/serverExperimentalAnalyticsSource.test.ts @@ -1,12 +1,18 @@ -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import seed from "../../../packages/contracts/fixtures/experimental-analytics.seed.json"; +import { ADMIN_COOKIE_NAME, createAdminSessionValue } from "../lib/adminSession"; import { loadLatestExperimentalAnalytics } from "../lib/serverExperimentalAnalyticsSource"; import type { ExperimentalAnalytics } from "../lib/contracts"; const seedPayload = seed as ExperimentalAnalytics; describe("loadLatestExperimentalAnalytics", () => { - it("loads latest experimental analytics from the same-origin proxy URL", async () => { + afterEach(() => { + vi.unstubAllEnvs(); + }); + + it("loads latest experimental analytics directly from FastAPI on the server", async () => { + vi.stubEnv("GAMMASCOPE_API_BASE_URL", "http://fastapi.test/"); const payload: ExperimentalAnalytics = { ...seedPayload, meta: { @@ -26,7 +32,7 @@ describe("loadLatestExperimentalAnalytics", () => { "x-forwarded-proto": "https" }))).resolves.toEqual(payload); - expect(fetcher).toHaveBeenCalledWith("https://gamma.example/api/spx/0dte/experimental/latest", { + expect(fetcher).toHaveBeenCalledWith("http://fastapi.test/api/spx/0dte/experimental/latest", { cache: "no-store", headers: { Accept: "application/json" @@ -34,24 +40,31 @@ describe("loadLatestExperimentalAnalytics", () => { }); }); - it("forwards the request cookie when present", async () => { + it("forwards the backend admin token for a valid server admin session", async () => { + vi.stubEnv("GAMMASCOPE_API_BASE_URL", "http://fastapi.test"); + vi.stubEnv("GAMMASCOPE_ADMIN_TOKEN", "api-admin-token"); + vi.stubEnv("GAMMASCOPE_WEB_ADMIN_USERNAME", "admin"); + vi.stubEnv("GAMMASCOPE_WEB_ADMIN_PASSWORD", "password"); + vi.stubEnv("GAMMASCOPE_WEB_ADMIN_SESSION_SECRET", "x".repeat(32)); + const fetcher = vi.fn(async () => new Response(JSON.stringify(seedPayload), { status: 200, headers: { "Content-Type": "application/json" } })); + const sessionCookie = `${ADMIN_COOKIE_NAME}=${encodeURIComponent(createAdminSessionValue())}`; await loadLatestExperimentalAnalytics(fetcher as typeof fetch, new Headers({ host: "gamma.local", - cookie: "session=abc" + cookie: sessionCookie })); - expect(fetcher).toHaveBeenCalledWith("http://gamma.local/api/spx/0dte/experimental/latest", { + expect(fetcher).toHaveBeenCalledWith("http://fastapi.test/api/spx/0dte/experimental/latest", { cache: "no-store", headers: { Accept: "application/json", - Cookie: "session=abc" + "X-GammaScope-Admin-Token": "api-admin-token" } }); }); diff --git a/apps/web/tests/serverHeatmapSource.test.ts b/apps/web/tests/serverHeatmapSource.test.ts index ef84b9b..77f5aeb 100644 --- a/apps/web/tests/serverHeatmapSource.test.ts +++ b/apps/web/tests/serverHeatmapSource.test.ts @@ -1,8 +1,38 @@ -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { loadLatestHeatmaps } from "../lib/serverHeatmapSource"; import type { HeatmapPayload } from "../lib/clientHeatmapSource"; describe("loadLatestHeatmaps", () => { + afterEach(() => { + vi.unstubAllEnvs(); + }); + + it("loads heatmap symbols directly from FastAPI on the server", async () => { + vi.stubEnv("GAMMASCOPE_API_BASE_URL", "http://fastapi.test/"); + const fetcher = vi.fn(async (input: string) => { + const symbol = new URL(input).searchParams.get("symbol"); + + return new Response(JSON.stringify(heatmapPayload(toSupportedSymbol(symbol))), { + status: 200, + headers: { + "Content-Type": "application/json" + } + }); + }); + + await loadLatestHeatmaps(fetcher as typeof fetch, new Headers({ host: "gamma.test" })); + + expect(fetcher).toHaveBeenCalledWith( + "http://fastapi.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=SPX", + expect.objectContaining({ + cache: "no-store", + headers: { + Accept: "application/json" + } + }) + ); + }); + it("keeps all supported panel slots when a symbol request is unavailable", async () => { const fetcher = vi.fn(async (input: string) => { const symbol = new URL(input).searchParams.get("symbol"); diff --git a/apps/web/tests/snapshotRoute.test.ts b/apps/web/tests/snapshotRoute.test.ts index 44511b2..60a90c4 100644 --- a/apps/web/tests/snapshotRoute.test.ts +++ b/apps/web/tests/snapshotRoute.test.ts @@ -1,14 +1,19 @@ -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import type { AnalyticsSnapshot } from "../lib/contracts"; import { seedSnapshot } from "../lib/seedSnapshot"; -const loadDashboardSnapshot = vi.fn<() => Promise>(); +const loadDashboardSnapshot = vi.fn(); -vi.mock("../lib/snapshotSource", () => ({ +vi.mock("../lib/serverSnapshotSource", () => ({ loadDashboardSnapshot })); describe("GET /api/spx/0dte/snapshot/latest", () => { + afterEach(() => { + loadDashboardSnapshot.mockReset(); + vi.resetModules(); + }); + it("returns the latest dashboard snapshot without caching", async () => { const snapshot = { ...seedSnapshot, @@ -25,9 +30,17 @@ describe("GET /api/spx/0dte/snapshot/latest", () => { loadDashboardSnapshot.mockResolvedValue(snapshot); const { GET } = await import("../app/api/spx/0dte/snapshot/latest/route"); - const response = await GET(); + const request = new Request("http://localhost/api/spx/0dte/snapshot/latest", { + headers: { + cookie: "gammascope_admin=signed-session" + } + }); + const response = await GET(request); await expect(response.json()).resolves.toEqual(snapshot); expect(response.headers.get("Cache-Control")).toBe("no-store"); + expect(loadDashboardSnapshot).toHaveBeenCalledWith({ + requestHeaders: request.headers + }); }); }); diff --git a/apps/web/tests/snapshotSource.test.ts b/apps/web/tests/snapshotSource.test.ts index a9cfe15..daebbd8 100644 --- a/apps/web/tests/snapshotSource.test.ts +++ b/apps/web/tests/snapshotSource.test.ts @@ -1,5 +1,6 @@ import { afterEach, describe, expect, it, vi } from "vitest"; -import { loadDashboardSnapshot } from "../lib/snapshotSource"; +import { ADMIN_COOKIE_NAME, createAdminSessionValue } from "../lib/adminSession"; +import { loadDashboardSnapshot } from "../lib/serverSnapshotSource"; import { seedSnapshot } from "../lib/seedSnapshot"; import type { AnalyticsSnapshot } from "../lib/contracts"; @@ -67,6 +68,33 @@ describe("loadDashboardSnapshot", () => { }); }); + it("forwards the backend admin token for a valid server admin session", async () => { + vi.stubEnv("GAMMASCOPE_ADMIN_TOKEN", "api-admin-token"); + vi.stubEnv("GAMMASCOPE_WEB_ADMIN_USERNAME", "admin"); + vi.stubEnv("GAMMASCOPE_WEB_ADMIN_PASSWORD", "password"); + vi.stubEnv("GAMMASCOPE_WEB_ADMIN_SESSION_SECRET", "x".repeat(32)); + const snapshot = apiSnapshot(); + const fetcher = vi.fn(async () => jsonResponse(snapshot)); + const sessionCookie = `${ADMIN_COOKIE_NAME}=${encodeURIComponent(createAdminSessionValue())}`; + + await loadDashboardSnapshot({ + apiBaseUrl: "http://testserver", + fetcher, + requestHeaders: new Headers({ + cookie: sessionCookie, + host: "gamma.local" + }) + }); + + expect(fetcher).toHaveBeenCalledWith("http://testserver/api/spx/0dte/snapshot/latest", { + cache: "no-store", + headers: { + Accept: "application/json", + "X-GammaScope-Admin-Token": "api-admin-token" + } + }); + }); + it("falls back to the seed snapshot when fetching rejects", async () => { const fetcher = vi.fn(async () => { throw new Error("offline"); diff --git a/docs/superpowers/plans/2026-04-30-gammascope-backend-route-consolidation.md b/docs/superpowers/plans/2026-04-30-gammascope-backend-route-consolidation.md new file mode 100644 index 0000000..ac325cf --- /dev/null +++ b/docs/superpowers/plans/2026-04-30-gammascope-backend-route-consolidation.md @@ -0,0 +1,729 @@ +# GammaScope Backend Route Consolidation Implementation Plan + +> **For agentic workers:** REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Remove avoidable server-side proxy hops and centralize live snapshot materialization so frontend pages and backend routes use one clear live data path. + +**Architecture:** Keep one FastAPI backend, one Next.js web app, and source-specific collectors. Browser/client calls continue to use same-origin Next API routes for auth and cookie boundaries, while server components fetch FastAPI directly. FastAPI routes use a shared live snapshot service/cache keyed by collector state revision and session/symbol instead of rebuilding snapshots independently in each route. + +**Tech Stack:** Next.js App Router, Vitest, FastAPI, pytest, Pydantic-generated contracts, local collector state, Postgres-backed replay/heatmap persistence. + +--- + +## File Structure + +- Create: `apps/web/lib/serverBackendFetch.ts` + - Owns FastAPI URL construction and optional admin-token forwarding for server-side web loaders. +- Modify: `apps/web/lib/serverExperimentalAnalyticsSource.ts` + - Fetches `GET /api/spx/0dte/experimental/latest` directly from FastAPI on the server. +- Modify: `apps/web/lib/serverHeatmapSource.ts` + - Fetches `GET /api/spx/0dte/heatmap/latest` directly from FastAPI on the server. +- Modify: `apps/web/tests/serverExperimentalAnalyticsSource.test.ts` + - Updates expectations from same-origin proxy URL to FastAPI URL. + - Adds admin-token forwarding coverage. +- Modify: `apps/web/tests/serverHeatmapSource.test.ts` + - Updates expectations so symbol requests target FastAPI directly. +- Modify: `apps/api/gammascope_api/ingestion/collector_state.py` + - Adds a monotonic revision to safely invalidate cached live snapshots. +- Create: `apps/api/gammascope_api/ingestion/live_snapshot_service.py` + - Owns cached live dashboard/session/symbol snapshot materialization. +- Modify: `apps/api/gammascope_api/routes/snapshot.py` + - Reads latest dashboard snapshot through the shared service. +- Modify: `apps/api/gammascope_api/routes/experimental.py` + - Reads latest dashboard snapshot through the shared service. +- Modify: `apps/api/gammascope_api/routes/heatmap.py` + - Reads symbol snapshots through the shared service. +- Modify: `apps/api/gammascope_api/routes/scenario.py` + - Reads latest dashboard snapshot through the shared service. +- Modify: `apps/api/gammascope_api/routes/stream.py` + - Streams latest dashboard snapshots through the shared service. +- Create: `apps/api/tests/test_live_snapshot_service.py` + - Verifies cache reuse, invalidation, defensive copies, and symbol mapping. +- Modify existing route tests only if needed: + - `apps/api/tests/test_contract_endpoints.py` + - `apps/api/tests/test_experimental_routes.py` + - `apps/api/tests/test_heatmap_route.py` + - `apps/api/tests/test_stream_endpoint.py` + - `apps/api/tests/test_private_mode.py` + +## Chunk 1: Frontend Server Fetch Consolidation + +### Task 1: Write frontend failing tests for direct FastAPI server fetches + +**Files:** +- Modify: `apps/web/tests/serverExperimentalAnalyticsSource.test.ts` +- Modify: `apps/web/tests/serverHeatmapSource.test.ts` + +- [ ] **Step 1: Update the experimental server source URL test** + +Change the current same-origin expectation to a direct FastAPI expectation. + +```ts +it("loads latest experimental analytics directly from FastAPI on the server", async () => { + vi.stubEnv("GAMMASCOPE_API_BASE_URL", "http://fastapi.test/"); + const payload: ExperimentalAnalytics = { + ...seedPayload, + meta: { + ...seedPayload.meta, + sourceSessionId: "api-session" + } + }; + const fetcher = vi.fn(async () => new Response(JSON.stringify(payload), { + status: 200, + headers: { + "Content-Type": "application/json" + } + })); + + await expect(loadLatestExperimentalAnalytics(fetcher as typeof fetch, new Headers({ + "x-forwarded-host": "gamma.example", + "x-forwarded-proto": "https" + }))).resolves.toEqual(payload); + + expect(fetcher).toHaveBeenCalledWith("http://fastapi.test/api/spx/0dte/experimental/latest", { + cache: "no-store", + headers: { + Accept: "application/json" + } + }); +}); +``` + +- [ ] **Step 2: Add experimental admin-token forwarding coverage** + +Add imports: + +```ts +import { ADMIN_COOKIE_NAME, createAdminSessionValue } from "../lib/adminSession"; +``` + +Add a test that a valid web admin cookie causes the server loader to forward only the backend admin token, not the browser cookie. + +```ts +it("forwards the backend admin token for a valid server admin session", async () => { + vi.stubEnv("GAMMASCOPE_API_BASE_URL", "http://fastapi.test"); + vi.stubEnv("GAMMASCOPE_ADMIN_TOKEN", "api-admin-token"); + vi.stubEnv("GAMMASCOPE_WEB_ADMIN_USERNAME", "admin"); + vi.stubEnv("GAMMASCOPE_WEB_ADMIN_PASSWORD", "password"); + vi.stubEnv("GAMMASCOPE_WEB_ADMIN_SESSION_SECRET", "x".repeat(32)); + + const fetcher = vi.fn(async () => new Response(JSON.stringify(seedPayload), { + status: 200, + headers: { + "Content-Type": "application/json" + } + })); + const sessionCookie = `${ADMIN_COOKIE_NAME}=${encodeURIComponent(createAdminSessionValue())}`; + + await loadLatestExperimentalAnalytics(fetcher as typeof fetch, new Headers({ + host: "gamma.local", + cookie: sessionCookie + })); + + expect(fetcher).toHaveBeenCalledWith("http://fastapi.test/api/spx/0dte/experimental/latest", { + cache: "no-store", + headers: { + Accept: "application/json", + "X-GammaScope-Admin-Token": "api-admin-token" + } + }); +}); +``` + +- [ ] **Step 3: Add heatmap direct FastAPI URL coverage** + +Extend `apps/web/tests/serverHeatmapSource.test.ts` with a URL assertion. Keep the existing unavailable-slot behavior. + +```ts +it("loads heatmap symbols directly from FastAPI on the server", async () => { + vi.stubEnv("GAMMASCOPE_API_BASE_URL", "http://fastapi.test/"); + const fetcher = vi.fn(async (input: string) => { + const symbol = new URL(input).searchParams.get("symbol"); + return new Response(JSON.stringify(heatmapPayload(toSupportedSymbol(symbol))), { + status: 200, + headers: { + "Content-Type": "application/json" + } + }); + }); + + await loadLatestHeatmaps(fetcher as typeof fetch, new Headers({ host: "gamma.test" })); + + expect(fetcher).toHaveBeenCalledWith( + "http://fastapi.test/api/spx/0dte/heatmap/latest?metric=gex&symbol=SPX", + expect.objectContaining({ + cache: "no-store", + headers: { + Accept: "application/json" + } + }) + ); +}); +``` + +- [ ] **Step 4: Run frontend tests to verify RED** + +Run: + +```bash +pnpm --filter @gammascope/web test -- serverExperimentalAnalyticsSource.test.ts serverHeatmapSource.test.ts +``` + +Expected: FAIL because the current loaders call `http:///api/...`. + +### Task 2: Implement the shared server backend fetch helper + +**Files:** +- Create: `apps/web/lib/serverBackendFetch.ts` +- Modify: `apps/web/lib/serverExperimentalAnalyticsSource.ts` +- Modify: `apps/web/lib/serverHeatmapSource.ts` + +- [ ] **Step 1: Create `serverBackendFetch.ts`** + +Use this helper shape. Keep it small; do not move browser/client source helpers into it. + +```ts +import { verifyAdminRequest } from "./adminSession"; + +export const DEFAULT_API_BASE_URL = "http://127.0.0.1:8000"; +export const ADMIN_TOKEN_HEADER = "X-GammaScope-Admin-Token"; + +export function backendApiUrl( + path: string, + searchParams?: URLSearchParams, + apiBaseUrl = process.env.GAMMASCOPE_API_BASE_URL ?? DEFAULT_API_BASE_URL +): string { + const base = apiBaseUrl.replace(/\/+$/, ""); + const normalizedPath = path.startsWith("/") ? path : `/${path}`; + const query = searchParams?.toString(); + return query ? `${base}${normalizedPath}?${query}` : `${base}${normalizedPath}`; +} + +export function backendJsonHeaders(requestHeaders?: Pick): HeadersInit { + const headers: Record = { + Accept: "application/json" + }; + const adminToken = process.env.GAMMASCOPE_ADMIN_TOKEN?.trim(); + + if (adminToken && requestHeaders && requestHasValidAdminSession(requestHeaders)) { + headers[ADMIN_TOKEN_HEADER] = adminToken; + } + + return headers; +} + +function requestHasValidAdminSession(requestHeaders: Pick): boolean { + const cookie = requestHeaders.get("cookie"); + if (!cookie) { + return false; + } + + const host = requestHeaders.get("x-forwarded-host") ?? requestHeaders.get("host") ?? "localhost:3000"; + const protocol = requestHeaders.get("x-forwarded-proto") ?? "http"; + const request = new Request(`${protocol}://${host}/__gammascope_backend_fetch_auth`, { + headers: { + cookie + } + }); + + return verifyAdminRequest(request, { csrf: false }).ok; +} +``` + +- [ ] **Step 2: Update `serverExperimentalAnalyticsSource.ts`** + +Replace `sameOriginProxyUrl` and `proxyRequestHeaders` with the helper. + +```ts +import { backendApiUrl, backendJsonHeaders } from "./serverBackendFetch"; + +const EXPERIMENTAL_LATEST_PATH = "/api/spx/0dte/experimental/latest"; + +// inside loadLatestExperimentalAnalytics: +const response = await fetcher(backendApiUrl(EXPERIMENTAL_LATEST_PATH), { + cache: "no-store", + headers: backendJsonHeaders(requestHeaders) +}); +``` + +Delete the old `sameOriginProxyUrl` and `proxyRequestHeaders` functions from this file. + +- [ ] **Step 3: Update `serverHeatmapSource.ts`** + +Replace same-origin proxy URL construction with direct FastAPI URL construction. + +```ts +import { backendApiUrl, backendJsonHeaders } from "./serverBackendFetch"; + +const HEATMAP_PATH = "/api/spx/0dte/heatmap/latest"; + +// inside loadLatestHeatmapForSymbol: +const params = new URLSearchParams({ metric: "gex", symbol }); +const response = await fetcher(backendApiUrl(HEATMAP_PATH, params), { + cache: "no-store", + headers: backendJsonHeaders(requestHeaders) +}); +``` + +Delete the old `sameOriginProxyUrl` and `proxyRequestHeaders` functions from this file. + +- [ ] **Step 4: Run focused frontend tests to verify GREEN** + +Run: + +```bash +pnpm --filter @gammascope/web test -- serverExperimentalAnalyticsSource.test.ts serverHeatmapSource.test.ts +``` + +Expected: PASS. + +- [ ] **Step 5: Run frontend typecheck** + +Run: + +```bash +pnpm typecheck:web +``` + +Expected: PASS. + +## Chunk 2: Backend Live Snapshot Service + +### Task 3: Add collector state revision support + +**Files:** +- Modify: `apps/api/gammascope_api/ingestion/collector_state.py` +- Test: `apps/api/tests/test_latest_state_cache.py` or new assertions in `apps/api/tests/test_live_snapshot_service.py` + +- [ ] **Step 1: Write a failing revision test** + +Add this in `apps/api/tests/test_live_snapshot_service.py` once the file exists, or in `test_latest_state_cache.py` temporarily. + +```py +def test_collector_state_revision_increments_on_ingest() -> None: + state = CollectorState() + assert state.revision() == 0 + + state.ingest(CollectorEvents.model_validate(_health_event("2026-04-24T15:30:00Z"))) + + assert state.revision() == 1 + assert state.snapshot()["revision"] == 1 +``` + +Use an existing `_health_event` helper from the nearest test file, or define the minimal helper locally. + +- [ ] **Step 2: Run test to verify RED** + +Run: + +```bash +PYTHONPATH=apps/api .venv/bin/pytest apps/api/tests/test_live_snapshot_service.py::test_collector_state_revision_increments_on_ingest -q +``` + +Expected: FAIL because `revision()` does not exist. + +- [ ] **Step 3: Implement revision support** + +In `CollectorState`: + +```py +def clear(self) -> None: + self._revision = 0 + self._health_events = {} + self._contracts = {} + self._underlying_ticks = {} + self._option_ticks = {} + self._last_event_time = None + +def ingest(self, event: CollectorEvents) -> str: + ... + self._revision += 1 + return event_type + +def revision(self) -> int: + return self._revision +``` + +Include revision in `summary()` and `snapshot()`: + +```py +"revision": self._revision, +``` + +Restore it in `from_snapshot()`: + +```py +state._revision = int(snapshot.get("revision") or 0) +``` + +- [ ] **Step 4: Run focused test** + +Run: + +```bash +PYTHONPATH=apps/api .venv/bin/pytest apps/api/tests/test_live_snapshot_service.py::test_collector_state_revision_increments_on_ingest -q +``` + +Expected: PASS. + +### Task 4: Add the live snapshot service/cache + +**Files:** +- Create: `apps/api/gammascope_api/ingestion/live_snapshot_service.py` +- Test: `apps/api/tests/test_live_snapshot_service.py` + +- [ ] **Step 1: Write failing service tests** + +Add these tests with local event helpers copied from `apps/api/tests/test_heatmap_route.py` or `apps/api/tests/test_contract_endpoints.py`. + +```py +from gammascope_api.contracts.generated.collector_events import CollectorEvents +from gammascope_api.ingestion.collector_state import CollectorState +from gammascope_api.ingestion.live_snapshot_service import LiveSnapshotService + + +def test_live_snapshot_service_caches_dashboard_snapshot_until_state_revision_changes(monkeypatch) -> None: + state = CollectorState() + for event in _spx_events(spot=5200.0, event_time="2026-04-24T15:30:01Z"): + state.ingest(CollectorEvents.model_validate(event)) + calls = 0 + + def fake_builder(input_state): + nonlocal calls + calls += 1 + return {"session_id": "moomoo-spx-0dte-live", "spot": input_state.latest_underlying_tick()["spot"]} + + monkeypatch.setattr( + "gammascope_api.ingestion.live_snapshot_service.build_spx_dashboard_live_snapshot", + fake_builder, + ) + + service = LiveSnapshotService(lambda: state) + + assert service.dashboard_snapshot()["spot"] == 5200.0 + assert service.dashboard_snapshot()["spot"] == 5200.0 + assert calls == 1 + + for event in _spx_events(spot=5210.0, event_time="2026-04-24T15:30:02Z"): + state.ingest(CollectorEvents.model_validate(event)) + + assert service.dashboard_snapshot()["spot"] == 5210.0 + assert calls == 2 +``` + +```py +def test_live_snapshot_service_returns_defensive_copies(monkeypatch) -> None: + state = CollectorState() + state.ingest(CollectorEvents.model_validate(_health_event("2026-04-24T15:30:00Z"))) + + monkeypatch.setattr( + "gammascope_api.ingestion.live_snapshot_service.build_spx_dashboard_live_snapshot", + lambda _: {"session_id": "moomoo-spx-0dte-live", "rows": []}, + ) + + service = LiveSnapshotService(lambda: state) + first = service.dashboard_snapshot() + first["rows"].append({"mutated": True}) + + assert service.dashboard_snapshot()["rows"] == [] +``` + +```py +def test_live_snapshot_service_maps_heatmap_symbols_to_live_sessions(monkeypatch) -> None: + state = CollectorState() + state.ingest(CollectorEvents.model_validate(_health_event("2026-04-24T15:30:00Z"))) + requested = [] + + def fake_build_live_snapshot(_state, *, session_id=None): + requested.append(session_id) + return {"session_id": session_id} + + monkeypatch.setattr( + "gammascope_api.ingestion.live_snapshot_service.build_live_snapshot", + fake_build_live_snapshot, + ) + + service = LiveSnapshotService(lambda: state) + + assert service.symbol_snapshot("SPY") == {"session_id": "moomoo-spy-0dte-live"} + assert requested == ["moomoo-spy-0dte-live"] +``` + +- [ ] **Step 2: Run tests to verify RED** + +Run: + +```bash +PYTHONPATH=apps/api .venv/bin/pytest apps/api/tests/test_live_snapshot_service.py -q +``` + +Expected: FAIL because `live_snapshot_service.py` does not exist yet. + +- [ ] **Step 3: Implement `live_snapshot_service.py`** + +Use this shape: + +```py +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 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__" + + +@dataclass(frozen=True) +class _CachedSnapshot: + state_revision: int + 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_revision = state.revision() + cached = self._cache.get(cache_key) + + if cached is None or cached.state_revision != state_revision: + cached = _CachedSnapshot(state_revision=state_revision, snapshot=builder(state)) + self._cache[cache_key] = cached + + return deepcopy(cached.snapshot) if cached.snapshot is not None else None + + +_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() +``` + +- [ ] **Step 4: Run focused service tests to verify GREEN** + +Run: + +```bash +PYTHONPATH=apps/api .venv/bin/pytest apps/api/tests/test_live_snapshot_service.py -q +``` + +Expected: PASS. + +### Task 5: Route FastAPI live readers through the service + +**Files:** +- Modify: `apps/api/gammascope_api/routes/snapshot.py` +- Modify: `apps/api/gammascope_api/routes/experimental.py` +- Modify: `apps/api/gammascope_api/routes/heatmap.py` +- Modify: `apps/api/gammascope_api/routes/scenario.py` +- Modify: `apps/api/gammascope_api/routes/stream.py` + +- [ ] **Step 1: Update `snapshot.py`** + +Replace: + +```py +live_snapshot = build_spx_dashboard_live_snapshot(cached_or_memory_collector_state()) +``` + +with: + +```py +live_snapshot = get_live_snapshot_service().dashboard_snapshot() +``` + +Remove unused imports for `cached_or_memory_collector_state` and `build_spx_dashboard_live_snapshot`. + +- [ ] **Step 2: Update `experimental.py`** + +Use: + +```py +live_snapshot = get_live_snapshot_service().dashboard_snapshot() +``` + +Keep `build_experimental_payload(live_snapshot, "latest")` unchanged. + +- [ ] **Step 3: Update `scenario.py`** + +Use: + +```py +live_snapshot = get_live_snapshot_service().dashboard_snapshot() +``` + +Keep scenario calculation behavior unchanged. + +- [ ] **Step 4: Update `stream.py`** + +In `_current_snapshot()`, use: + +```py +live_snapshot = get_live_snapshot_service().dashboard_snapshot() +``` + +Keep replay websocket behavior unchanged. + +- [ ] **Step 5: Update `heatmap.py`** + +Import `MOOMOO_LIVE_REPLAY_SESSION_IDS`, `HeatmapSymbol`, and `get_live_snapshot_service` from `live_snapshot_service`. + +Delete the local `HeatmapSymbol` literal and `MOOMOO_LIVE_REPLAY_SESSION_IDS` mapping from `heatmap.py`. + +Use: + +```py +live_snapshot = get_live_snapshot_service().symbol_snapshot(symbol) +``` + +Keep `_latest_moomoo_live_replay_snapshot(symbol)` as the Postgres fallback. + +- [ ] **Step 6: Run focused backend route tests** + +Run: + +```bash +PYTHONPATH=apps/api .venv/bin/pytest \ + apps/api/tests/test_live_snapshot_service.py \ + apps/api/tests/test_contract_endpoints.py \ + apps/api/tests/test_experimental_routes.py \ + apps/api/tests/test_heatmap_route.py \ + apps/api/tests/test_stream_endpoint.py \ + apps/api/tests/test_private_mode.py \ + -q +``` + +Expected: PASS. + +If any tests fail because cached snapshots persist between tests with intentionally reused event fixtures, call `reset_live_snapshot_service_override()` or `_default_live_snapshot_service.cache_clear()` from the relevant test fixture. Prefer adding a shared autouse fixture only if multiple files need it. + +## Chunk 3: Full Verification and Cleanup + +### Task 6: Run full project verification + +**Files:** +- No planned source edits. + +- [ ] **Step 1: Run frontend verification** + +Run: + +```bash +pnpm typecheck:web && pnpm test:web +``` + +Expected: PASS. + +- [ ] **Step 2: Run backend verification** + +Run: + +```bash +pnpm test:api +``` + +Expected: PASS. + +- [ ] **Step 3: Run collector tests if collector contracts changed unexpectedly** + +Run this only if edits touched collector event contracts or collector state snapshots consumed by collector tests: + +```bash +pnpm test:collector +``` + +Expected: PASS. + +- [ ] **Step 4: Browser smoke check** + +With API and web running: + +```bash +GAMMASCOPE_API_BASE_URL=http://127.0.0.1:8000 \ +NEXT_PUBLIC_GAMMASCOPE_WS_URL=ws://127.0.0.1:8000/ws/spx/0dte \ +pnpm dev:web +``` + +Check: + +- `http://localhost:3000/experimental` renders live or fallback experimental analytics without console errors. +- `http://localhost:3000/heatmap` renders all supported heatmap panels. +- `http://localhost:3000/` still receives live dashboard updates. + +Expected: pages render, no browser console errors, API logs show FastAPI requests from Next server without same-origin `/api/...` recursion for server initial loads. + +### Task 7: Final review checklist + +- [ ] Confirm browser/client calls still use same-origin Next proxy paths. +- [ ] Confirm server components fetch `GAMMASCOPE_API_BASE_URL` directly. +- [ ] Confirm FastAPI route behavior is unchanged externally. +- [ ] Confirm `source` contract naming is not changed in this plan; source-neutral Moomoo/IBKR ingestion is a separate larger migration. +- [ ] Confirm no unrelated dirty file changes were reverted, especially generated `apps/web/next-env.d.ts`. + +## Expected End State + +- Server-rendered `/experimental` and `/heatmap` no longer do a Next route handler hop before reaching FastAPI. +- FastAPI live readers share one live snapshot service/cache. +- The Moomoo and IBKR capabilities remain intact. +- Existing public API paths remain compatible: + - `GET /api/spx/0dte/snapshot/latest` + - `GET /api/spx/0dte/experimental/latest` + - `GET /api/spx/0dte/heatmap/latest` + - `POST /api/spx/0dte/scenario` + - `WS /ws/spx/0dte` From cb52e61a3b0a3f4436c1c03d7662ea8873f0316a Mon Sep 17 00:00:00 2001 From: Sakura <32687351+zifanzhou1024@users.noreply.github.com> Date: Thu, 30 Apr 2026 02:27:16 -0700 Subject: [PATCH 2/2] Label Moomoo dashboard data source --- apps/web/components/DashboardView.tsx | 63 +++++++++++++++++-------- apps/web/lib/dashboardMetrics.ts | 23 ++++++++- apps/web/next-env.d.ts | 5 +- apps/web/tests/DashboardView.test.tsx | 22 +++++++++ apps/web/tests/dashboardMetrics.test.ts | 17 +++++++ 5 files changed, 108 insertions(+), 22 deletions(-) diff --git a/apps/web/components/DashboardView.tsx b/apps/web/components/DashboardView.tsx index a4f6551..b1b2867 100644 --- a/apps/web/components/DashboardView.tsx +++ b/apps/web/components/DashboardView.tsx @@ -33,6 +33,8 @@ import { formatStatusLabel, formatStrikeRange, getAtmMetricValue, + getCollectorSourceDetail, + getCollectorSourceLabel, getRowOperationalStatusDisplays, getTransportStatusDisplay, getComparisonStatusDisplay, @@ -100,6 +102,8 @@ export function DashboardView({ const atmVanna = getAtmMetricValue(snapshot, "custom_vanna"); const transportDisplay = transportStatus ? getTransportStatusDisplay(transportStatus) : null; const operationalNotices = deriveOperationalNotices(snapshot, collectorHealth, transportStatus); + const collectorSourceLabel = getCollectorSourceLabel(collectorHealth); + const collectorSourceDetail = collectorHealth ? getCollectorSourceDetail(collectorHealth) : null; const handleInspectStrike = (strike: number) => setInspectedStrike(strike); const handleClearInspection = () => setInspectedStrike(null); @@ -161,7 +165,7 @@ export function DashboardView({ Collector {formatStatusLabel(collectorHealth.status)} - IBKR {formatAccountMode(collectorHealth.ibkr_account_mode)} + {collectorSourceDetail} {collectorHealth.message} @@ -284,17 +288,23 @@ export function DashboardView({ {inspection ? : null} - + ); } const OptionChainSection = React.memo(function OptionChainSection({ snapshot, - initialChainSide + initialChainSide, + comparisonSourceLabel }: { snapshot: AnalyticsSnapshot; initialChainSide: ChainSide; + comparisonSourceLabel: string; }) { const [chainSide, setChainSide] = useState(initialChainSide); const rows = sortRowsByStrike(snapshot.rows); @@ -368,8 +378,13 @@ const OptionChainSection = React.memo(function OptionChainSection({ {formatPrice(row.call?.bid)} {formatPrice(row.call?.ask)} {formatPrice(row.call?.mid)} - - + + ) : null} @@ -382,8 +397,13 @@ const OptionChainSection = React.memo(function OptionChainSection({ {formatPrice(row.put?.bid)} {formatPrice(row.put?.ask)} {formatPrice(row.put?.mid)} - - + + ) : null} @@ -654,21 +674,20 @@ function formatVannaFlipDetail(level: ReturnType["vannaF return formatMarketLevel(level, "decimal"); } -function formatAccountMode(accountMode: CollectorHealth["ibkr_account_mode"]): string { - if (accountMode === "unknown") { - return "Unknown"; - } - - return formatStatusLabel(accountMode); -} - -function IvCell({ row }: { row: AnalyticsSnapshot["rows"][number] | null | undefined }) { +function IvCell({ + row, + comparisonSourceLabel +}: { + row: AnalyticsSnapshot["rows"][number] | null | undefined; + comparisonSourceLabel: string; +}) { return ( {formatPercent(row?.custom_iv)} {formatNumber(gamma, 5)} - {status.label} + {statusLabel} ); } return ( - {ibkrValue !== "—" ? IBKR {ibkrValue} : null} + {ibkrValue !== "—" ? {sourceLabel} {ibkrValue} : null} {diffValue !== "—" ? {diffValue} : null} ); diff --git a/apps/web/lib/dashboardMetrics.ts b/apps/web/lib/dashboardMetrics.ts index 7ef8abe..8c7cd12 100644 --- a/apps/web/lib/dashboardMetrics.ts +++ b/apps/web/lib/dashboardMetrics.ts @@ -241,6 +241,27 @@ export function getTransportStatusDisplay(status: LiveTransportStatus): Operatio return { label: "Connecting", tone: "muted" }; } +export function getCollectorSourceLabel(collectorHealth?: CollectorHealth | null): string { + const collectorId = collectorHealth?.collector_id.toLowerCase() ?? ""; + const message = collectorHealth?.message.toLowerCase() ?? ""; + + if (collectorId.includes("moomoo") || message.includes("moomoo")) { + return "Moomoo"; + } + + return "IBKR"; +} + +export function getCollectorSourceDetail(collectorHealth: CollectorHealth): string { + const sourceLabel = getCollectorSourceLabel(collectorHealth); + + if (sourceLabel === "Moomoo" && collectorHealth.ibkr_account_mode === "unknown") { + return "Moomoo Source"; + } + + return `${sourceLabel} ${formatCollectorAccountMode(collectorHealth.ibkr_account_mode)}`; +} + export function getRowOperationalStatusDisplay(row: AnalyticsRow | null | undefined): OperationalStatusDisplay | null { return getRowOperationalStatusDisplays(row)[0] ?? null; } @@ -363,7 +384,7 @@ export function deriveDataQuality( collector: collectorHealth ? { label: `Collector ${formatStatusLabel(collectorHealth.status)}`, - detail: `IBKR ${formatCollectorAccountMode(collectorHealth.ibkr_account_mode)}`, + detail: getCollectorSourceDetail(collectorHealth), tone: collectorHealth.status === "connected" ? "ok" : collectorStatusTone(collectorHealth.status) } : null, diff --git a/apps/web/next-env.d.ts b/apps/web/next-env.d.ts index 268d608..830fb59 100644 --- a/apps/web/next-env.d.ts +++ b/apps/web/next-env.d.ts @@ -1,5 +1,6 @@ /// /// +/// -// This file is generated by Next.js. It is committed here so initial typechecks -// work before the dev server has been run. +// NOTE: This file should not be edited +// see https://nextjs.org/docs/app/api-reference/config/typescript for more information. diff --git a/apps/web/tests/DashboardView.test.tsx b/apps/web/tests/DashboardView.test.tsx index ae161ac..ed98b68 100644 --- a/apps/web/tests/DashboardView.test.tsx +++ b/apps/web/tests/DashboardView.test.tsx @@ -346,6 +346,28 @@ describe("DashboardView", () => { expect(markup).toContain("IBKR market data delayed"); }); + it("labels Moomoo compatibility collector data as Moomoo instead of IBKR", async () => { + const { DashboardView } = await import("../components/DashboardView"); + const collectorHealth = { + schema_version: "1.0.0", + source: "ibkr", + collector_id: "local-moomoo", + status: "connected", + ibkr_account_mode: "unknown", + message: "Moomoo compatibility snapshot emitted", + event_time: "2026-04-30T15:00:00Z", + received_time: "2026-04-30T15:00:01Z" + } satisfies CollectorHealth; + const markup = renderToStaticMarkup(); + + expect(markup).toContain("Moomoo Source"); + expect(markup).toContain("Moomoo compatibility snapshot emitted"); + expect(markup).toContain("Moomoo 17.95%"); + expect(markup).toContain("Moomoo 0.01986"); + expect(markup).not.toContain("IBKR Unknown"); + expect(markup).not.toContain("IBKR 17.95%"); + }); + it("renders transport status, operational notices, and row issue chips", async () => { const { DashboardView } = await import("../components/DashboardView"); const degradedSnapshot = { diff --git a/apps/web/tests/dashboardMetrics.test.ts b/apps/web/tests/dashboardMetrics.test.ts index f49707d..3fcfdc7 100644 --- a/apps/web/tests/dashboardMetrics.test.ts +++ b/apps/web/tests/dashboardMetrics.test.ts @@ -385,6 +385,23 @@ describe("dashboard metrics", () => { expect(getTransportStatusDisplay("reconnecting")).toEqual({ label: "Reconnecting", tone: "muted" }); }); + it("derives Moomoo collector detail from the compatibility collector id", () => { + expect(deriveDataQuality(seedSnapshot, { + schema_version: "1.0.0", + source: "ibkr", + collector_id: "local-moomoo", + status: "connected", + ibkr_account_mode: "unknown", + message: "Moomoo compatibility snapshot emitted", + event_time: "2026-04-30T15:00:00Z", + received_time: "2026-04-30T15:00:01Z" + }, null, "realtime").collector).toMatchObject({ + label: "Collector Connected", + detail: "Moomoo Source", + tone: "ok" + }); + }); + it("derives an explicit disconnected transport notice", () => { expect(deriveOperationalNotices({ ...seedSnapshot, coverage_status: "full" }, null, "disconnected")).toEqual([ {