feat(b3b): persist backtest verdict to cockpit (backtest_summary table + --persist-cockpit)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/fxhnt/adapters/persistence/cockpit_models.py

@@ -36,6 +36,22 @@ class ForwardSummaryRow(CockpitBase):
     src_updated_at: Mapped[dt.datetime] = mapped_column(DateTime)
 
 
+class BacktestSummaryRow(CockpitBase):
+    __tablename__ = "backtest_summary"
+    strategy_id: Mapped[str] = mapped_column(String(64), primary_key=True)  # eqfactor_long / eqfactor_ls / eqfactor_tilt
+    as_of: Mapped[str] = mapped_column(String(10))            # last rebalance date in the backtest
+    cagr: Mapped[float] = mapped_column(Float)
+    ann_vol: Mapped[float] = mapped_column(Float)
+    sharpe: Mapped[float] = mapped_column(Float)
+    max_drawdown: Mapped[float] = mapped_column(Float)
+    passed: Mapped[bool] = mapped_column(Boolean)
+    dsr: Mapped[float] = mapped_column(Float)
+    is_sharpe: Mapped[float] = mapped_column(Float)
+    oos_sharpe: Mapped[float] = mapped_column(Float)
+    pvalue: Mapped[float] = mapped_column(Float)
+    src_updated_at: Mapped[dt.datetime] = mapped_column(DateTime)
+
+
 class StrategyRegistryRow(CockpitBase):
     __tablename__ = "strategy_registry"
     strategy_id: Mapped[str] = mapped_column(String(64), primary_key=True)

src/fxhnt/adapters/persistence/forward_nav.py

@@ -10,11 +10,13 @@ from sqlalchemy.orm import Session
 from sqlalchemy.pool import StaticPool
 
 from fxhnt.adapters.persistence.cockpit_models import (
+    BacktestSummaryRow,
     CockpitBase,
     ForwardNavRow,
     ForwardSummaryRow,
     StrategyRegistryRow,
 )
+from fxhnt.application.forward_models import BacktestSummary
 from fxhnt.application.forward_models import ForwardNavRow as NavRowDTO
 from fxhnt.application.forward_models import ForwardSummary
 from fxhnt.registry import STRATEGY_REGISTRY
@@ -66,6 +68,18 @@ class ForwardNavRepo:
                 gate_status=gate_status, gate_reason=gate_reason, src_updated_at=at))
             s.commit()
 
+    def upsert_backtest_summary(self, sm: BacktestSummary, at: dt.datetime) -> None:
+        with Session(self._engine) as s:
+            s.merge(BacktestSummaryRow(
+                strategy_id=sm.strategy_id, as_of=sm.as_of, cagr=sm.cagr, ann_vol=sm.ann_vol,
+                sharpe=sm.sharpe, max_drawdown=sm.max_drawdown, passed=sm.passed, dsr=sm.dsr,
+                is_sharpe=sm.is_sharpe, oos_sharpe=sm.oos_sharpe, pvalue=sm.pvalue, src_updated_at=at))
+            s.commit()
+
+    def all_backtest_summaries(self) -> list[BacktestSummaryRow]:
+        with Session(self._engine) as s:
+            return list(s.scalars(select(BacktestSummaryRow).order_by(BacktestSummaryRow.strategy_id)))
+
     def all_nav_histories(self) -> dict[str, list[NavRowDTO]]:
         """One query for every strategy's nav series, grouped by strategy_id (avoids N+1 on the cockpit)."""
         stmt = select(ForwardNavRow).order_by(ForwardNavRow.strategy_id, ForwardNavRow.date)

src/fxhnt/application/backtest_ingest.py

