gitadmin/fxhnt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(vrp): delete XspOptionBarsRepo class + _build_vrp migration builder (keep table + XspOptionBarRow)

Browse Source
This commit is contained in:
Jeroen Grusewski
2026-07-20 23:31:52 +00:00
parent 5b5fbb0e0b
commit fc663215ef
4 changed files with 25 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/fxhnt/adapters/orchestration/migration_builders.py
View File

@@ -68,15 +68,6 @@ def track_builders(settings: Settings, data_dir: str) -> TrackBuilders:
# builders here are gone with them. crypto_tstrend the BYBIT BOOK SLEEVE is unaffected — that lives
# under `bybit_4edge`'s own builder below, not here.)
# ---- vrp_nav: defined-risk XSP put-spread VRP on the frozen OPRA PIT table (recompute) ----------------
def _build_vrp() -> Any:
from fxhnt.adapters.persistence.xsp_option_bars import XspOptionBarsRepo
from fxhnt.application.vrp_book import VrpStrategy
return VrpStrategy(XspOptionBarsRepo(settings.operational_dsn))
builders["vrp"] = ("vrp_state", _build_vrp)
# ---- unlock_nav: single-sleeve unlock edge on the liquid universe (Phase 0b Task 2: own Bybit track, ----
# ---- the SAME single-sleeve construction as xsfunding/positioning, mirrored here; carries the unlock ----
# ---- calendar the sleeve needs, unlike xsfunding/positioning) --------------------------------------------

119
src/fxhnt/adapters/persistence/xsp_option_bars.py
View File

