From aa695edaf72b3cfcc610b8608cfab1a96cda4f88 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Tue, 30 Jun 2026 19:46:46 -0400 Subject: [PATCH 01/10] feat(observer): cheap catalog metadata via get_description + schema_map --- src/energex/observer/metadata.py | 71 +++++++++++++++++++++++++ src/energex/observer/routers/catalog.py | 24 +-------- src/energex/observer/schema_map.py | 33 ++++++++++++ tests/conftest.py | 16 ++++++ tests/test_observer_catalog.py | 47 ++++++++-------- tests/test_observer_metadata.py | 59 ++++++++++++++++++++ 6 files changed, 207 insertions(+), 43 deletions(-) create mode 100644 src/energex/observer/metadata.py create mode 100644 src/energex/observer/schema_map.py create mode 100644 tests/test_observer_metadata.py diff --git a/src/energex/observer/metadata.py b/src/energex/observer/metadata.py new file mode 100644 index 0000000..d571da5 --- /dev/null +++ b/src/energex/observer/metadata.py @@ -0,0 +1,71 @@ +"""Cheap catalog metadata: lib.get_description() (row_count + max valid_time, no data read) plus +the vintage sidecar for bitemporal symbols. Never reads full symbol data.""" + +from __future__ import annotations + +import logging + +from energex.core import symbology +from energex.observer.arctic import VINTAGE_SUFFIX, get_arctic +from energex.observer.schema_map import schema_for + +logger = logging.getLogger(__name__) + + +def _description(lib, symbol): + """(row_count, latest_valid_time) via ArcticDB get_description, no data read. + Attribute names verified against installed arcticdb: row_count (int) and + date_range (tuple of tz-naive Timestamps). Falls back to 0/None on error.""" + desc = lib.get_description(symbol) + row_count = int(getattr(desc, "row_count", 0) or 0) + date_range = getattr(desc, "date_range", (None, None)) + latest_valid_time = ( + date_range[1].isoformat() if date_range and date_range[1] is not None else None + ) + return row_count, latest_valid_time + + +def _vintage_meta(lib, symbol): + """(vintage_count, reconstructed_pct) for bitemporal symbols; (None, None) if no sidecar.""" + try: + v = lib.read(f"{symbol}{VINTAGE_SUFFIX}").data + except Exception: + return None, None + n = len(v) + pct = round(100.0 * float(v["vintage_reconstructed"].mean()), 1) if n else 0.0 + return n, pct + + +def _symbol_meta(lib, library, symbol): + row_count, latest_valid_time = _description(lib, symbol) + vintage_count, reconstructed_pct = _vintage_meta(lib, symbol) + schema = schema_for(library, symbol) + return { + "symbol": symbol, + "row_count": row_count, + "latest_valid_time": latest_valid_time, + "vintage_count": vintage_count, + "reconstructed_pct": reconstructed_pct, + "schema_name": schema.name if schema is not None else None, + } + + +def list_catalog() -> dict: + ac = get_arctic() + libraries = [] + for name in sorted(ac.list_libraries()): + lib = ac[name] + try: + mode = symbology.mode_for_library(name) + except Exception: + mode = "unknown" + syms = [s for s in lib.list_symbols() if not s.endswith(VINTAGE_SUFFIX)] + out, unreadable = [], 0 + for s in sorted(syms): + try: + out.append(_symbol_meta(lib, name, s)) + except Exception: + logger.warning("metadata: symbol %r in %r unreadable — skipping", s, name) + unreadable += 1 + libraries.append({"name": name, "mode": mode, "symbols": out, "unreadable": unreadable}) + return {"libraries": libraries} diff --git a/src/energex/observer/routers/catalog.py b/src/energex/observer/routers/catalog.py index fae8b37..f7f3498 100644 --- a/src/energex/observer/routers/catalog.py +++ b/src/energex/observer/routers/catalog.py @@ -1,33 +1,13 @@ from __future__ import annotations -import logging - from fastapi import APIRouter -from energex.observer.arctic import VINTAGE_SUFFIX, get_arctic +from energex.observer import metadata from energex.observer.auth import Role, require_role -logger = logging.getLogger(__name__) - router = APIRouter() @router.get("/catalog") def catalog(_claims: dict = require_role(Role.viewer)) -> dict: # noqa: B008 - ac = get_arctic() - out = [] - for name in sorted(ac.list_libraries()): - lib = ac[name] - syms = [s for s in lib.list_symbols() if not s.endswith(VINTAGE_SUFFIX)] - rows = 0 - unreadable = 0 - for s in syms: - try: - rows += len(lib.read(s).data) - except Exception: - logger.warning( - "catalog: could not read symbol %r in library %r — skipping", s, name - ) - unreadable += 1 - out.append({"name": name, "symbols": len(syms), "rows": rows, "unreadable": unreadable}) - return {"libraries": out} + return metadata.list_catalog() diff --git a/src/energex/observer/schema_map.py b/src/energex/observer/schema_map.py new file mode 100644 index 0000000..1af2cc6 --- /dev/null +++ b/src/energex/observer/schema_map.py @@ -0,0 +1,33 @@ +"""Library/symbol -> core.schemas routing. Mirrors orchestration/checks.py's gate assignments; +the constraint definitions live in core.schemas (single source of truth) — only routing is here.""" + +from __future__ import annotations + +import pandera as pa + +from energex.core import schemas + +_BY_LIBRARY: dict[str, pa.DataFrameSchema] = { + "prices.spot": schemas.FRED_SPOT, + "prices.intraday": schemas.OHLCV, + "prices.futures": schemas.DATED_CONTRACTS, + "weather": schemas.NOAA_HDDCDD, + "power.demand": schemas.POWER_REGION, + "power.demand_forecast": schemas.POWER_REGION, + "power.generation": schemas.POWER_REGION, + "power.interchange": schemas.POWER_REGION, + "power.generation_by_fuel": schemas.POWER_GEN_BY_FUEL, + "power.lmp": schemas.ERCOT_SPP, + "power.dalmp": schemas.ERCOT_SPP, + "power.load": schemas.ERCOT_LOAD, +} +_EIA_BY_SYMBOL: dict[str, pa.DataFrameSchema] = { + "ng_storage_lower48": schemas.EIA_GAS_STORAGE, + "pet_crude_stocks": schemas.EIA_PETROLEUM, +} + + +def schema_for(library: str, symbol: str) -> pa.DataFrameSchema | None: + if library == "fundamentals.eia": + return _EIA_BY_SYMBOL.get(symbol) + return _BY_LIBRARY.get(library) diff --git a/tests/conftest.py b/tests/conftest.py index c4aeae7..2cb542b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -114,3 +114,19 @@ def arctic_store(arctic_uri): def arctic_lib(arctic_store): """A single throwaway library; storage functions take the Library object directly.""" return arctic_store.create_library("phase2") + + +@pytest.fixture +def observer_arctic(arctic_store, arctic_uri, monkeypatch): + """LMDB-backed ArcticDB instance monkeypatched into energex.observer.arctic.get_arctic. + + Creates the ``power.load`` library used by observer metadata tests. + The test may create additional libraries via ``arctic_store`` if needed. + """ + monkeypatch.setenv("ENERGEX_ARCTIC_URI", arctic_uri) + arctic_store.create_library("power.load") + + from energex.observer.arctic import get_arctic + + get_arctic.cache_clear() + return arctic_store diff --git a/tests/test_observer_catalog.py b/tests/test_observer_catalog.py index 80fefe7..1fb0602 100644 --- a/tests/test_observer_catalog.py +++ b/tests/test_observer_catalog.py @@ -1,4 +1,4 @@ -"""/catalog lists libraries + per-library symbol/row counts from ArcticDB; viewer-gated.""" +"""/catalog lists libraries + per-symbol metadata from ArcticDB; viewer-gated.""" from __future__ import annotations @@ -25,11 +25,10 @@ def _hdr(role="viewer"): @pytest.fixture -def client(arctic_store, arctic_uri, monkeypatch): +def client(observer_arctic, monkeypatch): monkeypatch.setenv("OBSERVER_JWT_SECRET", SECRET) monkeypatch.setenv("OBSERVER_CORS_ORIGINS", "") - monkeypatch.setenv("ENERGEX_ARCTIC_URI", arctic_uri) - lib = arctic_store.create_library("power.lmp") + lib = observer_arctic.create_library("power.lmp") storage.commit_vintage( lib, "hb_houston", @@ -63,17 +62,22 @@ def test_catalog_lists_libraries(client): body = client.get("/catalog", headers=_hdr()).json() libs = {x["name"]: x for x in body["libraries"]} assert "power.lmp" in libs - assert libs["power.lmp"]["symbols"] == 1 and libs["power.lmp"]["rows"] == 1 - assert libs["power.lmp"]["unreadable"] == 0 + entry = libs["power.lmp"] + # symbols is now a list of objects, not a count + assert len(entry["symbols"]) == 1 + sym = entry["symbols"][0] + assert sym["symbol"] == "hb_houston" + assert sym["row_count"] == 1 + assert entry["unreadable"] == 0 + assert entry["mode"] == "bitemporal_merge" -def test_catalog_resilient_to_bad_symbol(arctic_store, arctic_uri, monkeypatch): +def test_catalog_resilient_to_bad_symbol(observer_arctic, arctic_uri, monkeypatch): """One corrupt symbol must not 500 the whole /catalog response.""" monkeypatch.setenv("OBSERVER_JWT_SECRET", SECRET) monkeypatch.setenv("OBSERVER_CORS_ORIGINS", "") - monkeypatch.setenv("ENERGEX_ARCTIC_URI", arctic_uri) - lib = arctic_store.create_library("power.bad") + lib = observer_arctic.create_library("power.bad") for sym in ("good_sym", "bad_sym"): storage.commit_vintage( lib, @@ -93,13 +97,13 @@ def test_catalog_resilient_to_bad_symbol(arctic_store, arctic_uri, monkeypatch): mode="bitemporal_merge", ) - # Patch the catalog router so bad_sym raises during the request - import energex.observer.routers.catalog as catalog_mod + # Patch metadata.list_catalog via its get_arctic dependency so bad_sym raises + import energex.observer.metadata as metadata_mod - _orig_get_arctic = catalog_mod.get_arctic + _orig_get_arctic = metadata_mod.get_arctic def _patched_get_arctic(): - ac = arctic_store + ac = observer_arctic class _PatchedArcticProxy: def list_libraries(self): @@ -108,14 +112,14 @@ def list_libraries(self): def __getitem__(self, name): lib_obj = ac[name] if name == "power.bad": - _real_read = lib_obj.read + _real_desc = lib_obj.get_description - def _bad_read(sym, *args, **kwargs): - if sym == "bad_sym": + def _bad_desc(s, *args, **kwargs): + if s == "bad_sym": raise RuntimeError("simulated corrupt symbol") - return _real_read(sym, *args, **kwargs) + return _real_desc(s, *args, **kwargs) - lib_obj.read = _bad_read + lib_obj.get_description = _bad_desc return lib_obj return _PatchedArcticProxy() @@ -123,7 +127,7 @@ def _bad_read(sym, *args, **kwargs): from energex.observer.arctic import get_arctic get_arctic.cache_clear() - monkeypatch.setattr(catalog_mod, "get_arctic", _patched_get_arctic) + monkeypatch.setattr(metadata_mod, "get_arctic", _patched_get_arctic) from energex.observer.app import create_app @@ -131,6 +135,7 @@ def _bad_read(sym, *args, **kwargs): assert resp.status_code == 200 libs = {x["name"]: x for x in resp.json()["libraries"]} entry = libs["power.bad"] - assert entry["symbols"] == 2 - assert entry["rows"] == 1 # only good_sym counted + # good_sym appears in symbols list; bad_sym is counted in unreadable + assert len(entry["symbols"]) == 1 + assert entry["symbols"][0]["symbol"] == "good_sym" assert entry["unreadable"] >= 1 diff --git a/tests/test_observer_metadata.py b/tests/test_observer_metadata.py new file mode 100644 index 0000000..7251c4c --- /dev/null +++ b/tests/test_observer_metadata.py @@ -0,0 +1,59 @@ +"""Tests for observer schema_map and metadata.list_catalog().""" + +from __future__ import annotations + +import datetime as dt + +import pandas as pd + +from energex.core import schemas +from energex.observer import schema_map + + +def test_schema_for_maps_known_libraries(): + assert schema_map.schema_for("power.load", "erco") is schemas.ERCOT_LOAD + assert schema_map.schema_for("prices.spot", "wti_spot") is schemas.FRED_SPOT + # fundamentals.eia splits by symbol + assert ( + schema_map.schema_for("fundamentals.eia", "ng_storage_lower48") is schemas.EIA_GAS_STORAGE + ) + assert schema_map.schema_for("fundamentals.eia", "pet_crude_stocks") is schemas.EIA_PETROLEUM + assert schema_map.schema_for("totally.unknown", "x") is None + + +def _seed_bitemporal(lib, symbol="erco"): + from energex.core import storage + + base = pd.DataFrame( + { + "instrument_id": ["ERCOT.LOAD"], + "valid_time": [pd.Timestamp("2026-06-01", tz="UTC")], + "value": [40000.0], + } + ).set_index(pd.DatetimeIndex([pd.Timestamp("2026-06-01")], name="Datetime")) + storage.commit_vintage( + lib, + symbol, + base, + as_of=dt.datetime(2026, 6, 2, tzinfo=dt.timezone.utc), + source="ercot", + source_url="x", + fetched_at=dt.datetime(2026, 6, 2, tzinfo=dt.timezone.utc), + mode="bitemporal_merge", + ) + + +def test_list_catalog_reports_cheap_metadata(observer_arctic): # fixture creates lib 'power.load' + lib = observer_arctic["power.load"] + _seed_bitemporal(lib) + from energex.observer import metadata + + cat = metadata.list_catalog() + powerload = next(lib for lib in cat["libraries"] if lib["name"] == "power.load") + assert powerload["mode"] == "bitemporal_merge" + sym = next(s for s in powerload["symbols"] if s["symbol"] == "erco") + assert sym["row_count"] == 1 + assert sym["vintage_count"] == 1 + assert sym["schema_name"] == "ERCOT_LOAD" + # the __vintages sidecar symbol must be excluded from the symbol list + assert all(not s["symbol"].endswith("__vintages") for s in powerload["symbols"]) From 8ce7c19e8cf8ce2d02ec87d3a8c8e52fe1890070 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Tue, 30 Jun 2026 19:51:15 -0400 Subject: [PATCH 02/10] fix(observer): skip vintage-sidecar read for degenerate symbols; test cleanup --- src/energex/observer/metadata.py | 9 ++++++--- tests/test_observer_catalog.py | 2 -- tests/test_observer_metadata.py | 1 + 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/src/energex/observer/metadata.py b/src/energex/observer/metadata.py index d571da5..c232ba3 100644 --- a/src/energex/observer/metadata.py +++ b/src/energex/observer/metadata.py @@ -36,9 +36,12 @@ def _vintage_meta(lib, symbol): return n, pct -def _symbol_meta(lib, library, symbol): +def _symbol_meta(lib, library, symbol, mode="unknown"): row_count, latest_valid_time = _description(lib, symbol) - vintage_count, reconstructed_pct = _vintage_meta(lib, symbol) + if "bitemporal" in mode: + vintage_count, reconstructed_pct = _vintage_meta(lib, symbol) + else: + vintage_count, reconstructed_pct = None, None schema = schema_for(library, symbol) return { "symbol": symbol, @@ -63,7 +66,7 @@ def list_catalog() -> dict: out, unreadable = [], 0 for s in sorted(syms): try: - out.append(_symbol_meta(lib, name, s)) + out.append(_symbol_meta(lib, name, s, mode=mode)) except Exception: logger.warning("metadata: symbol %r in %r unreadable — skipping", s, name) unreadable += 1 diff --git a/tests/test_observer_catalog.py b/tests/test_observer_catalog.py index 1fb0602..0cc8a02 100644 --- a/tests/test_observer_catalog.py +++ b/tests/test_observer_catalog.py @@ -100,8 +100,6 @@ def test_catalog_resilient_to_bad_symbol(observer_arctic, arctic_uri, monkeypatc # Patch metadata.list_catalog via its get_arctic dependency so bad_sym raises import energex.observer.metadata as metadata_mod - _orig_get_arctic = metadata_mod.get_arctic - def _patched_get_arctic(): ac = observer_arctic diff --git a/tests/test_observer_metadata.py b/tests/test_observer_metadata.py index 7251c4c..f1f552f 100644 --- a/tests/test_observer_metadata.py +++ b/tests/test_observer_metadata.py @@ -54,6 +54,7 @@ def test_list_catalog_reports_cheap_metadata(observer_arctic): # fixture create sym = next(s for s in powerload["symbols"] if s["symbol"] == "erco") assert sym["row_count"] == 1 assert sym["vintage_count"] == 1 + assert sym["reconstructed_pct"] == 0.0 assert sym["schema_name"] == "ERCOT_LOAD" # the __vintages sidecar symbol must be excluded from the symbol list assert all(not s["symbol"].endswith("__vintages") for s in powerload["symbols"]) From 2f740217cbb6370517a05f246f68066bbfac8368 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Tue, 30 Jun 2026 19:55:40 -0400 Subject: [PATCH 03/10] feat(observer): per-symbol series (point-in-time), schema, and vintages --- src/energex/observer/app.py | 3 +- src/energex/observer/routers/symbol.py | 88 ++++++++++++++ src/energex/observer/schema_map.py | 19 +++ tests/test_observer_symbol.py | 154 +++++++++++++++++++++++++ 4 files changed, 263 insertions(+), 1 deletion(-) create mode 100644 src/energex/observer/routers/symbol.py create mode 100644 tests/test_observer_symbol.py diff --git a/src/energex/observer/app.py b/src/energex/observer/app.py index a250a47..c71352e 100644 --- a/src/energex/observer/app.py +++ b/src/energex/observer/app.py @@ -11,7 +11,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware -from energex.observer.routers import catalog, meta +from energex.observer.routers import catalog, meta, symbol def create_app() -> FastAPI: @@ -29,6 +29,7 @@ def create_app() -> FastAPI: app.include_router(meta.router) # /healthz lives here (open) + /me, /admin/ping app.include_router(catalog.router) + app.include_router(symbol.router) return app diff --git a/src/energex/observer/routers/symbol.py b/src/energex/observer/routers/symbol.py new file mode 100644 index 0000000..49ff31c --- /dev/null +++ b/src/energex/observer/routers/symbol.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import datetime as dt + +from fastapi import APIRouter, HTTPException + +from energex.core import storage, symbology +from energex.observer.arctic import VINTAGE_SUFFIX, get_arctic +from energex.observer.auth import Role, require_role +from energex.observer.schema_map import describe_schema, schema_for + +router = APIRouter(prefix="/symbol/{library}/{symbol}") + + +def _parse(ts: str | None): + return dt.datetime.fromisoformat(ts.replace("Z", "+00:00")) if ts else None + + +def _lib_or_404(library: str): + ac = get_arctic() + if library not in ac.list_libraries(): + raise HTTPException(404, f"unknown library {library!r}") + return ac[library] + + +def _mode_for_library(library: str) -> str | None: + try: + return symbology.mode_for_library(library) + except Exception: + return None + + +@router.get("/series") +def series( + library: str, + symbol: str, + as_of: str | None = None, + start: str | None = None, + end: str | None = None, + _c: dict = require_role(Role.viewer), # noqa: B008 +) -> dict: + lib = _lib_or_404(library) + dr = (_parse(start), _parse(end)) if (start or end) else None + mode = _mode_for_library(library) + try: + df = storage.read_as_of(lib, symbol, as_of=_parse(as_of), date_range=dr, mode=mode) + except Exception as exc: + raise HTTPException(404, f"cannot read {symbol!r}: {exc}") from exc + out = df.reset_index() + for col in out.columns: + if str(out[col].dtype).startswith("datetime"): + out[col] = out[col].astype(str) + return {"library": library, "symbol": symbol, "rows": out.to_dict(orient="records")} + + +@router.get("/schema") +def schema( + library: str, + symbol: str, + _c: dict = require_role(Role.viewer), # noqa: B008 +) -> dict: + sch = schema_for(library, symbol) + if sch is None: + return { + "library": library, + "symbol": symbol, + "schema_name": None, + "columns": [], + "checks": [], + } + return {"library": library, "symbol": symbol, **describe_schema(sch)} + + +@router.get("/vintages") +def vintages( + library: str, + symbol: str, + _c: dict = require_role(Role.viewer), # noqa: B008 +) -> dict: + lib = _lib_or_404(library) + try: + v = lib.read(f"{symbol}{VINTAGE_SUFFIX}").data.reset_index(drop=True) + except Exception: + return {"library": library, "symbol": symbol, "vintages": []} # degenerate: no sidecar + for col in v.columns: + if str(v[col].dtype).startswith("datetime"): + v[col] = v[col].astype(str) + return {"library": library, "symbol": symbol, "vintages": v.to_dict(orient="records")} diff --git a/src/energex/observer/schema_map.py b/src/energex/observer/schema_map.py index 1af2cc6..de9c995 100644 --- a/src/energex/observer/schema_map.py +++ b/src/energex/observer/schema_map.py @@ -31,3 +31,22 @@ def schema_for(library: str, symbol: str) -> pa.DataFrameSchema | None: if library == "fundamentals.eia": return _EIA_BY_SYMBOL.get(symbol) return _BY_LIBRARY.get(library) + + +def describe_schema(schema: pa.DataFrameSchema) -> dict: + cols = [] + for cname, col in schema.columns.items(): + checks = [str(c) for c in (col.checks or [])] + cols.append( + { + "name": cname, + "dtype": str(col.dtype), + "nullable": bool(col.nullable), + "checks": checks, + } + ) + return { + "schema_name": schema.name, + "columns": cols, + "checks": [str(c) for c in (schema.checks or [])], + } diff --git a/tests/test_observer_symbol.py b/tests/test_observer_symbol.py new file mode 100644 index 0000000..adb9987 --- /dev/null +++ b/tests/test_observer_symbol.py @@ -0,0 +1,154 @@ +"""Per-symbol detail endpoints: /series (point-in-time), /schema, /vintages.""" + +from __future__ import annotations + +import datetime as dt +import time + +import jwt +import pandas as pd +import pytest +from fastapi.testclient import TestClient + +from energex.core import storage + +SECRET = "test-jwt-secret" + + +def _make_token(role="viewer"): + claims = { + "sub": "u1", + "exp": int(time.time()) + 3600, + "aud": "authenticated", + "user_role": role, + } + return jwt.encode(claims, SECRET, algorithm="HS256") + + +def _hdr(role="viewer"): + return {"Authorization": f"Bearer {_make_token(role)}"} + + +@pytest.fixture +def observer_client(observer_arctic, monkeypatch): + monkeypatch.setenv("OBSERVER_JWT_SECRET", SECRET) + monkeypatch.setenv("OBSERVER_CORS_ORIGINS", "") + from energex.observer.arctic import get_arctic + + get_arctic.cache_clear() + from energex.observer.app import create_app + + client = TestClient(create_app()) + client.headers.update(_hdr("viewer")) + return client + + +def _frame(value, vt="2026-06-01"): + idx = pd.DatetimeIndex([pd.Timestamp(vt)], name="Datetime") + return pd.DataFrame( + { + "instrument_id": ["ERCOT.LOAD"], + "valid_time": [pd.Timestamp(vt, tz="UTC")], + "value": [value], + }, + index=idx, + ) + + +def test_series_point_in_time(observer_client, observer_arctic): + lib = observer_arctic["power.load"] + storage.commit_vintage( + lib, + "erco", + _frame(40000.0), + as_of=dt.datetime(2026, 6, 2, tzinfo=dt.timezone.utc), + source="ercot", + source_url="x", + fetched_at=dt.datetime(2026, 6, 2, tzinfo=dt.timezone.utc), + mode="bitemporal_merge", + ) + storage.commit_vintage( + lib, + "erco", + _frame(41000.0), + as_of=dt.datetime(2026, 6, 5, tzinfo=dt.timezone.utc), + source="ercot", + source_url="x", + fetched_at=dt.datetime(2026, 6, 5, tzinfo=dt.timezone.utc), + mode="bitemporal_merge", + ) + # as_of between the two commits -> sees the first (40000), not the later revision + r = observer_client.get( + "/symbol/power.load/erco/series", params={"as_of": "2026-06-03T00:00:00Z"} + ) + assert r.status_code == 200 + rows = r.json()["rows"] + assert rows[-1]["value"] == 40000.0 + # latest (no as_of) -> sees the revision + r2 = observer_client.get("/symbol/power.load/erco/series") + assert r2.json()["rows"][-1]["value"] == 41000.0 + + +def test_series_requires_auth(observer_arctic, monkeypatch): + monkeypatch.setenv("OBSERVER_JWT_SECRET", SECRET) + from energex.observer.arctic import get_arctic + + get_arctic.cache_clear() + from energex.observer.app import create_app + + anon = TestClient(create_app()) + assert anon.get("/symbol/power.load/erco/series").status_code == 401 + + +def test_series_unknown_library_returns_404(observer_client): + r = observer_client.get("/symbol/no.such.lib/erco/series") + assert r.status_code == 404 + + +def test_schema_known_library(observer_client): + r = observer_client.get("/symbol/power.load/erco/schema") + assert r.status_code == 200 + body = r.json() + assert body["schema_name"] == "ERCOT_LOAD" + col_names = [c["name"] for c in body["columns"]] + assert "value" in col_names + + +def test_schema_unknown_library_returns_null(observer_client): + # unknown library -> schema_name is None + r = observer_client.get("/symbol/no.such.lib/erco/schema") + assert r.status_code == 200 + body = r.json() + assert body["schema_name"] is None + assert body["columns"] == [] + + +def test_vintages_seeded_symbol(observer_client, observer_arctic): + lib = observer_arctic["power.load"] + storage.commit_vintage( + lib, + "erco_vt", + _frame(99.0), + as_of=dt.datetime(2026, 6, 10, tzinfo=dt.timezone.utc), + source="ercot", + source_url="x", + fetched_at=dt.datetime(2026, 6, 10, tzinfo=dt.timezone.utc), + mode="bitemporal_merge", + ) + r = observer_client.get("/symbol/power.load/erco_vt/vintages") + assert r.status_code == 200 + body = r.json() + assert len(body["vintages"]) == 1 + + +def test_vintages_degenerate_symbol_returns_empty(observer_client, observer_arctic): + """A symbol with no vintage sidecar (degenerate) returns an empty list.""" + lib = observer_arctic["power.load"] + # write_bars requires a degenerate mode; use lib.write directly for a bare symbol + lib.write( + "bare_sym", + _frame(1.0), + ) + r = observer_client.get("/symbol/power.load/bare_sym/vintages") + assert r.status_code == 200 + assert r.json()["vintages"] == [] From f40c6d9a0babf19aa441c62f9cf8d486517d8640 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Tue, 30 Jun 2026 20:02:40 -0400 Subject: [PATCH 04/10] =?UTF-8?q?feat(observer):=20per-symbol=20quality=20?= =?UTF-8?q?=E2=80=94=20re-run=20the=20pandera=20gate=20+=20gaps=20+=20anom?= =?UTF-8?q?alies?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/energex/observer/quality_service.py | 71 +++++++++++++ src/energex/observer/routers/symbol.py | 12 +++ tests/test_observer_quality.py | 133 ++++++++++++++++++++++++ 3 files changed, 216 insertions(+) create mode 100644 src/energex/observer/quality_service.py create mode 100644 tests/test_observer_quality.py diff --git a/src/energex/observer/quality_service.py b/src/energex/observer/quality_service.py new file mode 100644 index 0000000..c999f20 --- /dev/null +++ b/src/energex/observer/quality_service.py @@ -0,0 +1,71 @@ +"""Per-symbol veracity: re-run the platform's pandera gate against stored data (the SAME definition +of broken the pipeline uses), plus valid_time gap counting and OHLCV anomaly summary. On-demand only. +""" + +from __future__ import annotations + +import datetime as dt + +import polars as pl + +from energex.analysis.quality import DataQualityChecker +from energex.core import quality, storage, symbology +from energex.core.exceptions import QualityGateError +from energex.observer.arctic import get_arctic +from energex.observer.schema_map import schema_for + +_OHLCV_COLS = {"Open", "High", "Low", "Close", "Volume"} + + +def _failures_to_list(failures) -> list[dict]: + try: + rows = failures.to_dict(orient="records") + except Exception: + return [{"check": str(failures)}] + return [ + { + "check": str(r.get("check", "")), + "column": str(r.get("column", "")), + "failure_case": str(r.get("failure_case", "")), + } + for r in rows + ][:50] + + +def symbol_quality(library: str, symbol: str, as_of: dt.datetime | None = None) -> dict: + lib = get_arctic()[library] + mode = symbology.mode_for_library(library) + df = storage.read_as_of(lib, symbol, as_of=as_of, mode=mode) + schema = schema_for(library, symbol) + result: dict = { + "library": library, + "symbol": symbol, + "schema_name": schema.name if schema else None, + "passed": None, + "failures": [], + "gaps": 0, + "anomalies": None, + } + if schema is not None: + try: + quality.validate(df.reset_index(), schema, as_of=dt.datetime.now(dt.timezone.utc)) + result["passed"] = True + except QualityGateError as exc: + result["passed"] = False + result["failures"] = _failures_to_list(exc.failures) + # valid_time gaps: count distinct missing steps at the modal cadence + if "valid_time" in df.columns and len(df) > 2: + vt = pl.Series(df["valid_time"].sort_values().to_numpy()) + deltas = vt.diff().drop_nulls() + if len(deltas): + modal = deltas.mode().to_list()[0] + result["gaps"] = int((deltas > modal).sum()) if modal else 0 + # OHLCV-only anomaly summary (requires Symbol and Datetime columns) + if _OHLCV_COLS.issubset(set(df.columns)): + try: + result["anomalies"] = DataQualityChecker( + pl.from_pandas(df.reset_index()) + ).check_tick_quality() + except Exception: + result["anomalies"] = None + return result diff --git a/src/energex/observer/routers/symbol.py b/src/energex/observer/routers/symbol.py index 49ff31c..630f27d 100644 --- a/src/energex/observer/routers/symbol.py +++ b/src/energex/observer/routers/symbol.py @@ -7,6 +7,7 @@ from energex.core import storage, symbology from energex.observer.arctic import VINTAGE_SUFFIX, get_arctic from energex.observer.auth import Role, require_role +from energex.observer.quality_service import symbol_quality from energex.observer.schema_map import describe_schema, schema_for router = APIRouter(prefix="/symbol/{library}/{symbol}") @@ -71,6 +72,17 @@ def schema( return {"library": library, "symbol": symbol, **describe_schema(sch)} +@router.get("/quality") +def quality_endpoint( + library: str, + symbol: str, + as_of: str | None = None, + _c: dict = require_role(Role.viewer), # noqa: B008 +) -> dict: + _lib_or_404(library) + return symbol_quality(library, symbol, as_of=_parse(as_of)) + + @router.get("/vintages") def vintages( library: str, diff --git a/tests/test_observer_quality.py b/tests/test_observer_quality.py new file mode 100644 index 0000000..0df4e32 --- /dev/null +++ b/tests/test_observer_quality.py @@ -0,0 +1,133 @@ +"""Veracity endpoint tests: /symbol/{library}/{symbol}/quality re-runs the pandera gate.""" + +from __future__ import annotations + +import datetime as dt +import time + +import jwt +import pandas as pd +import pytest +from fastapi.testclient import TestClient + +from energex.core import storage + +SECRET = "test-jwt-secret" + + +def _make_token(role="viewer"): + claims = { + "sub": "u1", + "exp": int(time.time()) + 3600, + "aud": "authenticated", + "user_role": role, + } + return jwt.encode(claims, SECRET, algorithm="HS256") + + +def _hdr(role="viewer"): + return {"Authorization": f"Bearer {_make_token(role)}"} + + +@pytest.fixture +def observer_client(observer_arctic, monkeypatch): + monkeypatch.setenv("OBSERVER_JWT_SECRET", SECRET) + monkeypatch.setenv("OBSERVER_CORS_ORIGINS", "") + from energex.observer.arctic import get_arctic + + get_arctic.cache_clear() + from energex.observer.app import create_app + + client = TestClient(create_app()) + client.headers.update(_hdr("viewer")) + return client + + +def _load_frame(value, vt): + idx = pd.DatetimeIndex([pd.Timestamp(vt)], name="Datetime") + return pd.DataFrame( + { + "instrument_id": ["ERCOT.LOAD"], + "valid_time": [pd.Timestamp(vt, tz="UTC")], + "value": [value], + }, + index=idx, + ) + + +def test_symbol_quality_passes_for_fresh_inband(observer_arctic): + lib = observer_arctic["power.load"] + today = dt.datetime.now(dt.timezone.utc) + storage.commit_vintage( + lib, + "fresh", + _load_frame(40000.0, today.date().isoformat()), + as_of=today, + source="ercot", + source_url="x", + fetched_at=today, + mode="bitemporal_merge", + ) + from energex.observer import quality_service + + res = quality_service.symbol_quality("power.load", "fresh", as_of=None) + assert res["passed"] is True + assert res["schema_name"] == "ERCOT_LOAD" + + +def test_symbol_quality_fails_for_stale(observer_arctic): + lib = observer_arctic["power.load"] + old = dt.datetime(2025, 1, 1, tzinfo=dt.timezone.utc) + storage.commit_vintage( + lib, + "stale", + _load_frame(40000.0, "2025-01-01"), + as_of=old, + source="ercot", + source_url="x", + fetched_at=old, + mode="bitemporal_merge", + ) + from energex.observer import quality_service + + # validated against now() — the 2025-01-01 valid_time will be stale + res = quality_service.symbol_quality("power.load", "stale", as_of=None) + assert res["passed"] is False + assert any("staler" in f["check"] for f in res["failures"]) + + +def test_quality_endpoint_returns_200(observer_client, observer_arctic): + lib = observer_arctic["power.load"] + today = dt.datetime.now(dt.timezone.utc) + storage.commit_vintage( + lib, + "ep_fresh", + _load_frame(40000.0, today.date().isoformat()), + as_of=today, + source="ercot", + source_url="x", + fetched_at=today, + mode="bitemporal_merge", + ) + r = observer_client.get("/symbol/power.load/ep_fresh/quality") + assert r.status_code == 200 + body = r.json() + assert body["passed"] is True + assert body["schema_name"] == "ERCOT_LOAD" + + +def test_quality_endpoint_requires_auth(observer_arctic, monkeypatch): + monkeypatch.setenv("OBSERVER_JWT_SECRET", SECRET) + monkeypatch.setenv("OBSERVER_CORS_ORIGINS", "") + from energex.observer.arctic import get_arctic + + get_arctic.cache_clear() + from energex.observer.app import create_app + + anon = TestClient(create_app()) + assert anon.get("/symbol/power.load/x/quality").status_code == 401 + + +def test_quality_unknown_library_returns_404(observer_client): + r = observer_client.get("/symbol/no.such.lib/x/quality") + assert r.status_code == 404 From 282e88afc56665dc3fefda581454bd0e6c9d93ac Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Tue, 30 Jun 2026 20:10:00 -0400 Subject: [PATCH 05/10] fix(observer): functional/honest OHLCV anomaly path + guard unmapped-library mode lookup --- src/energex/observer/quality_service.py | 15 ++++++---- src/energex/observer/routers/symbol.py | 6 +++- tests/test_observer_quality.py | 39 +++++++++++++++++++++++++ 3 files changed, 54 insertions(+), 6 deletions(-) diff --git a/src/energex/observer/quality_service.py b/src/energex/observer/quality_service.py index c999f20..916ab82 100644 --- a/src/energex/observer/quality_service.py +++ b/src/energex/observer/quality_service.py @@ -60,12 +60,17 @@ def symbol_quality(library: str, symbol: str, as_of: dt.datetime | None = None) if len(deltas): modal = deltas.mode().to_list()[0] result["gaps"] = int((deltas > modal).sum()) if modal else 0 - # OHLCV-only anomaly summary (requires Symbol and Datetime columns) + # OHLCV-only anomaly summary. + # The stored frame uses `instrument_id` (not `Symbol`) with a `Datetime` index. + # We adapt it: reset the index to materialise `Datetime`, rename `instrument_id` + # -> `Symbol`, and hand the Polars frame to DataQualityChecker. if _OHLCV_COLS.issubset(set(df.columns)): + flat = df.reset_index() # Datetime index -> column + if "instrument_id" in flat.columns: + flat = flat.rename(columns={"instrument_id": "Symbol"}) try: - result["anomalies"] = DataQualityChecker( - pl.from_pandas(df.reset_index()) - ).check_tick_quality() - except Exception: + result["anomalies"] = DataQualityChecker(pl.from_pandas(flat)).check_tick_quality() + except Exception as exc: result["anomalies"] = None + result["anomalies_note"] = f"anomaly check unavailable: {exc}" return result diff --git a/src/energex/observer/routers/symbol.py b/src/energex/observer/routers/symbol.py index 630f27d..48fe912 100644 --- a/src/energex/observer/routers/symbol.py +++ b/src/energex/observer/routers/symbol.py @@ -5,6 +5,7 @@ from fastapi import APIRouter, HTTPException from energex.core import storage, symbology +from energex.core.exceptions import SymbologyError from energex.observer.arctic import VINTAGE_SUFFIX, get_arctic from energex.observer.auth import Role, require_role from energex.observer.quality_service import symbol_quality @@ -80,7 +81,10 @@ def quality_endpoint( _c: dict = require_role(Role.viewer), # noqa: B008 ) -> dict: _lib_or_404(library) - return symbol_quality(library, symbol, as_of=_parse(as_of)) + try: + return symbol_quality(library, symbol, as_of=_parse(as_of)) + except SymbologyError as exc: + raise HTTPException(422, f"library {library!r} has no known revision mode") from exc @router.get("/vintages") diff --git a/tests/test_observer_quality.py b/tests/test_observer_quality.py index 0df4e32..616ae1a 100644 --- a/tests/test_observer_quality.py +++ b/tests/test_observer_quality.py @@ -131,3 +131,42 @@ def test_quality_endpoint_requires_auth(observer_arctic, monkeypatch): def test_quality_unknown_library_returns_404(observer_client): r = observer_client.get("/symbol/no.such.lib/x/quality") assert r.status_code == 404 + + +def _ohlcv_frame(symbol: str, n: int = 5) -> pd.DataFrame: + base = pd.Timestamp("2026-01-02 14:30:00", tz="UTC") + rows = [] + for i in range(n): + vt = base + pd.Timedelta(minutes=i) + px = 75.0 + i * 0.1 + rows.append( + { + "instrument_id": [symbol], + "valid_time": [vt], + "Open": [px], + "High": [px + 0.2], + "Low": [px - 0.2], + "Close": [px + 0.05], + "Volume": [1000 + i * 10], + } + ) + return pd.concat([pd.DataFrame(r) for r in rows], ignore_index=True) + + +def test_symbol_quality_ohlcv_anomalies_not_silent_noop(observer_arctic): + """OHLCV anomaly path adapts instrument_id->Symbol; result is a real dict, not None.""" + observer_arctic.create_library("prices.intraday") + lib = observer_arctic["prices.intraday"] + fetched_at = dt.datetime.now(dt.timezone.utc) + storage.write_bars( + lib, "CL_FRONT", _ohlcv_frame("CME.CL.FRONT"), fetched_at=fetched_at, mode="degenerate" + ) + + from energex.observer import quality_service + + res = quality_service.symbol_quality("prices.intraday", "CL_FRONT", as_of=None) + assert isinstance(res["anomalies"], dict), ( + f"expected dict, got: {res.get('anomalies_note', res['anomalies'])}" + ) + assert "total_records" in res["anomalies"] + assert res["anomalies"]["total_records"] == 5 From e160b2493c5b028bce5e9acc3fc58557a671b3d8 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Tue, 30 Jun 2026 20:13:50 -0400 Subject: [PATCH 06/10] feat(observer): 4V metrics overview + per-symbol freshness health cache --- src/energex/observer/app.py | 3 +- src/energex/observer/health.py | 97 +++++++++++ src/energex/observer/metrics.py | 46 +++++ src/energex/observer/routers/metrics.py | 20 +++ tests/test_observer_metrics.py | 217 ++++++++++++++++++++++++ 5 files changed, 382 insertions(+), 1 deletion(-) create mode 100644 src/energex/observer/health.py create mode 100644 src/energex/observer/metrics.py create mode 100644 src/energex/observer/routers/metrics.py create mode 100644 tests/test_observer_metrics.py diff --git a/src/energex/observer/app.py b/src/energex/observer/app.py index c71352e..3340c30 100644 --- a/src/energex/observer/app.py +++ b/src/energex/observer/app.py @@ -11,7 +11,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware -from energex.observer.routers import catalog, meta, symbol +from energex.observer.routers import catalog, meta, metrics, symbol def create_app() -> FastAPI: @@ -30,6 +30,7 @@ def create_app() -> FastAPI: app.include_router(meta.router) # /healthz lives here (open) + /me, /admin/ping app.include_router(catalog.router) app.include_router(symbol.router) + app.include_router(metrics.router) return app diff --git a/src/energex/observer/health.py b/src/energex/observer/health.py new file mode 100644 index 0000000..3e50759 --- /dev/null +++ b/src/energex/observer/health.py @@ -0,0 +1,97 @@ +"""Per-symbol freshness heuristic, cached with a TTL. Cheap: max valid_time (get_description) vs a +per-schema business-day tolerance mirroring schemas.py. NOT the full gate (that's quality_service).""" + +from __future__ import annotations + +import datetime as dt +import os +import time + +import numpy as np + +from energex.core.schemas import _EIA_FRESHNESS_DAYS, _FRED_FRESHNESS_DAYS +from energex.observer.arctic import get_arctic +from energex.observer.metadata import _description +from energex.observer.schema_map import schema_for + +# Mirror of schemas.py freshness days; a test cross-checks the importable ones. +_FRESHNESS_DAYS: dict[str, int] = { + "OHLCV": 2, + "DATED_CONTRACTS": 5, + "EIA_GAS_STORAGE": _EIA_FRESHNESS_DAYS, + "EIA_PETROLEUM": _EIA_FRESHNESS_DAYS, + "FRED_SPOT": _FRED_FRESHNESS_DAYS, + "ERCOT_SPP": 2, + "ERCOT_LOAD": 2, + "POWER_REGION": 2, + "POWER_GEN_BY_FUEL": 2, + "NOAA_HDDCDD": 45, +} + +_TTL: int = int(os.environ.get("OBSERVER_HEALTH_TTL_SECONDS", "300")) + +# (library, symbol) -> (monotonic_timestamp, row_dict) +_cache: dict[tuple[str, str], tuple[float, dict]] = {} + + +def _business_days_since(latest_iso: str | None) -> int | None: + if not latest_iso: + return None + latest = dt.datetime.fromisoformat(latest_iso) + now = dt.datetime.now(dt.timezone.utc) + if latest.tzinfo is None: + latest = latest.replace(tzinfo=dt.timezone.utc) + return int(np.busday_count(latest.date(), now.date())) + + +def _compute(library: str, symbol: str) -> dict: + lib = get_arctic()[library] + schema = schema_for(library, symbol) + try: + row_count, latest_valid_time = _description(lib, symbol) + except Exception: + return { + "symbol": symbol, + "freshness_status": "error", + "age_days": None, + "latest_valid_time": None, + "row_count": None, + "vintage_count": None, + "reconstructed_pct": None, + } + age = _business_days_since(latest_valid_time) + tol = _FRESHNESS_DAYS.get(schema.name) if schema else None + if age is None: + status = "error" + elif tol is not None and age > tol: + status = "stale" + else: + status = "ok" + vc = rp = None + try: + from energex.observer.arctic import VINTAGE_SUFFIX + + v = lib.read(f"{symbol}{VINTAGE_SUFFIX}").data + vc = len(v) + rp = round(100.0 * float(v["vintage_reconstructed"].mean()), 1) if vc else 0.0 + except Exception: + pass + return { + "symbol": symbol, + "freshness_status": status, + "age_days": age, + "latest_valid_time": latest_valid_time, + "row_count": row_count, + "vintage_count": vc, + "reconstructed_pct": rp, + } + + +def health_row(library: str, symbol: str) -> dict: + key = (library, symbol) + hit = _cache.get(key) + if hit and (time.monotonic() - hit[0]) < _TTL: + return hit[1] + row = _compute(library, symbol) + _cache[key] = (time.monotonic(), row) + return row diff --git a/src/energex/observer/metrics.py b/src/energex/observer/metrics.py new file mode 100644 index 0000000..68436b8 --- /dev/null +++ b/src/energex/observer/metrics.py @@ -0,0 +1,46 @@ +"""4V metrics: volume, velocity, variety, veracity over the full catalog.""" + +from __future__ import annotations + +from energex.observer import health, metadata + + +def health_rows() -> list[dict]: + rows = [] + for lib in metadata.list_catalog()["libraries"]: + for s in lib["symbols"]: + r = health.health_row(lib["name"], s["symbol"]) + rows.append({"library": lib["name"], **r, "schema_name": s["schema_name"]}) + return rows + + +def overview() -> dict: + cat = metadata.list_catalog() + rows = health_rows() + schemas_seen = { + s["schema_name"] for lib in cat["libraries"] for s in lib["symbols"] if s["schema_name"] + } + modes = {lib["mode"] for lib in cat["libraries"]} + stale = [r for r in rows if r["freshness_status"] in ("stale", "error")] + return { + "volume": { + "libraries": len(cat["libraries"]), + "symbols": sum(len(lib["symbols"]) for lib in cat["libraries"]), + "rows": sum((s["row_count"] or 0) for lib in cat["libraries"] for s in lib["symbols"]), + }, + "velocity": { + "ok": sum(1 for r in rows if r["freshness_status"] == "ok"), + "stale": sum(1 for r in rows if r["freshness_status"] == "stale"), + "error": sum(1 for r in rows if r["freshness_status"] == "error"), + }, + "variety": { + "schemas": len(schemas_seen), + "revision_modes": sorted(modes), + }, + "veracity": { + "broken": len(stale), + "broken_symbols": [{"library": r["library"], "symbol": r["symbol"]} for r in stale][ + :50 + ], + }, + } diff --git a/src/energex/observer/routers/metrics.py b/src/energex/observer/routers/metrics.py new file mode 100644 index 0000000..ae6f511 --- /dev/null +++ b/src/energex/observer/routers/metrics.py @@ -0,0 +1,20 @@ +"""GET /metrics/overview and GET /metrics/health — viewer-gated 4V metrics.""" + +from __future__ import annotations + +from fastapi import APIRouter + +from energex.observer import metrics +from energex.observer.auth import Role, require_role + +router = APIRouter(prefix="/metrics") + + +@router.get("/overview") +def overview(_c: dict = require_role(Role.viewer)) -> dict: # noqa: B008 + return metrics.overview() + + +@router.get("/health") +def health_endpoint(_c: dict = require_role(Role.viewer)) -> dict: # noqa: B008 + return {"rows": metrics.health_rows()} diff --git a/tests/test_observer_metrics.py b/tests/test_observer_metrics.py new file mode 100644 index 0000000..d0e6ba7 --- /dev/null +++ b/tests/test_observer_metrics.py @@ -0,0 +1,217 @@ +"""4V health layer: per-symbol freshness + metrics overview + /metrics/* endpoints.""" + +from __future__ import annotations + +import datetime as dt +import time + +import jwt +import pandas as pd +import pytest +from fastapi.testclient import TestClient + +from energex.core import storage + +SECRET = "test-jwt-secret" + + +def _make_token(role="viewer"): + claims = { + "sub": "u1", + "exp": int(time.time()) + 3600, + "aud": "authenticated", + "user_role": role, + } + return jwt.encode(claims, SECRET, algorithm="HS256") + + +def _hdr(role="viewer"): + return {"Authorization": f"Bearer {_make_token(role)}"} + + +def _f(vt): + idx = pd.DatetimeIndex([pd.Timestamp(vt)], name="Datetime") + return pd.DataFrame( + { + "instrument_id": ["ERCOT.LOAD"], + "valid_time": [pd.Timestamp(vt, tz="UTC")], + "value": [40000.0], + }, + index=idx, + ) + + +@pytest.fixture +def metrics_client(observer_arctic, monkeypatch): + monkeypatch.setenv("OBSERVER_JWT_SECRET", SECRET) + monkeypatch.setenv("OBSERVER_CORS_ORIGINS", "") + from energex.observer.arctic import get_arctic + + get_arctic.cache_clear() + from energex.observer.app import create_app + + return TestClient(create_app()) + + +# ── unit-level health tests ──────────────────────────────────────────────────── + + +def test_health_flags_stale(observer_arctic): + lib = observer_arctic["power.load"] + now = dt.datetime.now(dt.timezone.utc) + storage.commit_vintage( + lib, + "fresh", + _f(now.date().isoformat()), + as_of=now, + source="e", + source_url="x", + fetched_at=now, + mode="bitemporal_merge", + ) + storage.commit_vintage( + lib, + "stale", + _f("2025-01-01"), + as_of=now, + source="e", + source_url="x", + fetched_at=now, + mode="bitemporal_merge", + ) + # Clear cache so we get fresh reads. + from energex.observer import health + + health._cache.clear() + assert health.health_row("power.load", "fresh")["freshness_status"] == "ok" + assert health.health_row("power.load", "stale")["freshness_status"] == "stale" + + +def test_health_cache_returns_same_object(observer_arctic): + """Within TTL, the same dict object is returned without re-reading the store.""" + lib = observer_arctic["power.load"] + now = dt.datetime.now(dt.timezone.utc) + storage.commit_vintage( + lib, + "cached_sym", + _f(now.date().isoformat()), + as_of=now, + source="e", + source_url="x", + fetched_at=now, + mode="bitemporal_merge", + ) + from energex.observer import health + + health._cache.clear() + row1 = health.health_row("power.load", "cached_sym") + row2 = health.health_row("power.load", "cached_sym") + assert row1 is row2 + + +def test_health_error_for_missing_symbol(observer_arctic): + """A symbol that has never been written should return freshness_status='error'.""" + from energex.observer import health + + health._cache.clear() + row = health.health_row("power.load", "does_not_exist") + assert row["freshness_status"] == "error" + assert row["latest_valid_time"] is None + + +# ── freshness mirror cross-check ─────────────────────────────────────────────── + + +def test_freshness_mirror_matches_importable_constants(): + """Guard against the in-module mirror silently drifting from schemas.py constants.""" + from energex.core.schemas import _EIA_FRESHNESS_DAYS, _FRED_FRESHNESS_DAYS + from energex.observer.health import _FRESHNESS_DAYS + + assert _FRESHNESS_DAYS["EIA_GAS_STORAGE"] == _EIA_FRESHNESS_DAYS + assert _FRESHNESS_DAYS["EIA_PETROLEUM"] == _EIA_FRESHNESS_DAYS + assert _FRESHNESS_DAYS["FRED_SPOT"] == _FRED_FRESHNESS_DAYS + + +# ── metrics.overview() unit ──────────────────────────────────────────────────── + + +def test_overview_counts(observer_arctic): + from energex.observer import health, metrics + + health._cache.clear() + ov = metrics.overview() + assert ov["volume"]["libraries"] >= 1 + assert set(ov.keys()) == {"volume", "velocity", "variety", "veracity"} + # velocity keys present + assert {"ok", "stale", "error"} == set(ov["velocity"].keys()) + # veracity key + assert "broken" in ov["veracity"] + + +def test_health_rows_shape(observer_arctic): + lib = observer_arctic["power.load"] + now = dt.datetime.now(dt.timezone.utc) + storage.commit_vintage( + lib, + "hr_sym", + _f(now.date().isoformat()), + as_of=now, + source="e", + source_url="x", + fetched_at=now, + mode="bitemporal_merge", + ) + from energex.observer import health, metrics + + health._cache.clear() + rows = metrics.health_rows() + assert len(rows) >= 1 + row = next(r for r in rows if r["symbol"] == "hr_sym") + assert row["library"] == "power.load" + assert row["freshness_status"] in ("ok", "stale", "error") + + +# ── HTTP endpoint tests ──────────────────────────────────────────────────────── + + +def test_overview_endpoint_viewer_200(metrics_client, observer_arctic): + from energex.observer import health + + health._cache.clear() + r = metrics_client.get("/metrics/overview", headers=_hdr("viewer")) + assert r.status_code == 200 + body = r.json() + assert set(body.keys()) == {"volume", "velocity", "variety", "veracity"} + + +def test_overview_endpoint_anon_401(observer_arctic, monkeypatch): + monkeypatch.setenv("OBSERVER_JWT_SECRET", SECRET) + monkeypatch.setenv("OBSERVER_CORS_ORIGINS", "") + from energex.observer.arctic import get_arctic + + get_arctic.cache_clear() + from energex.observer.app import create_app + + anon = TestClient(create_app()) + assert anon.get("/metrics/overview").status_code == 401 + + +def test_health_endpoint_viewer_200(metrics_client, observer_arctic): + from energex.observer import health + + health._cache.clear() + r = metrics_client.get("/metrics/health", headers=_hdr("viewer")) + assert r.status_code == 200 + assert "rows" in r.json() + + +def test_health_endpoint_anon_401(observer_arctic, monkeypatch): + monkeypatch.setenv("OBSERVER_JWT_SECRET", SECRET) + monkeypatch.setenv("OBSERVER_CORS_ORIGINS", "") + from energex.observer.arctic import get_arctic + + get_arctic.cache_clear() + from energex.observer.app import create_app + + anon = TestClient(create_app()) + assert anon.get("/metrics/health").status_code == 401 From 970feb493eeb1d38af062152fb4ed22238d8e2b9 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Tue, 30 Jun 2026 20:21:17 -0400 Subject: [PATCH 07/10] feat(observer-web): 4V Health dashboard (tiles, freshness heatmap, broken-data rail) --- observer-web/package-lock.json | 32 ++++ observer-web/package.json | 1 + observer-web/src/app/(app)/page.tsx | 105 ++++++++----- .../components/__tests__/broken-rail.test.tsx | 15 ++ .../__tests__/four-v-tiles.test.tsx | 57 +++++++ observer-web/src/components/broken-rail.tsx | 32 ++++ observer-web/src/components/four-v-tiles.tsx | 144 ++++++++++++++++++ .../src/components/freshness-heatmap.tsx | 64 ++++++++ observer-web/src/lib/api.ts | 19 +++ 9 files changed, 427 insertions(+), 42 deletions(-) create mode 100644 observer-web/src/components/__tests__/broken-rail.test.tsx create mode 100644 observer-web/src/components/__tests__/four-v-tiles.test.tsx create mode 100644 observer-web/src/components/broken-rail.tsx create mode 100644 observer-web/src/components/four-v-tiles.tsx create mode 100644 observer-web/src/components/freshness-heatmap.tsx diff --git a/observer-web/package-lock.json b/observer-web/package-lock.json index d4ce2f7..b7fa849 100644 --- a/observer-web/package-lock.json +++ b/observer-web/package-lock.json @@ -10,6 +10,7 @@ "dependencies": { "@supabase/ssr": "^0.12.0", "@supabase/supabase-js": "^2.109.0", + "echarts": "^6.1.0", "next": "16.2.9", "react": "19.2.4", "react-dom": "19.2.4" @@ -3927,6 +3928,22 @@ "node": ">= 0.4" } }, + "node_modules/echarts": { + "version": "6.1.0", + "resolved": "https://registry.npmjs.org/echarts/-/echarts-6.1.0.tgz", + "integrity": "sha512-q0yaFPggC9FUdsWH4blavRWFmxdrIodbkoKNAjJudAI6CA9gNPxHtV2RcZNEepZVlk4yvBYkOkbk6HIVpIyHZA==", + "license": "Apache-2.0", + "dependencies": { + "tslib": "2.3.0", + "zrender": "6.1.0" + } + }, + "node_modules/echarts/node_modules/tslib": { + "version": "2.3.0", + "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.3.0.tgz", + "integrity": "sha512-N82ooyxVNm6h1riLCoyS9e3fuJ3AMG2zIZs2Gd1ATcSFjSA23Q0fzjjZeh0jbJvWVDZ0cJT8yaNNaaXHzueNjg==", + "license": "0BSD" + }, "node_modules/electron-to-chromium": { "version": "1.5.383", "resolved": "https://registry.npmjs.org/electron-to-chromium/-/electron-to-chromium-1.5.383.tgz", @@ -8537,6 +8554,21 @@ "peerDependencies": { "zod": "^3.25.0 || ^4.0.0" } + }, + "node_modules/zrender": { + "version": "6.1.0", + "resolved": "https://registry.npmjs.org/zrender/-/zrender-6.1.0.tgz", + "integrity": "sha512-oEGMDB6pOP2S6OwRR4PdVv610zrjnA3Bh+JnSG12fYJlBKjtNAoEb5fSUoCOOINlH96I2fU38/A2UpRKs67xYQ==", + "license": "BSD-3-Clause", + "dependencies": { + "tslib": "2.3.0" + } + }, + "node_modules/zrender/node_modules/tslib": { + "version": "2.3.0", + "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.3.0.tgz", + "integrity": "sha512-N82ooyxVNm6h1riLCoyS9e3fuJ3AMG2zIZs2Gd1ATcSFjSA23Q0fzjjZeh0jbJvWVDZ0cJT8yaNNaaXHzueNjg==", + "license": "0BSD" } } } diff --git a/observer-web/package.json b/observer-web/package.json index 697ad24..d919739 100644 --- a/observer-web/package.json +++ b/observer-web/package.json @@ -12,6 +12,7 @@ "dependencies": { "@supabase/ssr": "^0.12.0", "@supabase/supabase-js": "^2.109.0", + "echarts": "^6.1.0", "next": "16.2.9", "react": "19.2.4", "react-dom": "19.2.4" diff --git a/observer-web/src/app/(app)/page.tsx b/observer-web/src/app/(app)/page.tsx index 006c78c..045ba1e 100644 --- a/observer-web/src/app/(app)/page.tsx +++ b/observer-web/src/app/(app)/page.tsx @@ -1,61 +1,82 @@ import { apiFetch } from "@/lib/api"; +import type { OverviewMetrics, HealthRow } from "@/lib/api"; +import { FourVTiles } from "@/components/four-v-tiles"; +import { FreshnessHeatmap } from "@/components/freshness-heatmap"; +import { BrokenRail } from "@/components/broken-rail"; -interface Library { - name: string; - symbols: number; - rows: number; - unreadable: number; +async function getOverview(): Promise { + try { + return await apiFetch("/metrics/overview"); + } catch (err) { + console.error("[OverviewPage] getOverview failed:", err); + const msg = err instanceof Error ? err.message : String(err); + return { error: msg }; + } } -async function getCatalog(): Promise<{ libraries: Library[] } | { error: string } | null> { +async function getHealth(): Promise<{ rows: HealthRow[] } | { error: string } | null> { try { - return await apiFetch<{ libraries: Library[] }>("/catalog"); + return await apiFetch<{ rows: HealthRow[] }>("/metrics/health"); } catch (err) { - console.error("[OverviewPage] getCatalog failed:", err); + console.error("[OverviewPage] getHealth failed:", err); const msg = err instanceof Error ? err.message : String(err); return { error: msg }; } } +function isAuthError(msg: string): boolean { + return /: 40[13]/.test(msg); +} + +function ErrorBanner({ message }: { message: string }) { + const authError = isAuthError(message); + return ( +
+

