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

refactor(vrp): delete vrp_nav Dagster asset + OPRA-freeze/backfill helper chain

Browse Source
This commit is contained in:
Jeroen Grusewski
2026-07-20 23:24:06 +00:00
parent eb6ecf8272
commit 5b5fbb0e0b
1 changed files with 2 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files

233
src/fxhnt/adapters/orchestration/assets.py
View File

@@ -2,7 +2,7 @@
Key assets: `futures_bars` (Databento front-month arrays), `bybit_warehouse_refresh` (nightly Bybit
`bybit_features` update), the per-edge forward-nav tracks (xsfunding/unlock/positioning/bybit_4edge/
sixtyforty/multistrat/vrp), and `cockpit_forward` (recompute the forward state from the DB anchors and
sixtyforty/multistrat), and `cockpit_forward` (recompute the forward state from the DB anchors and
upsert into the operational cockpit DB). The DB is the single SSOT — no JSON state files.
NOTE: this module intentionally omits `from __future__ import annotations` because
@@ -489,235 +489,6 @@ def multistrat_levered_nav(context: AssetExecutionContext) -> dict: # type: ign
)
def _iso_date(s: str): # type: ignore[no-untyped-def]
import datetime as _dt
return _dt.date.fromisoformat(s)
def _forward_from_marks(chain, marks: dict, as_of: str) -> float: # type: ignore[no-untyped-def]
"""Put-call-parity forward from a PRE-FETCHED single-day marks dict `{osi_symbol: close}`. Pure (no OPRA)
— shared by the per-day fetch path (`_xsp_forward_estimate`) and the batched backfill (in-memory marks),
so both derive the forward identically. Nearest-to-30-DTE expiry, ATM common strike by min distance to
the mid; ties over a SORTED sequence (deterministic across PYTHONHASHSEED)."""
from fxhnt.application.vrp_pricing import parity_forward
near = [c for c in chain if 20 <= (c.expiry - _iso_date(as_of)).days <= 45]
if not near:
raise RuntimeError(f"no XSP chain for {as_of} — cannot estimate forward")
exp = min(sorted({c.expiry for c in near}), key=lambda e: abs((e - _iso_date(as_of)).days - 30))
puts = {c.strike: c for c in near if c.expiry == exp and c.right == "P"}
calls = {c.strike: c for c in near if c.expiry == exp and c.right == "C"}
common = sorted(set(puts) & set(calls))
if not common:
raise RuntimeError(f"no matched call/put strike for {as_of} forward")
mid = common[len(common) // 2]
for k in sorted(common, key=lambda s: abs(s - mid)):
p, c = marks.get(puts[k].osi_symbol), marks.get(calls[k].osi_symbol)
if p is not None and c is not None:
return parity_forward(call_atm=c, put_atm=p, k_atm=k)
raise RuntimeError(f"no ATM call/put pair with marks for {as_of}")
def _band_contracts(chain, as_of: str, forward: float): # type: ignore[no-untyped-def]
"""The daily freeze band: WIDE put band (DTE[1,45], moneyness[0.70,1.20] — wider than the ~30-DTE ENTRY
window so a spread's legs stay frozen through their whole held life + market moves, no zombie positions)
plus near-ATM CALLS (for the parity forward off frozen data). Pure filter."""
d = _iso_date(as_of)
puts = [c for c in chain if c.right == "P" and 1 <= (c.expiry - d).days <= 45
and 0.70 <= c.strike / forward <= 1.20]
calls = [c for c in chain if c.right == "C" and 20 <= (c.expiry - d).days <= 45
and 0.96 <= c.strike / forward <= 1.04]
return puts + calls
def _band_rows_from_marks(chain, marks: dict, as_of: str, forward: float) -> list: # type: ignore[type-arg,no-untyped-def]
"""Band rows for `as_of` from a PRE-FETCHED single-day marks dict `{osi_symbol: close}`. Pure — shared by
the per-day path and the batched backfill."""
return [dict(osi_symbol=c.osi_symbol, date=as_of, expiry=c.expiry.isoformat(),
strike=c.strike, right=c.right, close=marks[c.osi_symbol])
for c in _band_contracts(chain, as_of, forward) if c.osi_symbol in marks]
def _day_marks(dbn, osis, as_of: str) -> dict: # type: ignore[type-arg,no-untyped-def]
"""Fetch `osis` for a single day and flatten `{osi:{date:close}}` -> `{osi: close}` (dropping missing)."""
return {osi: m[as_of] for osi, m in dbn.fetch_option_bars(osis, as_of, as_of).items() if as_of in m}
def _xsp_forward_estimate(dbn, as_of: str, chain=None) -> float: # type: ignore[no-untyped-def]
"""Per-day parity forward: fetch the near-ATM expiry's marks, then `_forward_from_marks`. `chain` may be
passed in to avoid a redundant OPRA definition fetch (see `_freeze_day`)."""
full = chain if chain is not None else dbn.fetch_option_chain("XSP.OPT", as_of)
near = [c for c in full if 20 <= (c.expiry - _iso_date(as_of)).days <= 45]
if not near:
raise RuntimeError(f"no XSP chain for {as_of} — cannot estimate forward")
exp = min(sorted({c.expiry for c in near}), key=lambda e: abs((e - _iso_date(as_of)).days - 30))
atm_osis = [c.osi_symbol for c in near if c.expiry == exp]
return _forward_from_marks(full, _day_marks(dbn, atm_osis, as_of), as_of)
def _freeze_xsp_slice(dbn, bars, *, as_of: str, forward: float, at, chain=None) -> None: # type: ignore[no-untyped-def]
"""Per-day freeze: fetch the band's marks, then upsert `_band_rows_from_marks`. `chain` may be passed in
to avoid a redundant OPRA definition fetch (see `_freeze_day`)."""
full = chain if chain is not None else dbn.fetch_option_chain("XSP.OPT", as_of)
band = _band_contracts(full, as_of, forward)
marks = _day_marks(dbn, [c.osi_symbol for c in band], as_of)
bars.upsert_bars(_band_rows_from_marks(full, marks, as_of, forward), at)
def _freeze_day(dbn, bars, *, as_of: str, at) -> float: # type: ignore[no-untyped-def]
"""Fetch the XSP option chain ONCE for `as_of`, derive the parity forward from it, and freeze the band
from the SAME chain. Avoids the double `fetch_option_chain` (definition) call — `_xsp_forward_estimate`
and `_freeze_xsp_slice` each used to fetch it independently, doubling OPRA definition cost on every
nightly `vrp_nav` run and the 2013-> backfill. Returns the forward."""
chain = dbn.fetch_option_chain("XSP.OPT", as_of)
fwd = _xsp_forward_estimate(dbn, as_of, chain=chain)
_freeze_xsp_slice(dbn, bars, as_of=as_of, forward=fwd, at=at, chain=chain)
return fwd
def _prev_trading_day(d): # type: ignore[no-untyped-def]
"""One trading day BEFORE `d`, skipping Sat/Sun (isoweekday 6/7). A market holiday is not a fixed calendar
rule, so it is absorbed by one more step-back in `_freeze_latest_session`'s bound rather than skipped here."""
import datetime as _dt
d -= _dt.timedelta(days=1)
while d.isoweekday() >= 6: # Sat=6, Sun=7 — never a trading session
d -= _dt.timedelta(days=1)
return d
# Message fragments raised by `_xsp_forward_estimate`/`_forward_from_marks` when a PUBLISHED session is too
# degenerate to form the parity forward (no near-DTE chain / no matched call-put strike / no ATM pair with
# marks). Matched narrowly so `_freeze_latest_session` steps back on a broken session but never swallows an
# unrelated RuntimeError (e.g. a DB write failure inside the freeze).
_DEGENERATE_FORWARD_MARKERS = ("cannot estimate forward", "matched call/put strike", "ATM call/put pair")
def _is_degenerate_forward_error(e: Exception) -> bool:
msg = str(e)
return any(marker in msg for marker in _DEGENERATE_FORWARD_MARKERS)
def _freeze_latest_session(dbn, bars, *, at, max_steps: int = 5) -> str: # type: ignore[no-untyped-def]
"""Freeze the LATEST COMPLETE, USABLE OPRA session (robust, timing-independent). The candidate starts at
OPRA's available-END date (`dbn.last_available_option_date()`); today's session is still settling until
after the close, so freezing it fails with the normalized `DatabentoRangeUnavailable` (the
`data_schema_not_fully_available` 422 OR the C1a empty-window clamp). A PUBLISHED-but-DEGENERATE session
instead surfaces as a forward-estimation `RuntimeError` ("no XSP chain … cannot estimate forward" / "no
matched call/put strike …" / "no ATM call/put pair …"). Either signal — unavailable OR degenerate — steps
the candidate back ONE trading day (skip Sat/Sun) and retries, BOUNDED to `max_steps` (a market holiday or
one broken session just costs one more step). The FIRST candidate that freezes is the target; the ISO date
frozen is RETURNED (vrp_nav logs it). Unrelated `RuntimeError`s (not forward-estimation) propagate.
If no usable session is found within the bound, raise `DatabentoRangeUnavailable` — the C1b resilience
wrapper in `vrp_nav` catches it → recompute off the frozen table, and the health axis keeps vrp STALE
(correct: we genuinely could not advance the freeze). This is LOUD by design; NEVER a silent no-op, and a
bad/bar-less date is NEVER returned — only the last COMPLETE, non-degenerate session's marks are frozen."""
from fxhnt.adapters.data.databento import DatabentoRangeUnavailable
start = dbn.last_available_option_date()
candidate = start
last_err: Exception | None = None
for _ in range(max_steps + 1): # the start candidate + up to `max_steps` bounded step-backs
iso = candidate.isoformat()
try:
_freeze_day(dbn, bars, as_of=iso, at=at)
return iso
except DatabentoRangeUnavailable as e: # still-settling / not-yet-available session → step back
last_err = e
candidate = _prev_trading_day(candidate)
except RuntimeError as e: # published-but-degenerate session (no derivable forward)
if not _is_degenerate_forward_error(e):
raise # an unrelated RuntimeError — do NOT silently swallow it
last_err = e
candidate = _prev_trading_day(candidate)
raise DatabentoRangeUnavailable(
f"no complete, non-degenerate OPRA session within {max_steps} trading days back "
f"from {start.isoformat()}") from last_err
def _rough_forward(chain, as_of: str): # type: ignore[no-untyped-def]
"""Coarse forward = median strike of the ~30-DTE expiry (NO marks needed) — used ONLY to bound the bulk
monthly ohlcv fetch to a wide strike band. The precise per-day forward is always recomputed from real
marks via `_forward_from_marks`. XSP lists strikes densely+symmetrically around ATM, so the median strike
is a good ATM proxy for bounding."""
d = _iso_date(as_of)
near = [c for c in chain if 20 <= (c.expiry - d).days <= 45]
if not near:
return None
exp = min(sorted({c.expiry for c in near}), key=lambda e: abs((e - d).days - 30))
strikes = sorted({c.strike for c in near if c.expiry == exp})
return strikes[len(strikes) // 2] if strikes else None
def _backfill_month(dbn, bars_repo, *, month_start: str, month_end: str, at) -> tuple: # type: ignore[type-arg,no-untyped-def]
"""BATCHED one-month freeze: ONE range definition fetch + ONE range ohlcv fetch (bulk), then process each
trading day LOCALLY (parity forward + band + upsert). ~2 OPRA calls/month vs ~6-8/day — the per-day
pattern was ~26k API round-trips over 2013->, ~45h; batched is ~2 calls/month * ~160 months, ~1h.
Returns (n_ok, n_fail) days. A day with no arb-free ATM pair (sparse early-XSP data) is counted failed,
never aborts the month."""
import datetime as _dt
chain = dbn.fetch_option_chain_range("XSP.OPT", month_start, month_end)
if not chain:
return (0, 0)
rough = _rough_forward(chain, month_start) or _rough_forward(chain, month_end)
if rough is None:
return (0, 0)
ms, me = _iso_date(month_start), _iso_date(month_end)
# bound the bulk fetch: any expiry a day this month could reference at DTE<=45, and a strike band WIDE
# enough (+-50% of the month's median) to cover intra-month moves + the precise [0.70,1.20] daily band
# even in a crash month (COVID Mar-2020 XSP fell ~34% intramonth).
band = [c for c in chain
if (ms + _dt.timedelta(days=1)) <= c.expiry <= (me + _dt.timedelta(days=45))
and 0.50 <= c.strike / rough <= 1.50]
if not band:
return (0, 0)
bars = dbn.fetch_option_bars([c.osi_symbol for c in band], month_start, month_end) # bulk range, chunked
days = sorted({dd for m in bars.values() for dd in m})
n_ok = n_fail = 0
for iso in days:
day_marks = {osi: m[iso] for osi, m in bars.items() if iso in m}
try:
fwd = _forward_from_marks(band, day_marks, iso)
bars_repo.upsert_bars(_band_rows_from_marks(band, day_marks, iso, fwd), at)
n_ok += 1
except Exception: # a sparse/degenerate day (no ATM pair with marks) — skip, don't abort the month
n_fail += 1
return (n_ok, n_fail)
@asset
def vrp_nav(context: AssetExecutionContext) -> dict: # type: ignore[type-arg]
"""Paper-forward defined-risk XSP put-credit-spread VRP on REAL OPRA data. Freezes today's DTE/moneyness
put slice (plus near-ATM calls, for the parity forward) PIT into xsp_option_bars, then recomputes off the
FROZEN table (spec §A) so the record is stable against any OPRA revision. persist_backtest_ref=True
publishes the basis-matched real ref for the gate."""
import datetime as _dt
from fxhnt.adapters.data.databento import DatabentoDataProvider
from fxhnt.adapters.persistence.xsp_option_bars import XspOptionBarsRepo
from fxhnt.application.vrp_book import VrpStrategy
from fxhnt.config import get_settings
s = get_settings()
bars = XspOptionBarsRepo(s.operational_dsn)
bars.migrate()
dbn = DatabentoDataProvider(max_cost_usd=s.databento.max_cost_usd)
at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)
try:
# Freeze the LATEST COMPLETE OPRA session via a bounded step-back — robust and timing-independent, so
# the freeze advances deterministically regardless of when the job runs (never freezes today's still-
# settling partial session's marks). Returns the ISO date actually frozen.
session = _freeze_latest_session(dbn, bars, at=at)
context.log.info(f"vrp_nav: froze OPRA session {session}")
except Exception as e: # noqa: BLE001 — freeze is best-effort; the tracker recomputes off the FROZEN table
# A bad-data day, no complete session within the step-back bound (DatabentoRangeUnavailable), or a
# cost-cap trip must NOT kill vrp_nav: it would zero out the forward anchor AND skip the downstream
# gate. Log + continue — the health axis keeps vrp STALE (loud), which is the correct signal.
context.log.warning(f"vrp_nav: freeze step failed (recomputing off the frozen table): {e}")
return _run_paper_tracker(
context, "vrp_nav", lambda: VrpStrategy(bars), persist_backtest_ref=True,
)
# RETIRED 2026-06-20 (cockpit cleanup): the live tracks funding_nav (absolute-hurdle, superseded by
# cross-sectional xsfunding), gd_nav / multistrat_nav (futures/TradFi premia — falsified this session,
# scale-gated, no deployable edge), and poc_nav (old crypto residual-mom+VRP PoC, superseded by the two
@@ -1128,7 +899,7 @@ def bybit_paper_book(context: AssetExecutionContext) -> dict: # type: ignore[ty
@asset(deps=[sixtyforty_nav, xsfunding_nav, unlock_nav, multistrat_nav, bybit_4edge_nav,
bybit_4edge_levered_nav, positioning_nav]) # vrp_nav ARCHIVED 2026-07-15 — de-wired (fn kept below)
bybit_4edge_levered_nav, positioning_nav])
def cockpit_forward(context: AssetExecutionContext) -> None:
"""Evaluate + persist the forward gate from the DB record the engine already wrote (all_summaries +
nav_history — see `forward_ingest.ingest_forward_state`), then auto-promote any `promote_on_pass`