@@ -1,119 +0,0 @@
"""PIT store for frozen XSP option daily bars — the deterministic substrate the VRP recompute reads. OPRA is
hit only when FILLING this table (asset/backfill); VrpStrategy never calls OPRA. Same freeze discipline as the
equity Yahoo-PIT snapshot."""
from __future__ import annotations
import datetime as dt
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session
from fxhnt.adapters.persistence.cockpit_models import CockpitBase, XspOptionBarRow
from fxhnt.domain.models import OptionContract
class XspOptionBarsRepo:
def __init__(self, dsn: str) -> None:
kw = {} if dsn.startswith("postgresql") else {"connect_args": {"check_same_thread": False}}
self._engine = create_engine(dsn, **kw)
self._pre: dict | None = None # in-memory replay cache (None = not preloaded; per-day DB path)
def migrate(self) -> None:
CockpitBase.metadata.create_all(self._engine)
def preload(self, after: str | None = None) -> None:
"""Bulk-load every frozen bar for days strictly after `after` (all days if None) into an in-memory
index, in ONE query. A full VrpStrategy replay then serves chain_asof/bars_for/trading_days from
memory instead of a fresh Session + round-trip per day — the N+1 that made a 1600-day replay issue
thousands of queries. The live single-day exec path (plan_and_record_vrp) never calls this, so it
keeps its cheap per-day DB queries unchanged. The cache is a faithful mirror of the covered range:
every cached method falls back to the DB for any date OUTSIDE that range, so it can never
under-return. Idempotent (re-call to re-scope)."""
lo = dt.date.fromisoformat(after) if after else None
marks: dict[str, dict[str, float]] = {}
rows_by_date: dict[str, list[tuple[str, dt.date, float, str]]] = {}
with Session(self._engine) as s:
q = select(XspOptionBarRow.osi_symbol, XspOptionBarRow.date, XspOptionBarRow.expiry,
XspOptionBarRow.strike, XspOptionBarRow.right, XspOptionBarRow.close)
if lo is not None:
q = q.where(XspOptionBarRow.date > lo)
for osi, date, expiry, strike, right, close in s.execute(q):
di = date.isoformat()
marks.setdefault(di, {})[osi] = close
rows_by_date.setdefault(di, []).append((osi, expiry, strike, right))
self._pre = {"after": after, "marks": marks, "rows": rows_by_date, "days": sorted(rows_by_date)}
def _pre_covers(self, date_iso: str) -> bool:
"""True iff `date_iso` is inside the preloaded range (strictly after the preload bound)."""
pre = self._pre
return pre is not None and (pre["after"] is None or date_iso > pre["after"])
def upsert_bars(self, rows: list[dict], at: dt.datetime) -> None:
with Session(self._engine) as s:
for r in rows:
obj = XspOptionBarRow(
osi_symbol=r["osi_symbol"], date=dt.date.fromisoformat(r["date"]),
expiry=dt.date.fromisoformat(r["expiry"]), strike=float(r["strike"]),
right=r["right"], close=float(r["close"]), src_updated_at=at)
s.merge(obj) # merge = insert-or-overwrite on the (osi_symbol, date) PK
s.commit()
def bars_for(self, osi_symbols: list[str], start: str, end: str) -> dict[str, dict[str, float]]:
if not osi_symbols:
return {}
# Preloaded fast path — serve from memory when the whole [start, end] range is inside the cache
# (start > preload bound => the entire range is, since start <= end). Iso date strings compare
# lexically like dt.date, so the range test matches the DB WHERE below.
if self._pre is not None and self._pre_covers(start):
want = set(osi_symbols)
out: dict[str, dict[str, float]] = {}
if start == end: # the replay's single-day case (hot path)
m = self._pre["marks"].get(start, {})
return {osi: {start: m[osi]} for osi in want if osi in m}
for di, m in self._pre["marks"].items():
if start <= di <= end:
for osi in want:
if osi in m:
out.setdefault(osi, {})[di] = m[osi]
return out
lo, hi = dt.date.fromisoformat(start), dt.date.fromisoformat(end)
out: dict[str, dict[str, float]] = {}
with Session(self._engine) as s:
q = select(XspOptionBarRow).where(
XspOptionBarRow.osi_symbol.in_(osi_symbols),
XspOptionBarRow.date >= lo, XspOptionBarRow.date <= hi)
for row in s.scalars(q):
out.setdefault(row.osi_symbol, {})[row.date.isoformat()] = row.close
return out
def trading_days(self, after: str | None) -> list[str]:
"""Ascending distinct dates present in the table, strictly after `after` (or all if None)."""
# Preloaded fast path — the cache holds every day > its bound, so it can answer any query whose
# `after` is at or beyond that bound (a full preload, after=None, answers everything).
pre = self._pre
if pre is not None and (pre["after"] is None or (after is not None and after >= pre["after"])):
return [d for d in pre["days"] if after is None or d > after]
from sqlalchemy import distinct
with Session(self._engine) as s:
q = select(distinct(XspOptionBarRow.date)).order_by(XspOptionBarRow.date)
days = [d.isoformat() for d in s.scalars(q)]
return [d for d in days if after is None or d > after]
def chain_asof(self, date: str, dte_lo: int, dte_hi: int, right: str = "P") -> list[OptionContract]:
d = dt.date.fromisoformat(date)
exp_lo, exp_hi = d + dt.timedelta(days=dte_lo), d + dt.timedelta(days=dte_hi)
# Preloaded fast path — same filter (date == d, right, expiry in [exp_lo, exp_hi]) + strike order.
if self._pre is not None and self._pre_covers(date):
hits = [(osi, exp, strike, r) for (osi, exp, strike, r) in self._pre["rows"].get(date, [])
if r == right and exp_lo <= exp <= exp_hi]
hits.sort(key=lambda t: t[2]) # deterministic strike order (matches ORDER BY)
return [OptionContract(osi_symbol=osi, expiry=exp, strike=strike, right=r)
for (osi, exp, strike, r) in hits]
with Session(self._engine) as s:
q = select(XspOptionBarRow).where(
XspOptionBarRow.date == d, XspOptionBarRow.right == right,
XspOptionBarRow.expiry >= exp_lo, XspOptionBarRow.expiry <= exp_hi
).order_by(XspOptionBarRow.strike) # deterministic strike order (no ORDER BY = arbitrary
# row order on Postgres — the ATM pairing below relies on stable iteration)
return [OptionContract(osi_symbol=r.osi_symbol, expiry=r.expiry, strike=r.strike, right=r.right)
for r in s.scalars(q)]

36
tests/integration/test_forward_health.py
View File