+ {authError + ? "Couldn't load data — you may not have access. Try signing in again." + : "Couldn't load data — confirm observer-api is running and that you're signed in with access."} +

+
+ ); +} + export default async function OverviewPage() { - const data = await getCatalog(); + const [overview, health] = await Promise.all([getOverview(), getHealth()]); + + const overviewError = overview == null || "error" in overview; + const healthError = health == null || "error" in health; return (

Overview

-
-

Data Libraries

- {data == null || "error" in data ? ( -

- {data != null && "error" in data && /: 40[13]/.test(data.error) - ? "Couldn't load the catalog — you may not have access. Try signing in again." - : "Couldn't load the catalog — confirm observer-api is running and that you're signed in with access."} -

- ) : data.libraries.length === 0 ? ( -

No libraries found.

- ) : ( - - - - - - - - - - - {data.libraries.map((lib) => ( - - - - - - - ))} - -
LibrarySymbolsRowsUnreadable
{lib.name}{lib.symbols}{lib.rows.toLocaleString()}{lib.unreadable}
- )} -
+ + {overviewError ? ( + + ) : ( + + )} + + {healthError ? ( + + ) : ( + <> + + r.freshness_status !== "ok") + .map((r) => ({ library: r.library, symbol: r.symbol })) + : overview.veracity.broken_symbols + } + /> + + )}
); } diff --git a/observer-web/src/components/__tests__/broken-rail.test.tsx b/observer-web/src/components/__tests__/broken-rail.test.tsx new file mode 100644 index 0000000..6c0863b --- /dev/null +++ b/observer-web/src/components/__tests__/broken-rail.test.tsx @@ -0,0 +1,15 @@ +import { render, screen } from "@testing-library/react"; +import { describe, it, expect } from "vitest"; +import { BrokenRail } from "../broken-rail"; + +describe("BrokenRail", () => { + it("lists broken symbols", () => { + render(); + expect(screen.getByText(/power\.load/)).toBeInTheDocument(); + expect(screen.getByText("erco")).toBeInTheDocument(); + }); + it("shows an all-clear when empty", () => { + render(); + expect(screen.getByText(/no broken data/i)).toBeInTheDocument(); + }); +}); diff --git a/observer-web/src/components/__tests__/four-v-tiles.test.tsx b/observer-web/src/components/__tests__/four-v-tiles.test.tsx new file mode 100644 index 0000000..869d3db --- /dev/null +++ b/observer-web/src/components/__tests__/four-v-tiles.test.tsx @@ -0,0 +1,57 @@ +import { render, screen } from "@testing-library/react"; +import { describe, it, expect, vi } from "vitest"; + +// Mock echarts to avoid canvas errors in jsdom +const chartInstance = { setOption: vi.fn(), resize: vi.fn(), dispose: vi.fn() }; +vi.mock("echarts/core", () => ({ + use: vi.fn(), + init: vi.fn(() => chartInstance), +})); +vi.mock("echarts/renderers", () => ({ CanvasRenderer: {} })); +vi.mock("echarts/charts", () => ({ PieChart: {} })); +vi.mock("echarts/components", () => ({ + TooltipComponent: {}, + LegendComponent: {}, +})); +vi.mock("echarts", () => ({ + use: vi.fn(), + init: vi.fn(() => chartInstance), +})); + +import { FourVTiles } from "../four-v-tiles"; +import type { OverviewMetrics } from "@/lib/api"; + +const metrics: OverviewMetrics = { + volume: { libraries: 3, symbols: 42, rows: 1_500_000 }, + velocity: { ok: 38, stale: 3, error: 1 }, + variety: { schemas: 5, revision_modes: ["snapshot", "append"] }, + veracity: { broken: 4, broken_symbols: [] }, +}; + +describe("FourVTiles", () => { + it("renders volume stats", () => { + render(); + expect(screen.getByText("Volume")).toBeInTheDocument(); + // libraries=3 appears, may also match stale=3 — use getAllByText + expect(screen.getAllByText("3").length).toBeGreaterThanOrEqual(1); + expect(screen.getByText("42")).toBeInTheDocument(); // symbols + }); + + it("renders variety info", () => { + render(); + expect(screen.getByText("Variety")).toBeInTheDocument(); + expect(screen.getByText("5")).toBeInTheDocument(); // schemas + }); + + it("renders veracity broken count", () => { + render(); + expect(screen.getByText("Veracity")).toBeInTheDocument(); + expect(screen.getByText("4")).toBeInTheDocument(); // broken + }); + + it("renders velocity section", () => { + render(); + expect(screen.getByText("Velocity")).toBeInTheDocument(); + expect(screen.getByText("38")).toBeInTheDocument(); // ok count + }); +}); diff --git a/observer-web/src/components/broken-rail.tsx b/observer-web/src/components/broken-rail.tsx new file mode 100644 index 0000000..d82f5fd --- /dev/null +++ b/observer-web/src/components/broken-rail.tsx @@ -0,0 +1,32 @@ +interface BrokenItem { + library: string; + symbol: string; +} + +interface Props { + items: BrokenItem[]; +} + +export function BrokenRail({ items }: Props) { + return ( +
+

