178f0388d9
book_empty was `not book` (the aggregated returns dict), so deploy members present but with no overlapping forward_nav data (e.g. a just-re-inceptioned deploy track) read as "first edge" and AUTO-PROMOTED with no correlation check — a one-way false-promote on a capital-adjacent gate. Now `not book_sids` (membership): no deploy members → promote (true first edge); members present but no data → indeterminate → WITHHOLD, per spec. + regression test. Also neutralizes the real registry's deploy-tier tracks in two pre-existing promotion tests that were unknowingly relying on the old inversion (they didn't seed nav data for the real deploy book). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
121 lines
6.6 KiB
Python
121 lines
6.6 KiB
Python
"""Auto-promotion research→deploy: a track flagged `promote_on_pass` in the registry (multistrat) is
|
|
promoted the moment its forward gate reads PASS. Promotion is ONE-WAY + persistent (never reverts if the
|
|
gate later dips), and only flagged tracks are promoted. The fleet's effective tier reflects the promotion.
|
|
File-backed sqlite so the seed repo and evaluate_promotions' own repo share state."""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
|
|
from fxhnt import registry
|
|
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
|
|
from fxhnt.application.dashboard_service import DashboardService
|
|
from fxhnt.application.forward_ingest import evaluate_promotions
|
|
from fxhnt.application.forward_models import ForwardNavRow as NavRowDTO
|
|
from fxhnt.application.forward_models import ForwardSummary
|
|
|
|
_AT = dt.datetime(2026, 7, 20)
|
|
|
|
|
|
def _summary(sid: str, days: int) -> ForwardSummary:
|
|
return ForwardSummary(sid, "2026-07-20", days, 1.1, 0.1, 1.5, -0.04)
|
|
|
|
|
|
def _clear_deploy_book(monkeypatch) -> None:
|
|
"""Neutralize the REAL registry's deploy-tier tracks (e.g. bybit_4edge, positioning) so tests that
|
|
don't seed nav data for them aren't read as "deploy members with no data" (indeterminate → withhold)
|
|
under the corrected backstop. Keeps these tests focused on the single track under test, as before."""
|
|
for sid, meta in list(registry.STRATEGY_REGISTRY.items()):
|
|
if meta.get("tier") == "deploy":
|
|
monkeypatch.setitem(registry.STRATEGY_REGISTRY, sid, {**meta, "tier": "research"})
|
|
|
|
|
|
def test_promotes_flagged_track_on_pass_not_unflagged(tmp_path, monkeypatch) -> None:
|
|
_clear_deploy_book(monkeypatch)
|
|
dsn = f"sqlite:///{tmp_path / 'c.db'}"
|
|
repo = ForwardNavRepo(dsn)
|
|
repo.migrate()
|
|
repo.upsert_summary(_summary("multistrat", 20), "PASS", "reconciles", _AT) # flagged, PASS → promote
|
|
repo.upsert_summary(_summary("crypto_tstrend", 20), "PASS", "reconciles", _AT) # NOT flagged → never promote
|
|
assert evaluate_promotions(dsn) == ["multistrat"]
|
|
assert ForwardNavRepo(dsn).promoted_strategy_ids() == {"multistrat"}
|
|
|
|
|
|
def test_no_promotion_while_gate_waits(tmp_path) -> None:
|
|
dsn = f"sqlite:///{tmp_path / 'c.db'}"
|
|
repo = ForwardNavRepo(dsn)
|
|
repo.migrate()
|
|
repo.upsert_summary(_summary("multistrat", 14), "WAIT", "building (14/20 days)", _AT)
|
|
assert evaluate_promotions(dsn) == []
|
|
assert ForwardNavRepo(dsn).promoted_strategy_ids() == set()
|
|
|
|
|
|
def test_promotion_is_one_way_and_persists_when_gate_later_dips(tmp_path, monkeypatch) -> None:
|
|
_clear_deploy_book(monkeypatch)
|
|
dsn = f"sqlite:///{tmp_path / 'c.db'}"
|
|
repo = ForwardNavRepo(dsn)
|
|
repo.migrate()
|
|
repo.upsert_summary(_summary("multistrat", 20), "PASS", "reconciles", _AT)
|
|
assert evaluate_promotions(dsn) == ["multistrat"]
|
|
# the gate later dips to WAIT (diverging) — the promotion must NOT be re-fired NOR reverted.
|
|
repo.upsert_summary(_summary("multistrat", 21), "WAIT", "diverging", _AT)
|
|
assert evaluate_promotions(dsn) == [] # not re-promoted (idempotent)
|
|
assert "multistrat" in ForwardNavRepo(dsn).promoted_strategy_ids() # stays promoted (one-way)
|
|
|
|
|
|
def test_fleet_effective_tier_flips_to_deploy_on_promotion(tmp_path) -> None:
|
|
dsn = f"sqlite:///{tmp_path / 'c.db'}"
|
|
repo = ForwardNavRepo(dsn)
|
|
repo.migrate() # seeds the registry (multistrat starts tier=research)
|
|
before = {f.strategy_id: f for f in DashboardService(repo).fleet()}
|
|
assert before["multistrat"].tier == "research"
|
|
repo.promote("multistrat", _AT)
|
|
after = {f.strategy_id: f for f in DashboardService(repo).fleet()}
|
|
assert after["multistrat"].tier == "deploy" # the persisted promotion overrides the registry tier
|
|
|
|
|
|
def test_promotion_withheld_when_correlated_and_fires_when_diversifying(tmp_path, monkeypatch):
|
|
dsn = f"sqlite:///{tmp_path}/promo.db"
|
|
repo = ForwardNavRepo(dsn)
|
|
repo.migrate()
|
|
at = dt.datetime(2026, 7, 9, 23, 30)
|
|
# a deploy-tier book track "d" and a PASS candidate "c" flagged promote_on_pass
|
|
monkeypatch.setitem(registry.STRATEGY_REGISTRY, "d",
|
|
{"display_name": "D", "sleeve": "x", "gate_spec": {}, "tier": "deploy"})
|
|
monkeypatch.setitem(registry.STRATEGY_REGISTRY, "c",
|
|
{"display_name": "C", "sleeve": "x", "gate_spec": {}, "tier": "research",
|
|
"promote_on_pass": True})
|
|
dates = [f"2026-06-{d:02d}" for d in range(1, 21)]
|
|
dret = [0.01 if i % 2 else -0.01 for i in range(20)]
|
|
for d, r in zip(dates, dret, strict=True):
|
|
repo.upsert_rows([NavRowDTO(strategy_id="d", date=d, ret=r, nav=1.0)], at=at)
|
|
repo.upsert_summary(ForwardSummary(strategy_id="c", as_of=dates[-1], days=20, nav=1.0,
|
|
total_return=0.0, sharpe=0.0, maxdd=0.0), "PASS", "ok", at)
|
|
# candidate PERFECTLY correlated with the book → withheld
|
|
for d, r in zip(dates, dret, strict=True):
|
|
repo.upsert_rows([NavRowDTO(strategy_id="c", date=d, ret=r, nav=1.0)], at=at)
|
|
assert evaluate_promotions(dsn) == []
|
|
# candidate ANTI-correlated (diversifying) → promoted
|
|
for d, r in zip(dates, dret, strict=True):
|
|
repo.upsert_rows([NavRowDTO(strategy_id="c", date=d, ret=-r, nav=1.0)], at=at)
|
|
assert evaluate_promotions(dsn) == ["c"]
|
|
|
|
|
|
def test_promotion_withheld_when_deploy_member_has_no_data(tmp_path, monkeypatch):
|
|
# a deploy-tier member EXISTS but has NO forward_nav rows → book membership non-empty, aggregated data
|
|
# empty → INDETERMINATE, must WITHHOLD (not a first-edge auto-promote). Guards the book_empty inversion.
|
|
dsn = f"sqlite:///{tmp_path}/nodata.db"
|
|
repo = ForwardNavRepo(dsn)
|
|
repo.migrate()
|
|
at = dt.datetime(2026, 7, 9, 23, 30)
|
|
monkeypatch.setitem(registry.STRATEGY_REGISTRY, "d",
|
|
{"display_name": "D", "sleeve": "x", "gate_spec": {}, "tier": "deploy"})
|
|
monkeypatch.setitem(registry.STRATEGY_REGISTRY, "c",
|
|
{"display_name": "C", "sleeve": "x", "gate_spec": {}, "tier": "research",
|
|
"promote_on_pass": True})
|
|
# candidate c has data + PASS; deploy member d has NO forward_nav rows at all
|
|
for i, dd in enumerate([f"2026-06-{x:02d}" for x in range(1, 21)]):
|
|
repo.upsert_rows([NavRowDTO(strategy_id="c", date=dd, ret=0.01 if i % 2 else -0.01, nav=1.0)], at=at)
|
|
repo.upsert_summary(ForwardSummary(strategy_id="c", as_of="2026-06-20", days=20, nav=1.0,
|
|
total_return=0.0, sharpe=0.0, maxdd=0.0), "PASS", "ok", at)
|
|
assert evaluate_promotions(dsn) == [] # deploy member present but no data → indeterminate → WITHHELD
|