@@ -5,8 +5,11 @@ from __future__ import annotations
import datetime as dt
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from fxhnt.adapters.persistence.cockpit_models import CockpitBase, XspOptionBarRow
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.adapters.persistence.xsp_option_bars import XspOptionBarsRepo
from fxhnt.application.forward_health import (
BROKEN,
HEALTHY,
@@ -97,6 +100,21 @@ def _seed_summary(repo: ForwardNavRepo, sid: str, as_of: str, at: dt.datetime) -
gate_status="WAIT", gate_reason="building", at=at)
def _seed_freeze_bars(dsn: str, rows: list[dict], at: dt.datetime) -> None:
"""Insert rows directly into the kept `xsp_option_bars` table (no VRP repo class involved) — mirrors
what the retired XspOptionBarsRepo.migrate()/upsert_bars() used to do, for the freeze-staleness axis."""
kw = {} if dsn.startswith("postgresql") else {"connect_args": {"check_same_thread": False}}
engine = create_engine(dsn, **kw)
CockpitBase.metadata.create_all(engine)
with Session(engine) as s:
for r in rows:
s.merge(XspOptionBarRow(
osi_symbol=r["osi_symbol"], date=dt.date.fromisoformat(r["date"]),
expiry=dt.date.fromisoformat(r["expiry"]), strike=float(r["strike"]),
right=r["right"], close=float(r["close"]), src_updated_at=at))
s.commit()
def test_evaluate_health_persists_and_catches_stale_freeze(tmp_path):
dsn = f"sqlite:///{tmp_path}/health.db"
at = dt.datetime(2026, 7, 14, 23, 30)
@@ -111,10 +129,8 @@ def test_evaluate_health_persists_and_catches_stale_freeze(tmp_path):
# that check first).
_seed_summary(repo, "unlock", "2026-07-13", at)
# seed the vrp freeze table with a max date weeks behind the run date.
xsp = XspOptionBarsRepo(dsn)
xsp.migrate()
xsp.upsert_bars([{"osi_symbol": "XSP 260620P00470000", "date": "2026-06-15", "expiry": "2026-07-17",
"strike": 470.0, "right": "P", "close": 1.0}], at)
_seed_freeze_bars(dsn, [{"osi_symbol": "XSP 260620P00470000", "date": "2026-06-15",
"expiry": "2026-07-17", "strike": 470.0, "right": "P", "close": 1.0}], at)
verdicts = {v.strategy_id: v for v in evaluate_health(dsn, at=at)}
assert verdicts["vrp"].health == STALE and verdicts["vrp"].stale_days > 3
@@ -134,8 +150,6 @@ def test_evaluate_health_replaces_snapshot_on_recovery(tmp_path):
dsn = f"sqlite:///{tmp_path}/rec.db"
repo = ForwardNavRepo(dsn)
repo.migrate()
xsp = XspOptionBarsRepo(dsn)
xsp.migrate()
# run 1: freeze stale → vrp STALE
at1 = dt.datetime(2026, 7, 14, 23, 30)
# a backtest ref so a recovered vrp resolves to HEALTHY (not NO_REF) — isolates the STALE→clear behavior.
@@ -143,12 +157,12 @@ def test_evaluate_health_replaces_snapshot_on_recovery(tmp_path):
sharpe=1.0, max_drawdown=-0.05, passed=True, dsr=0.9, is_sharpe=1.0,
oos_sharpe=0.9, pvalue=0.01), at1)
_seed_summary(repo, "vrp", "2026-07-13", at1)
xsp.upsert_bars([{"osi_symbol": "X", "date": "2026-06-15", "expiry": "2026-07-17", "strike": 470.0,
"right": "P", "close": 1.0}], at1)
_seed_freeze_bars(dsn, [{"osi_symbol": "X", "date": "2026-06-15", "expiry": "2026-07-17",
"strike": 470.0, "right": "P", "close": 1.0}], at1)
assert {v.strategy_id: v for v in evaluate_health(dsn, at=at1)}["vrp"].health == STALE
# run 2: freeze advances to yesterday → vrp HEALTHY, the old STALE row cleared
xsp.upsert_bars([{"osi_symbol": "Y", "date": "2026-07-13", "expiry": "2026-07-17", "strike": 470.0,
"right": "P", "close": 1.0}], at1)
_seed_freeze_bars(dsn, [{"osi_symbol": "Y", "date": "2026-07-13", "expiry": "2026-07-17",
"strike": 470.0, "right": "P", "close": 1.0}], at1)
_seed_summary(repo, "vrp", "2026-07-13", at1)
assert {v.strategy_id: v for v in evaluate_health(dsn, at=at1)}["vrp"].health == HEALTHY
assert {h.strategy_id: h for h in repo.all_health()}["vrp"].health == HEALTHY

109
tests/unit/test_xsp_option_bars_repo.py
View File