Broken / Stale Data

+ {items.length === 0 ? ( +

No broken data.

+ ) : ( +
    + {items.map((item, i) => ( +
  • +
    • + ))} +
+ )} +
+ ); +} diff --git a/observer-web/src/components/four-v-tiles.tsx b/observer-web/src/components/four-v-tiles.tsx new file mode 100644 index 0000000..9f50969 --- /dev/null +++ b/observer-web/src/components/four-v-tiles.tsx @@ -0,0 +1,144 @@ +"use client"; + +import { useEffect, useRef } from "react"; +import type { OverviewMetrics } from "@/lib/api"; + +// Lean ECharts import: core + canvas + pie only +import * as echarts from "echarts/core"; +import { PieChart } from "echarts/charts"; +import { TooltipComponent, LegendComponent } from "echarts/components"; +import { CanvasRenderer } from "echarts/renderers"; + +echarts.use([PieChart, TooltipComponent, LegendComponent, CanvasRenderer]); + +function VelocityDonut({ ok, stale, error }: { ok: number; stale: number; error: number }) { + const ref = useRef(null); + + useEffect(() => { + if (!ref.current) return; + const chart = echarts.init(ref.current, null, { renderer: "canvas" }); + chart.setOption({ + backgroundColor: "transparent", + tooltip: { trigger: "item", formatter: "{b}: {c}" }, + series: [ + { + type: "pie", + radius: ["55%", "85%"], + label: { show: false }, + itemStyle: { borderColor: "transparent", borderWidth: 2 }, + data: [ + { name: "ok", value: ok, itemStyle: { color: "var(--ok)" } }, + { name: "stale", value: stale, itemStyle: { color: "var(--warn)" } }, + { name: "error", value: error, itemStyle: { color: "var(--fail)" } }, + ], + }, + ], + }); + return () => chart.dispose(); + }, [ok, stale, error]); + + return
; +} + +interface Props { + metrics: OverviewMetrics; +} + +export function FourVTiles({ metrics }: Props) { + const { volume, velocity, variety, veracity } = metrics; + const totalVelocity = velocity.ok + velocity.stale + velocity.error; + + return ( +
+ {/* Volume */} +
+

Volume

+
+
+ Libraries + {volume.libraries} +
+
+ Symbols + {volume.symbols} +
+
+ Rows + {volume.rows.toLocaleString()} +
+
+
+ + {/* Velocity */} +
+

Velocity

+
+ +
+
+ + ok + {velocity.ok} +
+
+ + stale + {velocity.stale} +
+
+ + error + {velocity.error} +
+
+
+

{totalVelocity} symbols tracked

+
+ + {/* Variety */} +
+

Variety

+
+
+ Schemas + {variety.schemas} +
+
+

Revision modes

+
+ {variety.revision_modes.map((mode) => ( + + {mode} + + ))} +
+
+
+
+ + {/* Veracity */} +
+

Veracity

+
+
+ Broken + 0 ? "text-fail" : "text-ok"}`}> + {veracity.broken} + +
+ {veracity.broken === 0 && ( +

All symbols healthy

+ )} + {veracity.broken > 0 && ( +

+ {veracity.broken} symbol{veracity.broken !== 1 ? "s" : ""} stale or broken +

+ )} +
+
+
+ ); +} diff --git a/observer-web/src/components/freshness-heatmap.tsx b/observer-web/src/components/freshness-heatmap.tsx new file mode 100644 index 0000000..36faa58 --- /dev/null +++ b/observer-web/src/components/freshness-heatmap.tsx @@ -0,0 +1,64 @@ +import type { HealthRow } from "@/lib/api"; + +function statusColor(row: HealthRow): string { + if (row.freshness_status === "error") return "bg-fail/20 border-fail/40 text-fail"; + if (row.freshness_status === "stale") return "bg-warn/20 border-warn/40 text-warn"; + // ok — ramp from green toward ochre based on age_days + const age = row.age_days ?? 0; + if (age <= 1) return "bg-ok/20 border-ok/40 text-ok"; + if (age <= 3) return "bg-ok/10 border-ok/30 text-ok"; + return "bg-accent-tint border-accent-dim/40 text-accent"; +} + +function ageBadge(row: HealthRow): string { + if (row.age_days === null) return "—"; + if (row.age_days === 0) return "today"; + if (row.age_days === 1) return "1d"; + return `${row.age_days}d`; +} + +interface Props { + rows: HealthRow[]; +} + +export function FreshnessHeatmap({ rows }: Props) { + if (rows.length === 0) { + return ( +
+

Freshness Heatmap

+

No symbols to display.

+
+ ); + } + + // Group by library + const byLibrary = rows.reduce>((acc, r) => { + (acc[r.library] ??= []).push(r); + return acc; + }, {}); + + return ( +
+

Freshness Heatmap

+
+ {Object.entries(byLibrary).map(([lib, libRows]) => ( +
+

{lib}

+
+ {libRows.map((row) => ( +
+
{row.symbol}
+
{ageBadge(row)}
+
+ ))} +
+
+ ))} +
+
+ ); +} diff --git a/observer-web/src/lib/api.ts b/observer-web/src/lib/api.ts index 391425b..d4335e6 100644 --- a/observer-web/src/lib/api.ts +++ b/observer-web/src/lib/api.ts @@ -1,5 +1,24 @@ import { createClient } from "@/lib/supabase/server"; +export interface OverviewMetrics { + volume: { libraries: number; symbols: number; rows: number }; + velocity: { ok: number; stale: number; error: number }; + variety: { schemas: number; revision_modes: string[] }; + veracity: { broken: number; broken_symbols: { library: string; symbol: string }[] }; +} + +export interface HealthRow { + library: string; + symbol: string; + freshness_status: "ok" | "stale" | "error"; + age_days: number | null; + latest_valid_time: string | null; + row_count: number | null; + vintage_count: number | null; + reconstructed_pct: number | null; + schema_name: string | null; +} + const API = process.env.OBSERVER_API_URL ?? "http://localhost:8090"; export async function apiFetch(path: string): Promise { From 58e6839986c93fb3d48cfb12c4fbc1880c92863f Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Tue, 30 Jun 2026 20:29:40 -0400 Subject: [PATCH 08/10] feat(observer-web): Catalog/Explorer tree + detail panel (Overview/Schema/Vintages) --- observer-web/src/app/(app)/catalog/page.tsx | 143 +++++++++++++ observer-web/src/app/(app)/layout.tsx | 7 +- .../__tests__/catalog-tree.test.tsx | 33 +++ .../src/components/catalog-tree-client.tsx | 28 +++ observer-web/src/components/catalog-tree.tsx | 73 +++++++ .../src/components/nav-rail-active.tsx | 27 +++ observer-web/src/components/symbol-detail.tsx | 43 ++++ observer-web/src/components/symbol-tabs.tsx | 189 ++++++++++++++++++ observer-web/src/lib/api.ts | 36 ++++ 9 files changed, 574 insertions(+), 5 deletions(-) create mode 100644 observer-web/src/app/(app)/catalog/page.tsx create mode 100644 observer-web/src/components/__tests__/catalog-tree.test.tsx create mode 100644 observer-web/src/components/catalog-tree-client.tsx create mode 100644 observer-web/src/components/catalog-tree.tsx create mode 100644 observer-web/src/components/nav-rail-active.tsx create mode 100644 observer-web/src/components/symbol-detail.tsx create mode 100644 observer-web/src/components/symbol-tabs.tsx diff --git a/observer-web/src/app/(app)/catalog/page.tsx b/observer-web/src/app/(app)/catalog/page.tsx new file mode 100644 index 0000000..0f01c38 --- /dev/null +++ b/observer-web/src/app/(app)/catalog/page.tsx @@ -0,0 +1,143 @@ +import { apiFetch } from "@/lib/api"; +import type { CatalogLibrary, SchemaDescription, VintageRow } from "@/lib/api"; +import { CatalogTreeClient } from "@/components/catalog-tree-client"; +import { SymbolDetail } from "@/components/symbol-detail"; + +interface SearchParams { + library?: string | string[]; + symbol?: string | string[]; +} + +function str(v: string | string[] | undefined): string | null { + if (!v) return null; + return Array.isArray(v) ? v[0] : v; +} + +function isAuthError(msg: string): boolean { + return /: 40[13]/.test(msg); +} + +function ErrorBanner({ message }: { message: string }) { + const authError = isAuthError(message); + return ( +
+

+ {authError + ? "Couldn't load data — you may not have access. Try signing in again." + : "Couldn't load data — confirm observer-api is running and that you're signed in with access."} +

+
+ ); +} + +export default async function CatalogPage({ + searchParams, +}: { + searchParams: Promise; +}) { + const sp = await searchParams; + const selectedLibrary = str(sp.library); + const selectedSymbol = str(sp.symbol); + + // Always fetch catalog + let catalogResult: { libraries: CatalogLibrary[] } | { error: string } | null = null; + try { + catalogResult = await apiFetch<{ libraries: CatalogLibrary[] }>("/catalog"); + } catch (err) { + console.error("[CatalogPage] catalog fetch failed:", err); + catalogResult = { error: err instanceof Error ? err.message : String(err) }; + } + + const catalogError = catalogResult == null || "error" in catalogResult; + const libraries = + !catalogError && catalogResult !== null && "libraries" in catalogResult + ? catalogResult.libraries + : []; + + // Find selected symbol metadata from catalog (no extra fetch needed for overview data) + const selectedLib = selectedLibrary + ? libraries.find((l) => l.name === selectedLibrary) + : null; + const selectedSym = selectedLib && selectedSymbol + ? selectedLib.symbols.find((s) => s.symbol === selectedSymbol) + : null; + + // Fetch schema + vintages if a symbol is selected + let schema: SchemaDescription | null = null; + let schemaError: string | null = null; + let vintages: VintageRow[] = []; + let vintagesError: string | null = null; + + if (selectedLibrary && selectedSymbol) { + const base = `/symbol/${selectedLibrary}/${selectedSymbol}`; + + const [schemaResult, vintagesResult] = await Promise.allSettled([ + apiFetch<{ schema_name: string | null; columns: SchemaDescription["columns"]; checks: string[] }>( + `${base}/schema` + ), + apiFetch<{ library: string; symbol: string; vintages: VintageRow[] }>( + `${base}/vintages` + ), + ]); + + if (schemaResult.status === "fulfilled") { + schema = schemaResult.value; + } else { + console.error("[CatalogPage] schema fetch failed:", schemaResult.reason); + schemaError = schemaResult.reason instanceof Error + ? schemaResult.reason.message + : String(schemaResult.reason); + } + + if (vintagesResult.status === "fulfilled") { + vintages = vintagesResult.value.vintages; + } else { + console.error("[CatalogPage] vintages fetch failed:", vintagesResult.reason); + vintagesError = vintagesResult.reason instanceof Error + ? vintagesResult.reason.message + : String(vintagesResult.reason); + } + } + + return ( +
+ {/* Left pane: tree */} + + + {/* Right pane: detail */} +
+ {selectedSym && selectedLibrary ? ( + + ) : ( +
+ {selectedLibrary && selectedSymbol && !selectedSym + ? `Symbol "${selectedSymbol}" not found in library "${selectedLibrary}".` + : "Select a symbol from the tree to view details."} +
+ )} +
+
+ ); +} diff --git a/observer-web/src/app/(app)/layout.tsx b/observer-web/src/app/(app)/layout.tsx index c24a9cf..0477eee 100644 --- a/observer-web/src/app/(app)/layout.tsx +++ b/observer-web/src/app/(app)/layout.tsx @@ -1,7 +1,7 @@ import { redirect } from "next/navigation"; import { createClient } from "@/lib/supabase/server"; import { roleFromSession } from "@/lib/api"; -import { NavRail } from "@/components/nav-rail"; +import { NavRailActive } from "@/components/nav-rail-active"; export default async function AppLayout({ children, @@ -19,12 +19,9 @@ export default async function AppLayout({ const role = roleFromSession(session.access_token); - // active section is fixed until the other section routes exist - const active = "overview"; - return (
- +
Energex Observer diff --git a/observer-web/src/components/__tests__/catalog-tree.test.tsx b/observer-web/src/components/__tests__/catalog-tree.test.tsx new file mode 100644 index 0000000..992d777 --- /dev/null +++ b/observer-web/src/components/__tests__/catalog-tree.test.tsx @@ -0,0 +1,33 @@ +import { render, screen, fireEvent } from "@testing-library/react"; +import { describe, it, expect, vi } from "vitest"; +import { CatalogTree } from "../catalog-tree"; + +const cat = { + libraries: [ + { + name: "power.load", + mode: "bitemporal_merge", + symbols: [ + { + symbol: "erco", + row_count: 1, + latest_valid_time: null, + vintage_count: 1, + reconstructed_pct: 0, + schema_name: "ERCOT_LOAD", + }, + ], + unreadable: 0, + }, + ], +}; + +describe("CatalogTree", () => { + it("renders libraries and symbols and fires onSelect", () => { + const onSelect = vi.fn(); + render(); + expect(screen.getByText("power.load")).toBeInTheDocument(); + fireEvent.click(screen.getByText("erco")); + expect(onSelect).toHaveBeenCalledWith("power.load", "erco"); + }); +}); diff --git a/observer-web/src/components/catalog-tree-client.tsx b/observer-web/src/components/catalog-tree-client.tsx new file mode 100644 index 0000000..d7dc446 --- /dev/null +++ b/observer-web/src/components/catalog-tree-client.tsx @@ -0,0 +1,28 @@ +"use client"; + +import { useRouter } from "next/navigation"; +import { CatalogTree } from "@/components/catalog-tree"; +import type { CatalogLibrary } from "@/lib/api"; + +interface Props { + catalog: { libraries: CatalogLibrary[] }; + selectedLibrary: string | null; + selectedSymbol: string | null; +} + +export function CatalogTreeClient({ catalog, selectedLibrary, selectedSymbol }: Props) { + const router = useRouter(); + + function handleSelect(library: string, symbol: string) { + router.push(`/catalog?library=${encodeURIComponent(library)}&symbol=${encodeURIComponent(symbol)}`); + } + + return ( + + ); +} diff --git a/observer-web/src/components/catalog-tree.tsx b/observer-web/src/components/catalog-tree.tsx new file mode 100644 index 0000000..ad25a80 --- /dev/null +++ b/observer-web/src/components/catalog-tree.tsx @@ -0,0 +1,73 @@ +"use client"; + +import { useState } from "react"; +import type { CatalogLibrary } from "@/lib/api"; + +interface Catalog { + libraries: CatalogLibrary[]; +} + +interface Props { + catalog: Catalog; + selectedLibrary?: string | null; + selectedSymbol?: string | null; + onSelect: (library: string, symbol: string) => void; +} + +function FreshnessDot({ latestValidTime }: { latestValidTime: string | null }) { + if (!latestValidTime) return ; + const age = (Date.now() - new Date(latestValidTime).getTime()) / 86_400_000; + const color = age <= 1 ? "bg-ok" : age <= 3 ? "bg-warn" : "bg-fail"; + return ; +} + +export function CatalogTree({ catalog, selectedLibrary, selectedSymbol, onSelect }: Props) { + const [open, setOpen] = useState>(() => + Object.fromEntries(catalog.libraries.map((l) => [l.name, true])) + ); + + return ( +
+ {catalog.libraries.map((lib) => ( +
+ + {open[lib.name] && ( +
    + {lib.symbols.map((sym) => { + const isSelected = + selectedLibrary === lib.name && selectedSymbol === sym.symbol; + return ( +
  • + +
    • + ); + })} +
+ )} +
+ ))} +
+ ); +} diff --git a/observer-web/src/components/nav-rail-active.tsx b/observer-web/src/components/nav-rail-active.tsx new file mode 100644 index 0000000..6a65126 --- /dev/null +++ b/observer-web/src/components/nav-rail-active.tsx @@ -0,0 +1,27 @@ +"use client"; + +import { usePathname } from "next/navigation"; +import { NavRail } from "@/components/nav-rail"; + +const PATH_TO_SECTION: Record = { + "/": "overview", + "/catalog": "catalog", + "/map": "map", + "/graph": "graph", + "/quality": "quality", + "/admin": "admin", +}; + +function sectionFromPath(pathname: string): string { + // exact match first, then prefix + if (PATH_TO_SECTION[pathname]) return PATH_TO_SECTION[pathname]; + for (const [prefix, section] of Object.entries(PATH_TO_SECTION)) { + if (prefix !== "/" && pathname.startsWith(prefix)) return section; + } + return "overview"; +} + +export function NavRailActive({ role }: { role: string }) { + const pathname = usePathname(); + return ; +} diff --git a/observer-web/src/components/symbol-detail.tsx b/observer-web/src/components/symbol-detail.tsx new file mode 100644 index 0000000..49dd74a --- /dev/null +++ b/observer-web/src/components/symbol-detail.tsx @@ -0,0 +1,43 @@ +import type { CatalogSymbol, SchemaDescription, VintageRow } from "@/lib/api"; +import { SymbolTabs } from "@/components/symbol-tabs"; + +interface Props { + library: string; + sym: CatalogSymbol; + schema: SchemaDescription | null; + vintages: VintageRow[]; + schemaError?: string | null; + vintagesError?: string | null; +} + +export function SymbolDetail({ + library, + sym, + schema, + vintages, + schemaError, + vintagesError, +}: Props) { + return ( +
+
+

{sym.symbol}

+ {library} +
+ + {(schemaError || vintagesError) && ( +
+ {schemaError &&

Schema: {schemaError}

} + {vintagesError &&

Vintages: {vintagesError}

} +
+ )} + + +
+ ); +} diff --git a/observer-web/src/components/symbol-tabs.tsx b/observer-web/src/components/symbol-tabs.tsx new file mode 100644 index 0000000..fd34bce --- /dev/null +++ b/observer-web/src/components/symbol-tabs.tsx @@ -0,0 +1,189 @@ +"use client"; + +import { useState } from "react"; +import type { CatalogSymbol, SchemaDescription, VintageRow } from "@/lib/api"; + +type Tab = "overview" | "schema" | "vintages" | "series" | "quality"; + +const TABS: { id: Tab; label: string }[] = [ + { id: "overview", label: "Overview" }, + { id: "schema", label: "Schema" }, + { id: "vintages", label: "Vintages" }, + { id: "series", label: "Series" }, + { id: "quality", label: "Quality" }, +]; + +interface Props { + library: string; + sym: CatalogSymbol; + schema: SchemaDescription | null; + vintages: VintageRow[]; +} + +function OverviewTab({ library, sym }: { library: string; sym: CatalogSymbol }) { + return ( +
+
+ + + + + {sym.vintage_count !== null && ( + + )} + {sym.reconstructed_pct !== null && ( + + )} +
+
+

Latest valid time

+

+ {sym.latest_valid_time ?? none} +

+
+
+ ); +} + +function Badge({ label, value, mono }: { label: string; value: string; mono: boolean }) { + return ( +
+

{label}

+

{value}

+
+ ); +} + +function SchemaTab({ schema }: { schema: SchemaDescription | null }) { + if (!schema || !schema.schema_name) { + return

No schema registered for this symbol.

; + } + return ( +
+
+ Schema + {schema.schema_name} +
+
+ + + + + + + + + + + {schema.columns.map((col) => ( + + + + + + + ))} + +
ColumnDtypeNullableChecks
{col.name}{col.dtype} + + {col.nullable ? "yes" : "no"} + + + {col.checks.length > 0 ? ( + {col.checks.join(", ")} + ) : ( + — + )} +
+
+ {schema.checks.length > 0 && ( +
+

Table-level checks

+
+ {schema.checks.map((c, i) => ( + + {c} + + ))} +
+
+ )} +
+ ); +} + +function VintagesTab({ vintages }: { vintages: VintageRow[] }) { + if (vintages.length === 0) { + return

No vintage sidecar for this symbol.

; + } + return ( +
+ + + + + + + + + + + {vintages.map((v, i) => ( + + + + + + + ))} + +
As-ofVersionFetched atReconstructed
{v.as_of}{v.version}{v.fetched_at ?? "—"} + + {v.vintage_reconstructed ? "yes" : "no"} + +
+
+ ); +} + +function PlaceholderTab({ label }: { label: string }) { + return ( +
+

{label} — loads in Task 7

+
+ ); +} + +export function SymbolTabs({ library, sym, schema, vintages }: Props) { + const [active, setActive] = useState("overview"); + + return ( +
+ {/* Tab bar */} +
+ {TABS.map((t) => ( + + ))} +
+ + {/* Tab content */} +
+ {active === "overview" && } + {active === "schema" && } + {active === "vintages" && } + {active === "series" && } + {active === "quality" && } +
+
+ ); +} diff --git a/observer-web/src/lib/api.ts b/observer-web/src/lib/api.ts index d4335e6..d62f219 100644 --- a/observer-web/src/lib/api.ts +++ b/observer-web/src/lib/api.ts @@ -1,5 +1,41 @@ import { createClient } from "@/lib/supabase/server"; +export interface CatalogSymbol { + symbol: string; + row_count: number; + latest_valid_time: string | null; + vintage_count: number | null; + reconstructed_pct: number | null; + schema_name: string | null; +} + +export interface CatalogLibrary { + name: string; + mode: string; + symbols: CatalogSymbol[]; + unreadable: number; +} + +export interface SchemaColumn { + name: string; + dtype: string; + nullable: boolean; + checks: string[]; +} + +export interface SchemaDescription { + schema_name: string | null; + columns: SchemaColumn[]; + checks: string[]; +} + +export interface VintageRow { + as_of: string; + version: string; + fetched_at: string | null; + vintage_reconstructed: boolean; +} + export interface OverviewMetrics { volume: { libraries: number; symbols: number; rows: number }; velocity: { ok: number; stale: number; error: number }; From e55f5dba653e86048c980e525028edf39f2c4691 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Tue, 30 Jun 2026 20:40:09 -0400 Subject: [PATCH 09/10] feat(observer-web): bitemporal Series chart (as_of slider) + Quality panel + explorer smoke - Add same-origin GET proxy at /api/observer/[...path] that forwards to observer-api via apiFetch (session-gated, no open-proxy) - Add AsOfSlider component for point-in-time knowledge-time selection - Add SeriesChart component (ECharts line + DataZoom) that re-fetches on as_of change via the proxy route - Add QualityPanel component with gate verdict, failures table, gaps count, and anomalies summary - Wire Series + Quality tabs in SymbolTabs, removing Task-6 placeholders - Add Playwright @smoke explorer.spec.ts (deferred run needs composed stack) - Fix Task-6: VintageRow.version typed as number (was string); render with String(v.version) in VintagesTab; guard schema/vintages fetch on selectedSym --- observer-web/e2e/explorer.spec.ts | 55 +++++ observer-web/src/app/(app)/catalog/page.tsx | 2 +- .../src/app/api/observer/[...path]/route.ts | 47 +++++ .../__tests__/as-of-slider.test.tsx | 33 +++ observer-web/src/components/as-of-slider.tsx | 59 ++++++ observer-web/src/components/quality-panel.tsx | 146 ++++++++++++++ observer-web/src/components/series-chart.tsx | 189 ++++++++++++++++++ observer-web/src/components/symbol-tabs.tsx | 24 ++- observer-web/src/lib/api.ts | 2 +- 9 files changed, 544 insertions(+), 13 deletions(-) create mode 100644 observer-web/e2e/explorer.spec.ts create mode 100644 observer-web/src/app/api/observer/[...path]/route.ts create mode 100644 observer-web/src/components/__tests__/as-of-slider.test.tsx create mode 100644 observer-web/src/components/as-of-slider.tsx create mode 100644 observer-web/src/components/quality-panel.tsx create mode 100644 observer-web/src/components/series-chart.tsx diff --git a/observer-web/e2e/explorer.spec.ts b/observer-web/e2e/explorer.spec.ts new file mode 100644 index 0000000..1a16892 --- /dev/null +++ b/observer-web/e2e/explorer.spec.ts @@ -0,0 +1,55 @@ +import { test, expect } from "@playwright/test"; + +// @smoke +// Requires: next dev running on port 3000, observer-api on 8090, Supabase running on 54321 +// Seeded test user: admin@energex.local / energex-observer-dev (role: admin) +// Run: npx playwright test e2e/explorer.spec.ts +// Deferred: full run requires the composed stack (docker compose up). + +const E2E_EMAIL = process.env.E2E_EMAIL ?? "admin@energex.local"; +const E2E_PASSWORD = process.env.E2E_PASSWORD ?? "energex-observer-dev"; + +async function signIn(page: Parameters[1]>[0]) { + await page.goto("http://localhost:3000/login"); + await page.fill('input[type="email"]', E2E_EMAIL); + await page.fill('input[type="password"]', E2E_PASSWORD); + await page.click('button[type="submit"]'); + await expect(page).toHaveURL("http://localhost:3000/", { timeout: 10000 }); +} + +test.describe("@smoke explorer flow", () => { + test("home page shows the 4V tiles after sign-in", async ({ page }) => { + await signIn(page); + await expect(page.getByText("Volume")).toBeVisible({ timeout: 8000 }); + await expect(page.getByText("Velocity")).toBeVisible({ timeout: 8000 }); + await expect(page.getByText("Variety")).toBeVisible({ timeout: 8000 }); + await expect(page.getByText("Veracity")).toBeVisible({ timeout: 8000 }); + }); + + test("navigates to Catalog and selects a symbol", async ({ page }) => { + await signIn(page); + + // Navigate to Catalog via the nav rail + await page.click('a[href="/catalog"]'); + await expect(page).toHaveURL(/\/catalog/, { timeout: 8000 }); + + // The catalog tree should render the sidebar heading + await expect(page.getByText("Catalog")).toBeVisible({ timeout: 8000 }); + }); + + test("Series tab renders a chart canvas for a symbol", async ({ page }) => { + await signIn(page); + + // Navigate directly to catalog with a known seeded symbol + // (adjust library/symbol to match seeded test data) + await page.goto("http://localhost:3000/catalog?library=power.load&symbol=erco", { + waitUntil: "networkidle", + }); + + // Click the Series tab + await page.click('button:has-text("Series")'); + + // The chart canvas should mount (echarts renders a canvas inside the chart div) + await expect(page.locator('[aria-label="series chart"]')).toBeVisible({ timeout: 10000 }); + }); +}); diff --git a/observer-web/src/app/(app)/catalog/page.tsx b/observer-web/src/app/(app)/catalog/page.tsx index 0f01c38..4a8fa5f 100644 --- a/observer-web/src/app/(app)/catalog/page.tsx +++ b/observer-web/src/app/(app)/catalog/page.tsx @@ -68,7 +68,7 @@ export default async function CatalogPage({ let vintages: VintageRow[] = []; let vintagesError: string | null = null; - if (selectedLibrary && selectedSymbol) { + if (selectedLibrary && selectedSymbol && selectedSym) { const base = `/symbol/${selectedLibrary}/${selectedSymbol}`; const [schemaResult, vintagesResult] = await Promise.allSettled([ diff --git a/observer-web/src/app/api/observer/[...path]/route.ts b/observer-web/src/app/api/observer/[...path]/route.ts new file mode 100644 index 0000000..a07f4c1 --- /dev/null +++ b/observer-web/src/app/api/observer/[...path]/route.ts @@ -0,0 +1,47 @@ +/** + * Same-origin proxy to observer-api for client-side components. + * + * Design notes: + * - Only proxies GET requests; only ever calls observer-api via `apiFetch` (never + * a caller-supplied host) — not an open proxy to arbitrary URLs. + * - Requires an authenticated session: apiFetch resolves the Supabase bearer and + * throws on 401/403. A missing/invalid session returns 401 to the browser. + * - Maps apiFetch error status codes back to the response (parses ": NNN" from the + * thrown message; defaults to 502 for unknown upstream errors). + */ + +import type { NextRequest } from "next/server"; +import { apiFetch } from "@/lib/api"; + +function parseStatus(msg: string): number { + const m = msg.match(/:\s*([45]\d{2})/); + if (m) return Number(m[1]); + return 502; +} + +export async function GET( + request: NextRequest, + { params }: { params: Promise<{ path: string[] }> } +) { + const { path } = await params; + const apiPath = "/" + path.join("/"); + + // Forward the incoming query string to observer-api + const qs = request.nextUrl.search; // e.g. "?as_of=2026-06-02T00:00:00Z" or "" + const fullPath = `${apiPath}${qs}`; + + let data: unknown; + try { + data = await apiFetch(fullPath); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + const status = parseStatus(msg); + // Surface auth errors as 401 so the browser can redirect to /login + return new Response(JSON.stringify({ error: msg }), { + status, + headers: { "Content-Type": "application/json" }, + }); + } + + return Response.json(data); +} diff --git a/observer-web/src/components/__tests__/as-of-slider.test.tsx b/observer-web/src/components/__tests__/as-of-slider.test.tsx new file mode 100644 index 0000000..cb2b4d6 --- /dev/null +++ b/observer-web/src/components/__tests__/as-of-slider.test.tsx @@ -0,0 +1,33 @@ +import { render, screen, fireEvent } from "@testing-library/react"; +import { describe, it, expect, vi } from "vitest"; +import { AsOfSlider } from "../as-of-slider"; + +describe("AsOfSlider", () => { + it("emits the chosen as_of", () => { + const onChange = vi.fn(); + render(); + fireEvent.change(screen.getByRole("slider"), { target: { value: "0" } }); + expect(onChange).toHaveBeenCalledWith("2026-06-02T00:00:00Z"); + }); + + it("emits undefined for 'latest' position", () => { + const onChange = vi.fn(); + render(); + // Move to a vintage first, then back to latest + fireEvent.change(screen.getByRole("slider"), { target: { value: "0" } }); + fireEvent.change(screen.getByRole("slider"), { target: { value: "2" } }); + expect(onChange).toHaveBeenLastCalledWith(undefined); + }); + + it("shows 'latest' label when at latest position", () => { + render(); + expect(screen.getByText(/latest/i)).toBeInTheDocument(); + }); + + it("renders nothing special when no vintages", () => { + const onChange = vi.fn(); + render(); + // Only one stop (latest), no slider needed — or slider with just the latest stop + expect(screen.getByText(/latest/i)).toBeInTheDocument(); + }); +}); diff --git a/observer-web/src/components/as-of-slider.tsx b/observer-web/src/components/as-of-slider.tsx new file mode 100644 index 0000000..27f563f --- /dev/null +++ b/observer-web/src/components/as-of-slider.tsx @@ -0,0 +1,59 @@ +"use client"; + +import { useState } from "react"; + +interface Props { + vintages: string[]; // ISO strings (as_of), oldest → newest + onChange: (asOf: string | undefined) => void; +} + +/** + * Bitemporal as_of slider. + * + * Stops: [vintage[0], ..., vintage[n-1], "latest"] + * Index 0..n-1 → emit that vintage's as_of ISO + * Index n → "latest" → emit undefined (no as_of param) + * + * Default: "latest" (rightmost stop). + */ +export function AsOfSlider({ vintages, onChange }: Props) { + const latestIdx = vintages.length; + // Default to "latest" (rightmost stop) + const [idx, setIdx] = useState(latestIdx); + + // With no vintages there's only "latest" — nothing to slide + if (vintages.length === 0) { + return ( +
+ Knowledge time: + latest +
+ ); + } + + function handleChange(e: React.ChangeEvent) { + const i = Number(e.target.value); + setIdx(i); + onChange(i < latestIdx ? vintages[i] : undefined); + } + + const label = idx < latestIdx ? vintages[idx] : "latest"; + + return ( +
+ Knowledge time + + {label} +
+ ); +} diff --git a/observer-web/src/components/quality-panel.tsx b/observer-web/src/components/quality-panel.tsx new file mode 100644 index 0000000..051022e --- /dev/null +++ b/observer-web/src/components/quality-panel.tsx @@ -0,0 +1,146 @@ +"use client"; + +import { useEffect, useState } from "react"; + +interface FailureRow { + check: string; + column: string | null; + failure_case: string; +} + +interface QualityResponse { + library: string; + symbol: string; + passed: boolean; + failures: FailureRow[]; + gaps: number; + anomalies: Record | null; + anomalies_note: string | null; +} + +interface Props { + library: string; + symbol: string; +} + +export function QualityPanel({ library, symbol }: Props) { + const [data, setData] = useState(null); + const [loading, setLoading] = useState(false); + const [error, setError] = useState(null); + + useEffect(() => { + let cancelled = false; + setLoading(true); + setError(null); + + fetch(`/api/observer/symbol/${library}/${symbol}/quality`) + .then((r) => { + if (!r.ok) throw new Error(`quality fetch: ${r.status}`); + return r.json() as Promise; + }) + .then((d) => { + if (!cancelled) { + setData(d); + setLoading(false); + } + }) + .catch((err) => { + if (!cancelled) { + setError(err instanceof Error ? err.message : String(err)); + setLoading(false); + } + }); + + return () => { cancelled = true; }; + }, [library, symbol]); + + if (loading) { + return

Loading quality report…

; + } + + if (error) { + return ( +
+ {error} +
+ ); + } + + if (!data) return null; + + return ( +
+ {/* Gate verdict */} +
+ Quality gate + {data.passed ? ( + + pass + + ) : ( + + fail + + )} +
+ + {/* Failures table */} + {data.failures.length > 0 && ( +
+

Failures ({data.failures.length})

+
+ + + + + + + + + + {data.failures.map((f, i) => ( + + + + + + ))} + +
CheckColumnFailure case
{f.check}{f.column ?? "—"}{f.failure_case}
+
+
+ )} + + {data.failures.length === 0 && data.passed && ( +

No check failures.

+ )} + + {/* Gaps */} +
+ Data gaps + 0 ? "text-warn" : "text-ok"}`}>{data.gaps} +
+ + {/* Anomalies */} + {data.anomalies_note && ( +
+ {data.anomalies_note} +
+ )} + + {data.anomalies && !data.anomalies_note && ( +
+

Anomalies

+
+ {Object.entries(data.anomalies).map(([key, val]) => ( +
+

{key}

+

{String(val)}

+
+ ))} +
+
+ )} +
+ ); +} diff --git a/observer-web/src/components/series-chart.tsx b/observer-web/src/components/series-chart.tsx new file mode 100644 index 0000000..6870333 --- /dev/null +++ b/observer-web/src/components/series-chart.tsx @@ -0,0 +1,189 @@ +"use client"; + +import { useEffect, useRef, useState } from "react"; +import { AsOfSlider } from "./as-of-slider"; + +// Lean ECharts import: core + canvas + line +import * as echarts from "echarts/core"; +import { LineChart } from "echarts/charts"; +import { TooltipComponent, GridComponent, DataZoomComponent } from "echarts/components"; +import { CanvasRenderer } from "echarts/renderers"; + +echarts.use([LineChart, TooltipComponent, GridComponent, DataZoomComponent, CanvasRenderer]); + +interface SeriesRow { + valid_time: string; + [key: string]: unknown; +} + +interface SeriesResponse { + library: string; + symbol: string; + as_of: string | null; + columns: string[]; + rows: SeriesRow[]; +} + +interface Props { + library: string; + symbol: string; + vintageAsOfs: string[]; // ordered list of as_of ISOs from /vintages +} + +function firstValueColumn(columns: string[]): string | null { + // Skip valid_time, return first numeric-looking column + const skip = new Set(["valid_time", "symbol", "library"]); + return columns.find((c) => !skip.has(c)) ?? null; +} + +export function SeriesChart({ library, symbol, vintageAsOfs }: Props) { + const [asOf, setAsOf] = useState(undefined); // undefined = latest + const [data, setData] = useState(null); + const [loading, setLoading] = useState(false); + const [error, setError] = useState(null); + const chartRef = useRef(null); + const chartInstance = useRef(null); + + // Fetch series data whenever asOf changes + useEffect(() => { + let cancelled = false; + setLoading(true); + setError(null); + + const qs = asOf ? `?as_of=${encodeURIComponent(asOf)}` : ""; + fetch(`/api/observer/symbol/${library}/${symbol}/series${qs}`) + .then((r) => { + if (!r.ok) throw new Error(`series fetch: ${r.status}`); + return r.json() as Promise; + }) + .then((d) => { + if (!cancelled) { + setData(d); + setLoading(false); + } + }) + .catch((err) => { + if (!cancelled) { + setError(err instanceof Error ? err.message : String(err)); + setLoading(false); + } + }); + + return () => { cancelled = true; }; + }, [asOf, library, symbol]); + + // Render / update ECharts when data changes + useEffect(() => { + if (!chartRef.current || !data) return; + + if (!chartInstance.current) { + chartInstance.current = echarts.init(chartRef.current, null, { renderer: "canvas" }); + } + + const col = firstValueColumn(data.columns); + if (!col) return; + + const xData = data.rows.map((r) => r.valid_time as string); + const yData = data.rows.map((r) => r[col] as number); + + chartInstance.current.setOption({ + backgroundColor: "transparent", + grid: { left: 60, right: 20, top: 20, bottom: 40 }, + tooltip: { + trigger: "axis", + axisPointer: { type: "cross" }, + backgroundColor: "var(--panel)", + borderColor: "var(--line)", + textStyle: { color: "var(--fg)", fontFamily: "var(--mono)", fontSize: 11 }, + }, + dataZoom: [{ type: "inside" }, { type: "slider", height: 16 }], + xAxis: { + type: "category", + data: xData, + axisLine: { lineStyle: { color: "var(--line)" } }, + axisLabel: { + color: "var(--muted)", + fontFamily: "var(--mono)", + fontSize: 10, + rotate: 30, + interval: Math.max(0, Math.floor(xData.length / 8) - 1), + }, + splitLine: { show: false }, + }, + yAxis: { + type: "value", + axisLine: { show: false }, + splitLine: { lineStyle: { color: "var(--line-soft)" } }, + axisLabel: { + color: "var(--muted)", + fontFamily: "var(--mono)", + fontSize: 10, + }, + }, + series: [ + { + name: col, + type: "line", + data: yData, + lineStyle: { color: "var(--accent)", width: 1.5 }, + itemStyle: { color: "var(--accent)" }, + showSymbol: false, + areaStyle: { color: "var(--accent-tint)", opacity: 0.6 }, + }, + ], + }, true); + + return () => { + // Don't dispose here — reuse on re-render + }; + }, [data]); + + // Dispose on unmount + useEffect(() => { + return () => { + chartInstance.current?.dispose(); + chartInstance.current = null; + }; + }, []); + + const valueCol = data ? firstValueColumn(data.columns) : null; + + return ( +
+ + + {loading && ( +
+ Loading series data… +
+ )} + + {error && !loading && ( +
+ {error} +
+ )} + + {!loading && !error && data && data.rows.length === 0 && ( +

No data for this symbol.

+ )} + + {!loading && !error && data && data.rows.length > 0 && ( +
+
+ {data.rows.length.toLocaleString()} rows + {valueCol && · column: {valueCol}} + {data.as_of && ( + · as_of: {data.as_of} + )} +
+
+
+ )} +
+ ); +} diff --git a/observer-web/src/components/symbol-tabs.tsx b/observer-web/src/components/symbol-tabs.tsx index fd34bce..80e8e95 100644 --- a/observer-web/src/components/symbol-tabs.tsx +++ b/observer-web/src/components/symbol-tabs.tsx @@ -2,6 +2,8 @@ import { useState } from "react"; import type { CatalogSymbol, SchemaDescription, VintageRow } from "@/lib/api"; +import { SeriesChart } from "./series-chart"; +import { QualityPanel } from "./quality-panel"; type Tab = "overview" | "schema" | "vintages" | "series" | "quality"; @@ -131,7 +133,7 @@ function VintagesTab({ vintages }: { vintages: VintageRow[] }) { {vintages.map((v, i) => ( {v.as_of} - {v.version} + {String(v.version)} {v.fetched_at ?? "—"} @@ -146,14 +148,6 @@ function VintagesTab({ vintages }: { vintages: VintageRow[] }) { ); } -function PlaceholderTab({ label }: { label: string }) { - return ( -
-

{label} — loads in Task 7

-
- ); -} - export function SymbolTabs({ library, sym, schema, vintages }: Props) { const [active, setActive] = useState("overview"); @@ -181,8 +175,16 @@ export function SymbolTabs({ library, sym, schema, vintages }: Props) { {active === "overview" && } {active === "schema" && } {active === "vintages" && } - {active === "series" && } - {active === "quality" && } + {active === "series" && ( + v.as_of)} + /> + )} + {active === "quality" && ( + + )}
); diff --git a/observer-web/src/lib/api.ts b/observer-web/src/lib/api.ts index d62f219..7b2dcc7 100644 --- a/observer-web/src/lib/api.ts +++ b/observer-web/src/lib/api.ts @@ -31,7 +31,7 @@ export interface SchemaDescription { export interface VintageRow { as_of: string; - version: string; + version: number; fetched_at: string | null; vintage_reconstructed: boolean; } From 702d1c15fdf9c7bbe4b61589fc40ad9f193d344a Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Tue, 30 Jun 2026 20:53:32 -0400 Subject: [PATCH 10/10] fix(observer): align /series contract (columns+as_of) and render quality no-schema as neutral, not fail --- .../__tests__/quality-panel.test.tsx | 49 +++++++++++++ .../__tests__/series-chart.test.tsx | 71 +++++++++++++++++++ observer-web/src/components/quality-panel.tsx | 8 ++- observer-web/src/components/series-chart.tsx | 3 +- src/energex/observer/routers/symbol.py | 8 ++- tests/test_observer_symbol.py | 16 ++++- 6 files changed, 148 insertions(+), 7 deletions(-) create mode 100644 observer-web/src/components/__tests__/quality-panel.test.tsx create mode 100644 observer-web/src/components/__tests__/series-chart.test.tsx diff --git a/observer-web/src/components/__tests__/quality-panel.test.tsx b/observer-web/src/components/__tests__/quality-panel.test.tsx new file mode 100644 index 0000000..8aa1825 --- /dev/null +++ b/observer-web/src/components/__tests__/quality-panel.test.tsx @@ -0,0 +1,49 @@ +import { render, screen } from "@testing-library/react"; +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { QualityPanel } from "../quality-panel"; + +beforeEach(() => { + vi.restoreAllMocks(); +}); + +function mockQualityFetch(passed: boolean | null) { + vi.spyOn(globalThis, "fetch").mockResolvedValueOnce({ + ok: true, + json: async () => ({ + library: "power.load", + symbol: "erco", + passed, + failures: passed === false ? [{ check: "not_null", column: "value", failure_case: "2 nulls" }] : [], + gaps: 0, + anomalies: null, + anomalies_note: null, + }), + } as Response); +} + +describe("QualityPanel gate verdict", () => { + it("renders pass badge when passed=true", async () => { + mockQualityFetch(true); + render(); + expect(await screen.findByText("pass")).toBeInTheDocument(); + expect(screen.queryByText("fail")).not.toBeInTheDocument(); + expect(screen.queryByText(/no schema/i)).not.toBeInTheDocument(); + }); + + it("renders fail badge when passed=false", async () => { + mockQualityFetch(false); + render(); + expect(await screen.findByText("fail")).toBeInTheDocument(); + expect(screen.queryByText("pass")).not.toBeInTheDocument(); + expect(screen.queryByText(/no schema/i)).not.toBeInTheDocument(); + }); + + it("renders neutral 'No schema registered' badge when passed=null (not a fail badge)", async () => { + mockQualityFetch(null); + render(); + expect(await screen.findByText(/no schema registered/i)).toBeInTheDocument(); + // Must NOT show fail — null means unmapped, not broken + expect(screen.queryByText("fail")).not.toBeInTheDocument(); + expect(screen.queryByText("pass")).not.toBeInTheDocument(); + }); +}); diff --git a/observer-web/src/components/__tests__/series-chart.test.tsx b/observer-web/src/components/__tests__/series-chart.test.tsx new file mode 100644 index 0000000..2d00e0e --- /dev/null +++ b/observer-web/src/components/__tests__/series-chart.test.tsx @@ -0,0 +1,71 @@ +import { render, screen } from "@testing-library/react"; +import { describe, it, expect, vi, beforeEach } from "vitest"; + +// Mock echarts to avoid canvas errors in jsdom +const chartInstance = { setOption: vi.fn(), resize: vi.fn(), dispose: vi.fn() }; +vi.mock("echarts/core", () => ({ + use: vi.fn(), + init: vi.fn(() => chartInstance), +})); +vi.mock("echarts/renderers", () => ({ CanvasRenderer: {} })); +vi.mock("echarts/charts", () => ({ LineChart: {} })); +vi.mock("echarts/components", () => ({ + TooltipComponent: {}, + GridComponent: {}, + DataZoomComponent: {}, +})); + +import { SeriesChart } from "../series-chart"; + +const SERIES_RESPONSE = { + library: "power.load", + symbol: "erco", + as_of: "2026-06-03T00:00:00Z", + columns: ["Datetime", "instrument_id", "valid_time", "value"], + rows: [ + { valid_time: "2026-06-01T00:00:00+00:00", instrument_id: "ERCOT.LOAD", value: 40000.0 }, + ], +}; + +beforeEach(() => { + vi.restoreAllMocks(); +}); + +describe("SeriesChart contract", () => { + it("renders chart container without throwing when response includes columns + as_of", async () => { + vi.spyOn(globalThis, "fetch").mockResolvedValueOnce({ + ok: true, + json: async () => SERIES_RESPONSE, + } as Response); + + render( + + ); + + // Chart container must be rendered + const container = await screen.findByRole("img", { name: /series chart/i }).catch(() => null) ?? + await screen.findByLabelText(/series chart/i); + expect(container).toBeInTheDocument(); + }); + + it("does not throw when response is missing columns (defensive guard)", async () => { + const incompleteResponse = { + library: "power.load", + symbol: "erco", + // no columns, no as_of — old backend shape + rows: [{ valid_time: "2026-06-01T00:00:00+00:00", value: 40000.0 }], + }; + + vi.spyOn(globalThis, "fetch").mockResolvedValueOnce({ + ok: true, + json: async () => incompleteResponse, + } as Response); + + // Should not throw — firstValueColumn guard handles missing columns + expect(() => + render( + + ) + ).not.toThrow(); + }); +}); diff --git a/observer-web/src/components/quality-panel.tsx b/observer-web/src/components/quality-panel.tsx index 051022e..4837bd1 100644 --- a/observer-web/src/components/quality-panel.tsx +++ b/observer-web/src/components/quality-panel.tsx @@ -11,7 +11,7 @@ interface FailureRow { interface QualityResponse { library: string; symbol: string; - passed: boolean; + passed: boolean | null; failures: FailureRow[]; gaps: number; anomalies: Record | null; @@ -73,7 +73,11 @@ export function QualityPanel({ library, symbol }: Props) { {/* Gate verdict */}
Quality gate - {data.passed ? ( + {data.passed === null ? ( + + No schema registered + + ) : data.passed ? ( pass diff --git a/observer-web/src/components/series-chart.tsx b/observer-web/src/components/series-chart.tsx index 6870333..7c01c0b 100644 --- a/observer-web/src/components/series-chart.tsx +++ b/observer-web/src/components/series-chart.tsx @@ -30,7 +30,8 @@ interface Props { vintageAsOfs: string[]; // ordered list of as_of ISOs from /vintages } -function firstValueColumn(columns: string[]): string | null { +function firstValueColumn(columns: string[] | undefined | null): string | null { + if (!columns || columns.length === 0) return null; // Skip valid_time, return first numeric-looking column const skip = new Set(["valid_time", "symbol", "library"]); return columns.find((c) => !skip.has(c)) ?? null; diff --git a/src/energex/observer/routers/symbol.py b/src/energex/observer/routers/symbol.py index 48fe912..12a22fd 100644 --- a/src/energex/observer/routers/symbol.py +++ b/src/energex/observer/routers/symbol.py @@ -52,7 +52,13 @@ def series( for col in out.columns: if str(out[col].dtype).startswith("datetime"): out[col] = out[col].astype(str) - return {"library": library, "symbol": symbol, "rows": out.to_dict(orient="records")} + return { + "library": library, + "symbol": symbol, + "as_of": as_of, + "columns": list(out.columns), + "rows": out.to_dict(orient="records"), + } @router.get("/schema") diff --git a/tests/test_observer_symbol.py b/tests/test_observer_symbol.py index adb9987..7f2ae16 100644 --- a/tests/test_observer_symbol.py +++ b/tests/test_observer_symbol.py @@ -82,11 +82,21 @@ def test_series_point_in_time(observer_client, observer_arctic): "/symbol/power.load/erco/series", params={"as_of": "2026-06-03T00:00:00Z"} ) assert r.status_code == 200 - rows = r.json()["rows"] + body = r.json() + rows = body["rows"] assert rows[-1]["value"] == 40000.0 - # latest (no as_of) -> sees the revision + # contract: columns list must be present and include "value" + assert isinstance(body["columns"], list) + assert len(body["columns"]) > 0 + assert "value" in body["columns"] + # contract: as_of echoed back when provided + assert body["as_of"] == "2026-06-03T00:00:00Z" + # latest (no as_of) -> sees the revision; as_of is null when not requested r2 = observer_client.get("/symbol/power.load/erco/series") - assert r2.json()["rows"][-1]["value"] == 41000.0 + body2 = r2.json() + assert body2["rows"][-1]["value"] == 41000.0 + assert body2["as_of"] is None + assert "value" in body2["columns"] def test_series_requires_auth(observer_arctic, monkeypatch):