diff --git a/src/fxhnt/adapters/orchestration/assets.py b/src/fxhnt/adapters/orchestration/assets.py index 730ca16..b9de5db 100644 --- a/src/fxhnt/adapters/orchestration/assets.py +++ b/src/fxhnt/adapters/orchestration/assets.py @@ -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`