@@ -0,0 +1,42 @@
+"""Map a gauntlet backtest report (evaluate_constructions output) into BacktestSummary DTOs for the
+cockpit. Pure function — no I/O — so it's trivially testable; the CLI does the persistence. Mirrors the
+forward_ingest seam (report -> DTOs -> repo.upsert)."""
+from __future__ import annotations
+
+from typing import Any
+
+from fxhnt.application.forward_models import BacktestSummary
+
+# Backtest constructions map to the three live equity-factor paper-track strategy ids.
+_CONSTRUCTION_TO_SID = {"long": "eqfactor_long", "ls": "eqfactor_ls", "tilt": "eqfactor_tilt"}
+
+
+def backtest_summaries_from_report(report: dict[str, Any]) -> list[BacktestSummary]:
+    """Convert evaluate_constructions(...) output -> one BacktestSummary per construction.
+
+    as_of = the last rebalance date in the run ("" if none). cagr/ann_vol/sharpe/max_drawdown come from
+    each construction's `stats`; passed/dsr/is_sharpe/oos_sharpe/pvalue from its `verdict`.
+    """
+    rebals = report.get("rebalance_dates", [])
+    as_of = rebals[-1] if rebals else ""
+    out: list[BacktestSummary] = []
+    for construction, block in report.get("constructions", {}).items():
+        sid = _CONSTRUCTION_TO_SID.get(construction)
+        if sid is None:
+            continue
+        stats = block["stats"]
+        verdict = block["verdict"]
+        out.append(BacktestSummary(
+            strategy_id=sid,
+            as_of=as_of,
+            cagr=float(stats["cagr"]),
+            ann_vol=float(stats["ann_vol"]),
+            sharpe=float(stats["sharpe"]),
+            max_drawdown=float(stats["max_drawdown"]),
+            passed=bool(verdict["passed"]),
+            dsr=float(verdict["dsr"]),
+            is_sharpe=float(verdict["is_sharpe"]),
+            oos_sharpe=float(verdict["oos_sharpe"]),
+            pvalue=float(verdict["pvalue"]),
+        ))
+    return out

src/fxhnt/application/forward_models.py

@@ -29,6 +29,24 @@ class ForwardSummary:
         return 100.0 * self.total_return
 
 
+@dataclass(frozen=True, slots=True)
+class BacktestSummary:
+    """One construction's historical-backtest verdict, persisted to the cockpit for the dashboard.
+    Mirrors ForwardSummary but carries the gauntlet outcome (passed/dsr/is/oos/pvalue) instead of the
+    forward gate. `as_of` is the last rebalance date in the backtest run."""
+    strategy_id: str       # eqfactor_long | eqfactor_ls | eqfactor_tilt
+    as_of: str             # last rebalance date (ISO yyyy-mm-dd), "" if none
+    cagr: float
+    ann_vol: float
+    sharpe: float
+    max_drawdown: float    # <= 0
+    passed: bool
+    dsr: float
+    is_sharpe: float
+    oos_sharpe: float
+    pvalue: float
+
+
 @dataclass(frozen=True, slots=True)
 class GateVerdict:
     status: str        # "WAIT" | "GO" | "NO_GO"

src/fxhnt/cli.py

@@ -519,6 +519,7 @@ def backtest_equity(
     candidate_top_k: int = typer.Option(2000, help="Bounded candidate universe: top-K most-liquid names (ONLY used with --no-pit)."),
     candidate_min_history: int = typer.Option(300, help="Min daily bars (close+volume) a name needs to be a candidate (ONLY used with --no-pit)."),
     delisting_return: float = typer.Option(-0.30, "--delisting-return", help="Terminal return booked when a held name delists (data series ends mid-hold). -0.30 = Shumway canonical; 0.0 = none (biased); -1.0 = worst case."),
+    persist_cockpit: bool = typer.Option(False, "--persist-cockpit", help="Also upsert each construction's verdict into the cockpit DB (operational_dsn) for the dashboard."),
 ) -> None:
     """B3b: walk-forward backtest of the 3 price-only equity-factor constructions against the DEDICATED
     survivorship-free backtest warehouse (settings.backtest_warehouse_path) -> gauntlet verdict + JSON report."""