@@ -1,109 +0,0 @@
import datetime as dt
from fxhnt.adapters.persistence.xsp_option_bars import XspOptionBarsRepo
def _repo(tmp_path):
r = XspOptionBarsRepo(f"sqlite:///{tmp_path/'o.db'}")
r.migrate()
return r
def _bar(osi, date, expiry, strike, right, close):
return dict(osi_symbol=osi, date=date, expiry=expiry, strike=strike, right=right, close=close)
def test_upsert_and_bars_for(tmp_path):
r = _repo(tmp_path)
at = dt.datetime(2024, 7, 15)
r.upsert_bars([_bar("XSP240816P00470000", "2024-07-15", "2024-08-16", 470.0, "P", 1.20),
_bar("XSP240816P00470000", "2024-07-16", "2024-08-16", 470.0, "P", 1.35)], at)
assert r.bars_for(["XSP240816P00470000"], "2024-07-15", "2024-07-16") == {
"XSP240816P00470000": {"2024-07-15": 1.20, "2024-07-16": 1.35}}
def test_upsert_is_idempotent(tmp_path):
r = _repo(tmp_path)
at = dt.datetime(2024, 7, 15)
r.upsert_bars([_bar("A", "2024-07-15", "2024-08-16", 470.0, "P", 1.20)], at)
r.upsert_bars([_bar("A", "2024-07-15", "2024-08-16", 470.0, "P", 1.25)], at) # same PK -> overwrite
assert r.bars_for(["A"], "2024-07-15", "2024-07-15") == {"A": {"2024-07-15": 1.25}}
def test_chain_asof_filters_by_dte_window(tmp_path):
r = _repo(tmp_path)
at = dt.datetime(2024, 7, 15)
r.upsert_bars([
_bar("P30", "2024-07-15", "2024-08-14", 470.0, "P", 1.2), # 30 DTE -> in [20,45]
_bar("P10", "2024-07-15", "2024-07-25", 470.0, "P", 0.4), # 10 DTE -> out
_bar("P60", "2024-07-15", "2024-09-13", 470.0, "P", 2.0), # 60 DTE -> out
], at)
got = {c.osi_symbol for c in r.chain_asof("2024-07-15", 20, 45)}
assert got == {"P30"}
# ---- preload (replay N+1 -> one-query in-memory mirror) correctness ---------------------------------
_DAYS = ["2024-07-15", "2024-07-16", "2024-07-17", "2024-07-18"]
def _multi_day_rows():
rows = []
for di, day in enumerate(_DAYS):
for exp in ["2024-08-05", "2024-08-16", "2024-09-13"]:
for strike in [460.0, 470.0, 480.0]:
for right in ("P", "C"):
osi = f"XSP-{exp}-{int(strike)}-{right}"
rows.append(_bar(osi, day, exp, strike, right, round(1.0 + di * 0.1 + strike * 0.001, 4)))
return rows
def _load(tmp_path, name, rows):
r = XspOptionBarsRepo(f"sqlite:///{tmp_path / name}")
r.migrate()
r.upsert_bars(rows, dt.datetime(2024, 1, 1))
return r
def _chain(r, day, lo, hi, right):
return [(c.osi_symbol, c.expiry, c.strike, c.right) for c in r.chain_asof(day, lo, hi, right)]
def test_preload_full_mirror_matches_db_exactly(tmp_path):
"""The preloaded (in-memory) path must return results IDENTICAL to the per-day DB path — same
contracts, same strike order, same marks, same date filtering — for every query shape the replay uses."""
rows = _multi_day_rows()
db = _load(tmp_path, "db.db", rows) # never preloaded -> DB path
ca = _load(tmp_path, "ca.db", rows)
ca.preload() # full mirror (after=None)
assert ca.trading_days(None) == db.trading_days(None)
assert ca.trading_days("2024-07-16") == db.trading_days("2024-07-16")
for day in _DAYS:
for lo, hi in ((20, 45), (0, 60)):
for right in ("P", "C"):
assert _chain(ca, day, lo, hi, right) == _chain(db, day, lo, hi, right)
osis = [rows[0]["osi_symbol"], rows[20]["osi_symbol"], "MISSING-OSI"]
assert ca.bars_for(osis, "2024-07-16", "2024-07-16") == db.bars_for(osis, "2024-07-16", "2024-07-16")
assert ca.bars_for(osis, "2024-07-15", "2024-07-18") == db.bars_for(osis, "2024-07-15", "2024-07-18")
assert ca.bars_for([], "2024-07-15", "2024-07-15") == db.bars_for([], "2024-07-15", "2024-07-15")
def test_preload_scoped_covers_range_and_falls_back_outside(tmp_path):
"""A scoped preload (after=X) mirrors days > X and, critically, FALLS BACK to the DB for any date at
or before X — so it can never under-return an out-of-range query."""
rows = _multi_day_rows()
db = _load(tmp_path, "db2.db", rows)
sc = _load(tmp_path, "sc.db", rows)
sc.preload(after="2024-07-16") # covers 07-17, 07-18 in memory; 07-15/07-16 -> DB fallback
# covered days: served from cache, identical to DB
for day in ("2024-07-17", "2024-07-18"):
assert _chain(sc, day, 20, 45, "P") == _chain(db, day, 20, 45, "P")
osis = [rows[0]["osi_symbol"]]
assert sc.bars_for(osis, "2024-07-17", "2024-07-17") == db.bars_for(osis, "2024-07-17", "2024-07-17")
assert sc.trading_days("2024-07-16") == db.trading_days("2024-07-16")
# out-of-range (<= bound): must fall back to DB and still be correct, NOT return empty
assert _chain(sc, "2024-07-15", 20, 45, "P") == _chain(db, "2024-07-15", 20, 45, "P")
assert _chain(sc, "2024-07-15", 20, 45, "P") # non-empty proof the fallback fired
assert sc.bars_for(osis, "2024-07-15", "2024-07-15") == db.bars_for(osis, "2024-07-15", "2024-07-15")
assert sc.trading_days(None) == db.trading_days(None) # after=None under scoped preload -> DB fallback, all days