391400f233
The broad-Binance-universe momentum book (venue binance+glbx) was a survivorship/ microcap mirage superseded by the isolated crypto_tstrend edge + bybit_4edge. Removed the combined_forward_nav Dagster asset (+ its cockpit_forward/paper_book_snapshot deps), the registry entry (DB registry row auto-prunes via _seed_registry), the dashboard headline preference, and the dead migration builder. CombinedBook modules retained (still used by the forward-track CLI). Test fixtures repointed combined -> crypto_tstrend. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
300 lines
15 KiB
Python
300 lines
15 KiB
Python
"""END-TO-END forward-gate foundation test — the invariant net that the prior three prod resets/freezes lacked.
|
|
|
|
Real capital gates on the Bybit forward track. It reset/froze THREE times because every prior fix "passed
|
|
tests" yet the tests never simulated the actual failure mode: many nightly runs over ADVANCING dates, with a
|
|
crash-induced state loss in the middle. This drives the REAL `ForwardTracker` (both a deterministic nightly
|
|
strategy that mirrors the live cadence AND, network-free, the real `BybitFourEdgeStrategy` over a seeded
|
|
in-memory store) across ~20 simulated nights and asserts the three invariants that were violated in prod:
|
|
|
|
1. MONOTONIC ACCUMULATION — one `step()` per advancing day: forward_days goes 0→1→2→…, inception NEVER
|
|
changes after the first run, and the reconciliation gate goes WAIT("building N/14") → PASS at day ≥ 14.
|
|
2. SURVIVES CORRUPTION / STATE-LOSS / RESTART — mid-sequence the state JSON is lost / truncated / partially
|
|
written (the crash signature). The tracker must NEVER silently re-inception (forward_days→0, new T0): it
|
|
EITHER recovers the prior days from the `.bak` OR fails loudly (raises). Accumulation then continues from
|
|
where it was, not from 0.
|
|
3. ATOMIC WRITE — an interrupted save never destroys the last good state: the previous state stays loadable.
|
|
|
|
NO network — the real-strategy leg uses an in-memory SQLite feature store; external inputs are seeded.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from fxhnt.adapters.orchestration.assets import build_bybit_4edge_forward_nav
|
|
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
|
|
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
|
|
from fxhnt.application.bybit_forward_track import BybitFourEdgeStrategy
|
|
from fxhnt.application.forward_models import ForwardNavRow, ForwardSummary
|
|
from fxhnt.application.forward_tracker import ForwardTracker
|
|
from fxhnt.application.gate_policy import POLICY
|
|
from fxhnt.application.gate_policy_resolve import resolve_policy
|
|
from fxhnt.application.reconciliation_gate import BacktestReference, evaluate_reconciliation_gate
|
|
from fxhnt.registry import STRATEGY_REGISTRY
|
|
|
|
_DAY = 86_400
|
|
|
|
|
|
def _read_days_list_state(path: str, sid: str) -> tuple[ForwardSummary, list[ForwardNavRow]]:
|
|
"""Read the days-list state file `ForwardTracker` writes directly — the retired `ForwardStateReader`'s
|
|
days-list branch, inlined here since this is the ONLY shape `ForwardTracker` ever produces (the gate
|
|
ETL now reads the DB record instead; this test drives the standalone JSON-based tracker + gate math
|
|
directly, network-free)."""
|
|
state = json.loads(Path(path).read_text())
|
|
days = state["days"]
|
|
rows = [ForwardNavRow(strategy_id=sid, date=d["date"], ret=float(d["ret"]), nav=float(d["nav"]))
|
|
for d in days]
|
|
nav = float(state.get("nav", rows[-1].nav if rows else 1.0))
|
|
as_of = state.get("last_date") or (rows[-1].date if rows else "")
|
|
summary = ForwardSummary(strategy_id=sid, as_of=as_of, days=len(rows), nav=nav,
|
|
total_return=nav - 1.0, sharpe=0.0, maxdd=0.0)
|
|
return summary, rows
|
|
|
|
|
|
class _NightlyRecomputable:
|
|
"""Mirrors the live nightly cadence: a recomputable strategy whose known `(iso_date, ret)` series grows by
|
|
exactly ONE contiguous calendar day per simulated night (exactly like `BybitFourEdgeStrategy` returning the
|
|
full series each run — the tracker books only the days strictly after `last_date`). Deterministic return so
|
|
the gate transition and accumulation are exact."""
|
|
|
|
def __init__(self, start: dt.date, daily_ret: float) -> None:
|
|
self._start = start
|
|
self._daily = daily_ret
|
|
self._n = 0
|
|
self._series: list[tuple[str, float]] = []
|
|
|
|
def add_night(self) -> None:
|
|
d = self._start + dt.timedelta(days=self._n)
|
|
self._series.append((d.isoformat(), self._daily))
|
|
self._n += 1
|
|
|
|
def advance(self, last_date, extra): # noqa: ANN001, ANN201 — ForwardStrategy port
|
|
return list(self._series), dict(extra)
|
|
|
|
|
|
def _build_to_day(strat: _NightlyRecomputable, path: str, forward_days: int):
|
|
"""Freeze inception, then run one `step()` per night until `forward_days` days are booked."""
|
|
strat.add_night()
|
|
st = ForwardTracker(strat, path).step() # first run freezes T0, books nothing
|
|
assert st.forward_days == 0
|
|
for _ in range(forward_days):
|
|
strat.add_night()
|
|
st = ForwardTracker(strat, path).step()
|
|
assert st.forward_days == forward_days
|
|
return st
|
|
|
|
|
|
# ---- invariant 1: monotonic accumulation + gate WAIT→PASS ---------------------------------------------
|
|
|
|
def test_monotonic_accumulation_and_gate_waits_then_passes(tmp_path) -> None:
|
|
p = str(tmp_path / "bybit_4edge_state.json")
|
|
strat = _NightlyRecomputable(dt.date(2026, 7, 1), 0.002)
|
|
spec = STRATEGY_REGISTRY["bybit_4edge"]["gate_spec"]
|
|
policy = resolve_policy(POLICY, spec)
|
|
# bybit_4edge uses a longer 21-day window (lumpy edge; median ~17d between spikes), not the 14-day default.
|
|
min_days = policy.min_forward_days
|
|
assert min_days == 21
|
|
|
|
strat.add_night()
|
|
st = ForwardTracker(strat, p).step() # inception
|
|
assert st.forward_days == 0
|
|
inception = st.inception
|
|
|
|
seen = [0]
|
|
for night in range(1, min_days + 2): # run past the 21-day threshold to observe WAIT→PASS
|
|
strat.add_night()
|
|
st = ForwardTracker(strat, p).step()
|
|
assert st.inception == inception, "inception must NEVER move after the first run"
|
|
assert st.forward_days == night, "forward_days must accumulate 1 per advancing day"
|
|
seen.append(st.forward_days)
|
|
|
|
summary, rows = _read_days_list_state(p, "bybit_4edge")
|
|
# a backtest reference the forward reconciles with (positive edge, modest expectation).
|
|
bt = BacktestReference(expected_window_return=0.001,
|
|
daily_returns=[0.002] * st.forward_days)
|
|
verdict = evaluate_reconciliation_gate(summary, rows, bt, policy)
|
|
if st.forward_days < min_days:
|
|
assert verdict.status == "WAIT" and "building" in verdict.reason.lower(), verdict
|
|
else:
|
|
assert verdict.status == "PASS", verdict.reason
|
|
|
|
assert seen == list(range(0, min_days + 2)), "accumulation must be strictly monotonic"
|
|
|
|
|
|
# ---- invariant 2: survives corruption / state-loss / restart -----------------------------------------
|
|
|
|
def test_lost_state_file_does_not_silently_reinception(tmp_path) -> None:
|
|
"""THE PROD RESET: the state file is LOST (PVC loss / redeploy). On the broken foundation `_load` returns
|
|
None → the tracker silently re-inceptions (new T0, forward_days=0). It MUST instead recover from the `.bak`
|
|
and continue accumulating — never silently reset to day 0."""
|
|
p = str(tmp_path / "bybit_4edge_state.json")
|
|
strat = _NightlyRecomputable(dt.date(2026, 7, 1), 0.002)
|
|
st = _build_to_day(strat, p, 8)
|
|
inception = st.inception
|
|
|
|
os.remove(p) # crash: the live state is gone
|
|
strat.add_night()
|
|
st = ForwardTracker(strat, p).step()
|
|
assert st.inception == inception, "must NOT re-inception on a lost state file"
|
|
assert st.forward_days == 9, "must recover the prior 8 days from .bak and book the new one"
|
|
|
|
|
|
def test_empty_truncated_main_recovers_from_bak(tmp_path) -> None:
|
|
"""The non-atomic-writer crash signature: `open(w)` truncated the file, then SIGKILL before the write
|
|
finished → an EMPTY file. The tracker must recover from the `.bak`, not reset and not stay dead."""
|
|
p = str(tmp_path / "bybit_4edge_state.json")
|
|
strat = _NightlyRecomputable(dt.date(2026, 7, 1), 0.002)
|
|
st = _build_to_day(strat, p, 8)
|
|
inception = st.inception
|
|
|
|
Path(p).write_text("") # truncated-mid-save
|
|
strat.add_night()
|
|
st = ForwardTracker(strat, p).step()
|
|
assert st.inception == inception
|
|
assert st.forward_days == 9
|
|
|
|
|
|
def test_partial_json_main_recovers_from_bak(tmp_path) -> None:
|
|
p = str(tmp_path / "bybit_4edge_state.json")
|
|
strat = _NightlyRecomputable(dt.date(2026, 7, 1), 0.002)
|
|
st = _build_to_day(strat, p, 8)
|
|
inception = st.inception
|
|
|
|
Path(p).write_text('{"inception": "2026-07-01", "days": [') # partial JSON
|
|
strat.add_night()
|
|
st = ForwardTracker(strat, p).step()
|
|
assert st.inception == inception
|
|
assert st.forward_days == 9
|
|
|
|
|
|
def test_both_main_and_bak_unreadable_fails_loudly_never_resets(tmp_path) -> None:
|
|
"""If BOTH the state and its backup are gone/corrupt (but the track HAS existed), the tracker must FAIL
|
|
LOUDLY — never silently re-inception to day 0. A loud failure surfaces in the asset; a silent reset does
|
|
not (this is the mode that burned real capital)."""
|
|
p = str(tmp_path / "bybit_4edge_state.json")
|
|
strat = _NightlyRecomputable(dt.date(2026, 7, 1), 0.002)
|
|
_build_to_day(strat, p, 8)
|
|
|
|
Path(p).write_text("")
|
|
Path(p + ".bak").write_text("") # both destroyed
|
|
strat.add_night()
|
|
with pytest.raises((RuntimeError, ValueError)):
|
|
ForwardTracker(strat, p).step()
|
|
# and it must NOT have written a fresh day-0 inception in the process.
|
|
if Path(p).exists() and Path(p).read_text().strip():
|
|
st = json.loads(Path(p).read_text())
|
|
assert st.get("days"), "must not have reset to an empty days-list"
|
|
|
|
|
|
# ---- invariant 3: atomic write preserves the previous good state -------------------------------------
|
|
|
|
def test_interrupted_save_preserves_previous_good_state(tmp_path, monkeypatch) -> None:
|
|
"""A crash DURING the save (json.dump raises) must NEVER leave a readable-but-partial main file. With an
|
|
atomic tmp+replace writer the previous good state stays intact and loadable; the broken `open(w)` writer
|
|
truncates it and loses everything."""
|
|
import fxhnt.application.forward_tracker as ft
|
|
|
|
p = str(tmp_path / "bybit_4edge_state.json")
|
|
strat = _NightlyRecomputable(dt.date(2026, 7, 1), 0.002)
|
|
_build_to_day(strat, p, 5)
|
|
good = Path(p).read_text()
|
|
assert json.loads(good)["days"], "sanity: a good state exists before the interrupted save"
|
|
|
|
def _boom(*_a, **_k):
|
|
raise RuntimeError("SIGKILL during json.dump")
|
|
|
|
monkeypatch.setattr(ft.json, "dump", _boom)
|
|
strat.add_night()
|
|
with pytest.raises(RuntimeError):
|
|
ForwardTracker(strat, p).step()
|
|
monkeypatch.undo()
|
|
|
|
# the previous good state must be intact and loadable — the interrupted save did not destroy it.
|
|
st = ForwardTracker(strat, p).step()
|
|
assert st.forward_days == 6, "must continue from day 5, not reset after the interrupted save"
|
|
assert json.loads(Path(p).read_text())["days"]
|
|
|
|
|
|
# ---- integration: the REAL BybitFourEdgeStrategy over a seeded store, network-free -------------------
|
|
|
|
def _seed_momentum(store: TimescaleFeatureStore, *, n_symbols: int = 8, days: int, drift: float = 0.01) -> None:
|
|
for idx in range(n_symbols):
|
|
sign = 1.0 if idx % 2 == 0 else -1.0
|
|
px = 100.0
|
|
rows = []
|
|
for d in range(days):
|
|
wobble = 0.003 * math.cos(0.4 * d + idx)
|
|
px *= (1.0 + sign * drift + wobble)
|
|
rows.append((d * _DAY, {"close": px}))
|
|
store.write_features(f"SYM{idx:02d}USDT", rows)
|
|
|
|
|
|
def _seed_carry(store: TimescaleFeatureStore, *, days: int) -> None:
|
|
funding = {"AAAUSDT": 0.001, "BBBUSDT": 0.001, "CCCUSDT": 0.001,
|
|
"XXXUSDT": -0.001, "YYYUSDT": -0.001, "ZZZUSDT": -0.001}
|
|
for idx, (sym, fund) in enumerate(funding.items()):
|
|
rows = []
|
|
for d in range(days):
|
|
f = fund + 0.0002 * math.cos(0.5 * d + idx)
|
|
rows.append((d * _DAY, {"funding": f, "close": 100.0, "spot_close": 100.0}))
|
|
store.write_features(sym, rows)
|
|
|
|
|
|
def test_real_bybit_strategy_accumulates_through_the_deterministic_engine(tmp_path) -> None:
|
|
"""The REAL `BybitFourEdgeStrategy`, run through `forward_engine.run_track` (DB anchor + recompute)
|
|
instead of the retired JSON `ForwardTracker`: T0 freezes on the first run and NEVER moves as more data
|
|
lands (same definition_hash -> no re-inception), and the recomputed day count is monotonically
|
|
non-decreasing as the window grows — the same MONOTONIC ACCUMULATION invariant this file's synthetic-
|
|
strategy tests establish above, now proven with the real production strategy.
|
|
|
|
The old JSON STATE-LOSS/RESTART scenario (`os.remove(p)` mid-sequence, invariant 2 above) does not apply
|
|
to this track anymore: recompute mode has no accumulation state to lose in the first place — the DB
|
|
anchor (just T0 + a definition hash) is the ONLY persistent state, and the daily record is always
|
|
freshly recomputed from the warehouse on every run, never appended incrementally to a mutable file. The
|
|
crash-recovery contract itself (`ForwardTracker` + its `.bak`) is unchanged and still covered above by
|
|
the synthetic-strategy invariant-2 tests, since the class remains in use by the standalone `fxhnt
|
|
forward-track` CLI command (`cli.py`), the last caller left after the Dagster forward tracks moved onto
|
|
the engine (the crypto-momentum `combined` book that used this path was retired 2026-07-11)."""
|
|
repo = ForwardNavRepo(f"sqlite:///{tmp_path}/x.db")
|
|
repo.migrate()
|
|
at = dt.datetime(2026, 1, 1)
|
|
|
|
seed0 = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_momentum(seed0, n_symbols=8, days=60)
|
|
_seed_carry(seed0, days=60)
|
|
t0 = BybitFourEdgeStrategy(seed0, cost_bps=0.0).advance(None, {})[0][-1][0]
|
|
seed0.close()
|
|
|
|
def run(days: int):
|
|
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
|
|
_seed_momentum(store, n_symbols=8, days=days)
|
|
_seed_carry(store, days=days)
|
|
try:
|
|
return build_bybit_4edge_forward_nav(repo, store, today=t0, at=at, cost_bps=0.0)
|
|
finally:
|
|
store.close()
|
|
|
|
out0 = run(60) # freeze T0 (recompute mode books T0 itself → day 1)
|
|
assert out0["days"] == 1
|
|
inception = out0["t0"]
|
|
|
|
last = out0["days"]
|
|
for extra in range(1, 12): # 11 advancing nights
|
|
out = run(60 + extra)
|
|
assert out["t0"] == inception, "real-strategy T0 must be frozen across nights"
|
|
assert out["reinception"] is False
|
|
assert out["days"] >= last, "real-strategy day count must be non-decreasing"
|
|
last = out["days"]
|
|
assert last > 1, "the real strategy must actually recompute additional forward days"
|
|
|
|
# idempotent re-run on UNCHANGED data — the recompute-engine analogue of "no state loss": the record is
|
|
# deterministically derived from [T0, today]'s warehouse data, so repeating a run never drifts.
|
|
out_again = run(60 + 11)
|
|
assert out_again["t0"] == inception
|
|
assert out_again["days"] == last
|