@@ -546,6 +547,17 @@ def backtest_equity(
         store.close()
     with open(out, "w") as fh:
         json.dump(report, fh, indent=2)
+    if persist_cockpit:
+        import datetime as dt
+
+        from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
+        from fxhnt.application.backtest_ingest import backtest_summaries_from_report
+        at = dt.datetime.now(dt.timezone.utc).replace(tzinfo=None)  # naive-UTC, matches forward ingest
+        repo = ForwardNavRepo(s.operational_dsn)
+        repo.migrate()
+        for sm in backtest_summaries_from_report(report):
+            repo.upsert_backtest_summary(sm, at=at)
+        typer.echo(f"backtest-equity: persisted {len(report['constructions'])} verdict(s) to cockpit")
     for c, block in report["constructions"].items():
         v = block["verdict"]
         typer.echo(f"backtest-equity {c}: passed={v['passed']} dsr={v['dsr']:.3f} "

tests/integration/test_backtest_persist_cockpit_cli.py

@@ -0,0 +1,63 @@
+"""B3b: `backtest-equity --persist-cockpit` also upserts each construction's verdict into the
+cockpit operational DB (FXHNT_OPERATIONAL_DSN) so the dashboard can read it. No network: a
+synthetic backtest warehouse + a temp-file sqlite cockpit DSN."""
+from __future__ import annotations
+
+import datetime as dt
+
+from typer.testing import CliRunner
+
+from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
+from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
+
+_SPD = 86_400
+
+
+def _build_warehouse(path: str) -> DuckDbFeatureStore:
+    store = DuckDbFeatureStore(path)
+    start = dt.date(2022, 1, 3)
+    n = 520
+    drifts = {"AAA": 0.0015, "BBB": 0.0003, "CCC": -0.0010}
+    items = []
+    members = []
+    for sym, dr in drifts.items():
+        rows = []
+        px = 100.0
+        for i in range(n):
+            day = start + dt.timedelta(days=i)
+            px *= (1.0 + dr)
+            ts = (day - dt.date(1970, 1, 1)).days * _SPD
+            rows.append((ts, {"adjclose": px, "close": px, "volume": 1_000_000.0}))
+        items.append((sym, rows))
+        members.append((sym, "NYSE", "2000-01-01", "2026-06-01"))
+    store.write_features_bulk(items)
+    store.upsert_membership(members)
+    return store
+
+
+def test_cli_persist_cockpit_writes_backtest_summary_rows(tmp_path, monkeypatch) -> None:
+    import fxhnt.cli as cli
+
+    wh = str(tmp_path / "wh.duckdb")
+    _build_warehouse(wh).close()
+    out = tmp_path / "report.json"
+    cockpit_dsn = f"sqlite:///{tmp_path / 'cockpit.db'}"
+
+    monkeypatch.setenv("FXHNT_BACKTEST_WAREHOUSE_PATH", wh)
+    monkeypatch.setenv("FXHNT_OPERATIONAL_DSN", cockpit_dsn)
+    cli.get_settings.cache_clear()
+    res = CliRunner().invoke(
+        cli.app, ["backtest-equity", "--n", "10", "--out", str(out), "--persist-cockpit"]
+    )
+    cli.get_settings.cache_clear()
+
+    assert res.exit_code == 0, res.output
+    assert out.exists(), "JSON report must still be written"
+
+    # a fresh repo on the same on-disk sqlite file sees the persisted rows
+    repo = ForwardNavRepo(cockpit_dsn)
+    rows = {r.strategy_id: r for r in repo.all_backtest_summaries()}
+    assert set(rows) == {"eqfactor_long", "eqfactor_ls", "eqfactor_tilt"}
+    for r in rows.values():
+        assert isinstance(r.passed, bool)
+        assert isinstance(r.sharpe, float)

tests/integration/test_backtest_summary_repo.py

@@ -0,0 +1,115 @@
+"""Cockpit backtest_summary persistence: ORM table auto-creates via migrate(), upserts are
+idempotent (same strategy_id updates in place), and backtest_summaries_from_report maps a
+gauntlet report dict -> 3 BacktestSummary DTOs with the right strategy_ids + fields."""
+from __future__ import annotations
+
+import datetime as dt
+
+from sqlalchemy import create_engine, inspect
+
+from fxhnt.adapters.persistence.cockpit_models import CockpitBase
+from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
+from fxhnt.application.backtest_ingest import backtest_summaries_from_report
+from fxhnt.application.forward_models import BacktestSummary
+
+
+def _repo() -> ForwardNavRepo:
+    repo = ForwardNavRepo("sqlite://")
+    repo.migrate()
+    return repo
+
+
+def _sm(sid: str = "eqfactor_long") -> BacktestSummary:
+    return BacktestSummary(
+        strategy_id=sid, as_of="2026-05-29", cagr=0.12, ann_vol=0.15, sharpe=0.8,
+        max_drawdown=-0.2, passed=True, dsr=0.97, is_sharpe=0.9, oos_sharpe=0.6, pvalue=0.03,
+    )
+
+
+def test_migrate_creates_backtest_summary_table() -> None:
+    repo = _repo()
+    tables = set(inspect(repo._engine).get_table_names())
+    assert "backtest_summary" in tables
+
+
+def test_backtest_summary_table_in_metadata() -> None:
+    eng = create_engine("sqlite://", future=True)
+    CockpitBase.metadata.create_all(eng)
+    assert "backtest_summary" in set(inspect(eng).get_table_names())
+
+
+def test_upsert_backtest_summary_and_fetch() -> None:
+    repo = _repo()
+    repo.upsert_backtest_summary(_sm(), at=dt.datetime(2026, 6, 14))
+    got = {r.strategy_id: r for r in repo.all_backtest_summaries()}
+    assert "eqfactor_long" in got
+    r = got["eqfactor_long"]
+    assert r.as_of == "2026-05-29"
+    assert r.cagr == 0.12
+    assert r.ann_vol == 0.15
+    assert r.sharpe == 0.8
+    assert r.max_drawdown == -0.2
+    assert r.passed is True
+    assert r.dsr == 0.97
+    assert r.is_sharpe == 0.9
+    assert r.oos_sharpe == 0.6
+    assert r.pvalue == 0.03
+
+
+def test_upsert_backtest_summary_is_idempotent() -> None:
+    repo = _repo()
+    repo.upsert_backtest_summary(_sm(), at=dt.datetime(2026, 6, 14))
+    # same strategy_id, updated value -> 1 row, new value
+    updated = BacktestSummary(
+        strategy_id="eqfactor_long", as_of="2026-06-01", cagr=0.20, ann_vol=0.15, sharpe=1.1,
+        max_drawdown=-0.1, passed=False, dsr=0.50, is_sharpe=1.0, oos_sharpe=0.2, pvalue=0.40,
+    )
+    repo.upsert_backtest_summary(updated, at=dt.datetime(2026, 6, 15))
+    rows = repo.all_backtest_summaries()
+    assert len(rows) == 1
+    assert rows[0].cagr == 0.20
+    assert rows[0].passed is False
+    assert rows[0].as_of == "2026-06-01"
+
+
+def _report() -> dict:
+    def block(passed: bool) -> dict:
+        return {
+            "stats": {"cagr": 0.1, "ann_vol": 0.2, "sharpe": 0.5, "max_drawdown": -0.3, "n_obs": 400},
+            "verdict": {"passed": passed, "dsr": 0.96, "is_sharpe": 0.7, "oos_sharpe": 0.4,
+                        "n_trials": 3, "reasons": [], "pvalue": 0.02},
+        }
+    return {
+        "constructions": {"long": block(True), "ls": block(False), "tilt": block(True)},
+        "rebalance_dates": ["2025-01-31", "2025-02-28", "2026-05-29"],
+        "n_names_avg": 120.0,
+        "settings": {},
+    }
+
+
+def test_backtest_summaries_from_report_maps_3_constructions() -> None:
+    sms = backtest_summaries_from_report(_report())
+    by_sid = {s.strategy_id: s for s in sms}
+    assert set(by_sid) == {"eqfactor_long", "eqfactor_ls", "eqfactor_tilt"}
+    # as_of is the LAST rebalance date
+    for s in sms:
+        assert s.as_of == "2026-05-29"
+    # stats fields
+    assert by_sid["eqfactor_long"].cagr == 0.1
+    assert by_sid["eqfactor_long"].ann_vol == 0.2
+    assert by_sid["eqfactor_long"].sharpe == 0.5
+    assert by_sid["eqfactor_long"].max_drawdown == -0.3
+    # verdict fields
+    assert by_sid["eqfactor_long"].passed is True
+    assert by_sid["eqfactor_ls"].passed is False
+    assert by_sid["eqfactor_long"].dsr == 0.96
+    assert by_sid["eqfactor_long"].is_sharpe == 0.7
+    assert by_sid["eqfactor_long"].oos_sharpe == 0.4
+    assert by_sid["eqfactor_long"].pvalue == 0.02
+
+
+def test_backtest_summaries_from_report_empty_rebalance_dates() -> None:
+    rep = _report()
+    rep["rebalance_dates"] = []
+    sms = backtest_summaries_from_report(rep)
+    assert all(s.as_of == "" for s in sms)