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

refactor(vrp): remove execute-vrp/ingest-dvol/vrp-eval/vrp-defined-risk-eval/backfill-xsp-opra CLI commands

Browse Source
This commit is contained in:
Jeroen Grusewski
2026-07-20 23:20:00 +00:00
parent b71cd6b3b0
commit eb6ecf8272
1 changed files with 0 additions and 460 deletions
Download Patch File Download Diff File Expand all files Collapse all files

460
src/fxhnt/cli.py
View File

@@ -436,69 +436,6 @@ def execute_bybit(
f"| exec_return {out.get('exec_return')}")
@app.command("execute-vrp")
def execute_vrp(
do_execute: bool = typer.Option(False, "--execute", help="place combo orders (default: dry-run plan only)"),
paper_envelope: float = typer.Option(
0.0, "--paper-envelope",
help="PAPER-only validation envelope in $ (there is no B2a allocation for VRP yet, so this is the "
"ONLY way this track sizes a new ladder rung; 0 = no new rungs, existing rungs still mark/close)."),
) -> None:
"""Reconcile + record the EXECUTED XSP put-credit-spread VRP twin on IBKR (`mleg` combo), maintaining a
ladder of open rungs (mirrors `VrpStrategy.advance`'s discipline: close a rung at 50% profit or DTE<=1,
cap entries at `ladder`, size each new rung at envelope/ladder — never the full envelope). SUSPENDED: the
IBKR account currently lacks options trading permission and CI has no live OPRA, so this path is coded
but not runnable here — dry-run by default even once armed. Same paper-envelope refusal guard as
`execute-multistrat`/`execute-bybit`. Records the observed `vrp_exec` return."""
import datetime as _dt
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.application.vrp_exec_record import plan_and_record_vrp
settings = get_settings()
d = settings.databento
dbn = DatabentoDataProvider(max_cost_usd=d.max_cost_usd, equity_dataset=d.equity_dataset,
future_dataset=d.future_dataset, default_start=d.default_start)
repo = ForwardNavRepo(settings.operational_dsn)
repo.migrate()
today = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%d")
at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)
broker = IbkrBroker(settings.ibkr.host, settings.ibkr.port, settings.ibkr.client_id, settings.ibkr.timeout)
paper_env = _resolve_paper_envelope(paper_envelope, settings.paper_validation.envelope)
try:
with broker as brk:
acct = brk.account_state()
venue = _exec_venue_string(brk.name, acct.is_paper)
out = plan_and_record_vrp(
repo, dbn=dbn, broker=brk, nlv=acct.nlv, today=today, at=at,
paper_envelope=paper_env, account_is_paper=acct.is_paper, venue=venue, execute=do_execute)
except ValueError as e: # a deliberate safety refusal (e.g. paper-envelope on a non-paper account) —
typer.echo(f"paper-envelope refused: {e}") # distinct from a connectivity/data error
raise typer.Exit(1) from e
except Exception as e: # connection / OPRA / runtime
typer.echo(f"broker error: {type(e).__name__}: {str(e)[:100]}")
typer.echo(f" reachable? IBKR ({settings.ibkr.host}:{settings.ibkr.port}) "
f"— creds/gateway + options permission set? OPRA reachable?")
raise typer.Exit(1) from e
if out.get("noop"):
typer.echo(f"-> no-op: {out.get('reason')}")
return
typer.echo(f"to_close={out.get('to_close', 0)} closed={out.get('closed', 0)}")
spread = out.get("spread")
if spread:
typer.echo(f"spread: SELL {spread['short_osi']} / BUY {spread['long_osi']} qty={out.get('qty')} "
f"net ${out.get('net'):+.2f}")
for n in out.get("notes", []):
typer.echo(f" - {n}")
for err in out.get("errors", []):
typer.echo(f" order error: {err}")
typer.echo(f"-> exec_return {out.get('exec_return')} | placed {out.get('placed', 0)} "
f"| open_spreads {out.get('open_spreads')}")
@app.command("migrate-cockpit")
def migrate_cockpit() -> None:
"""Create the cockpit tables (+ TimescaleDB hypertable on Postgres) and seed the strategy registry."""
@@ -575,54 +512,6 @@ def unlock_calendar_import(
f"unlock_events @ {dsn.split('@')[-1]} (table now has {len(repo.read_events())})")
@app.command("backfill-xsp-opra")
def backfill_xsp_opra(
start: str = typer.Option(..., "--start", help="ISO date to begin the backfill from (e.g. 2013-04-01)"),
max_cost: float = typer.Option(
25.0, "--max-cost",
help="per-MONTH OPRA cost cap in $ (BATCHED: one range definition + one range ohlcv fetch per month, "
"so the guard is per-month, not per-day)."),
) -> None:
"""One-time historical freeze of the XSP OPRA slice from --start through today into `xsp_option_bars`,
BATCHED per month: one range `definition` + one range `ohlcv-1d` fetch per month, then every trading day
processed LOCALLY (parity forward + band + upsert) — ~2 OPRA calls/month vs the old ~6-8/day (~1h vs ~45h
over 2013->). Resumable/idempotent (`upsert_bars` merges on (osi_symbol, date)) — safe to re-run or resume
from a later --start. A bad month is logged + skipped, never aborts. Manual, one-time; NOT the nightly
asset graph. The per-fetch databento cost guard stays active (raise --max-cost, never disable it)."""
import datetime as _dt
from fxhnt.adapters.orchestration.assets import _backfill_month
from fxhnt.adapters.persistence.xsp_option_bars import XspOptionBarsRepo
settings = get_settings()
d = settings.databento
dbn = DatabentoDataProvider(max_cost_usd=max_cost, equity_dataset=d.equity_dataset,
future_dataset=d.future_dataset, default_start=d.default_start)
bars = XspOptionBarsRepo(settings.operational_dsn)
bars.migrate()
at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)
start_d = _dt.date.fromisoformat(start)
today = _dt.datetime.now(_dt.UTC).date()
cur = start_d.replace(day=1)
tot_ok = tot_fail = 0
while cur <= today:
nxt = (cur + _dt.timedelta(days=32)).replace(day=1) # first of the next month
m_start = max(cur, start_d)
m_end = min(nxt - _dt.timedelta(days=1), today)
try:
ok, fail = _backfill_month(dbn, bars, month_start=m_start.isoformat(),
month_end=m_end.isoformat(), at=at)
tot_ok += ok
tot_fail += fail
typer.echo(f"backfill-xsp-opra: {cur:%Y-%m} done ({ok} ok, {fail} fail) — {tot_ok} ok total")
except Exception as e: # a bad month (cost/network) — log + continue, never abort the whole backfill
typer.echo(f"backfill-xsp-opra: {cur:%Y-%m} FAILED ({type(e).__name__}: {str(e)[:120]}) — skipped")
cur = nxt
typer.echo(f"-> backfill-xsp-opra complete: {tot_ok} days frozen, {tot_fail} failed, "
f"{start}..{today.isoformat()}")
@app.command("backfill-etf-volume")
def backfill_etf_volume() -> None:
"""One-time/resumable historical freeze of each `FUND_INSTRUMENTS` ETF's raw Yahoo daily share volume
@@ -1214,58 +1103,6 @@ def bybit_ingest_klines(
f"oldest date {oldest_date}")
@app.command("ingest-dvol")
def ingest_dvol_cmd(
currencies: str = typer.Option(
"BTC,ETH", "--currencies",
help="comma-separated Deribit DVOL currencies to ingest (default BTC,ETH — the assets with a DVOL "
"index). Each is written as feature 'dvol' under '<CCY>USDT' in bybit_features."),
from_month: str = typer.Option(
"", "--from", help="floor for the backward pagination, YYYY-MM (default: the 2021-03 backstop, just "
"below DVOL inception 2021-03-24)."),
) -> None:
"""Ingest the Deribit DVOL annualised IMPLIED-vol index (the IV leg of the VRP edge) into the
Bybit-namespaced TimescaleDB table `bybit_features` as feature 'dvol', keyed by the perp symbol
(BTC -> BTCUSDT, ETH -> ETHUSDT). resolution=86400 -> one value/day. ONLY 'dvol' is written
(close/funding are the other ingests' job); idempotent/resumable. The VRP eval reads this `dvol` panel
alongside the kline `close` panel. No live paper_* tables are touched."""
import datetime as _dt
from fxhnt.adapters.data.deribit_dvol import DeribitDVOL
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
from fxhnt.application.dvol_ingest import ingest_dvol
settings = get_settings()
deribit = DeribitDVOL()
start_ms: int | None = None
if from_month:
try:
y, m = (int(x) for x in from_month.split("-"))
start_ms = int(_dt.datetime(y, m, 1, tzinfo=_dt.UTC).timestamp() * 1000)
except (ValueError, TypeError) as e:
typer.echo(f"error: --from must be YYYY-MM (got {from_month!r})", err=True)
raise typer.Exit(1) from e
ccys = tuple(c.strip().upper() for c in currencies.split(",") if c.strip())
if not ccys:
typer.echo("error: empty --currencies", err=True)
raise typer.Exit(1)
typer.echo(f"ingest-dvol: {ccys}, floor={from_month or '2021-03 (default)'} "
f"-> bybit_features feature='dvol' @ {settings.operational_dsn.split('@')[-1]}")
store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
try:
summary = ingest_dvol(store, deribit, currencies=ccys, start_ms=start_ms)
finally:
store.close()
oldest = summary["oldest_day"]
oldest_date = ((_dt.date(1970, 1, 1) + _dt.timedelta(days=oldest)).isoformat()
if oldest is not None else "n/a")
typer.echo(f"\ningest-dvol done: {summary['symbols']} currencies, "
f"{summary['day_rows']} day-rows, oldest date {oldest_date}")
@app.command("ingest-deribit-funding")
def ingest_deribit_funding_cmd(
instruments: str = typer.Option(
@@ -2189,303 +2026,6 @@ def bybit_positioning_eval(
typer.echo("\nNOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched. Memory-bounded: only the "
"(liquid, with --liquid) coins are READ — the full panels are never loaded.")
@app.command("vrp-eval")
def vrp_eval(
direction: str = typer.Option(
"short_vol", "--direction",
help="short_vol (DEFAULT: SELL variance — harvest the IV-over-RV premium; fat LEFT tail on vol "
"spikes) or long_vol (BUY variance — the negation). Both are RE-RUN so the cost subtracts "
"correctly in each."),
cost_bps: float = typer.Option(
5.5, "--cost-bps",
help="Per-day cost (bps, scaled by the vol-target leverage). NOTE: selling OPTIONS is NOT 5.5bp — "
"--verify shows a 50/100bp option-cost sensitivity."),
target_ann_vol: float = typer.Option(
0.15, "--target-vol", help="Vol-target the VRP book to this annual vol (default 0.15 = 15%/yr)."),
verify: bool = typer.Option(
False, "--verify",
help="ADVERSARIAL verification: BOTH directions' net-of-cost metrics across a cost grid INCLUDING "
"realistic OPTION costs (5.5/11/22/50/100 bps), the single WORST DAY (the vol-spike tail), "
"maxDD, per-year Sharpe+maxDD, AND the correlation with EACH existing edge "
"(tstrend/unlock/xsfunding/positioning)."),
walk_forward: bool = typer.Option(
False, "--walk-forward",
help="OOS / WALK-FORWARD validation (the TAIL-AWARE DEPLOY GATE): a TRAIN/TEST split that picks the "
"direction AND sizes k on TRAIN then scores OOS, rolling non-overlapping windows, a look-ahead "
"audit, and a PASS/FAIL verdict that FAILS a catastrophic-tail book even when Sharpe>0."),
window_days: int = typer.Option(
180, "--window-days", help="Width (days) of each rolling non-overlapping OOS window in --walk-forward."),
cost_grid: str = typer.Option(
"5.5,11,22,50,100", "--cost-grid",
help="comma-separated cost levels (bps) for --verify (default spans cheap perp-style + realistic "
"option costs)."),
) -> None:
"""READ-ONLY, TAIL-HONEST: evaluate the VRP (volatility-risk-premium) edge — a vol premium, a genuinely
DIFFERENT edge type from the directional/flow edges already in the book.
SIGNAL = Deribit DVOL (implied vol) vs the BTC/ETH close-to-close REALIZED vol. short-vol (default)
harvests the IV-over-RV spread via a ROLLING VARIANCE SWAP: strike K=(DVOL/100)² (implied annual variance,
causal) vs RV²=(365/swap_days)·Σ ret² over the swap's life; a daily ladder of overlapping 30-day swaps
gives a SMOOTH daily series, BTC+ETH equal-weight, vol-targeted to --target-vol. POSITIVE on calm windows,
big-NEGATIVE on vol-spike windows (the fat left tail). EXECUTION would be short Bybit option straddles
(a cross-venue caveat vs the Deribit signal).
Prints CAGR/Sharpe/maxDD + the single WORST DAY (the vol spike) + the vol-target leverage. --verify adds
the option-cost sensitivity, per-year tails, and the correlation with each existing edge. --walk-forward
is the tail-aware deploy gate (a clean "no / too-tail-heavy" is a VALID outcome). Writes NOTHING."""
if direction not in ("short_vol", "long_vol"):
raise typer.BadParameter("must be 'short_vol' or 'long_vol'", param_hint="--direction")
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
from fxhnt.application.vrp_eval import (
verify_vrp_edge,
vrp_metrics_from_store,
walk_forward_vrp,
)
settings = get_settings()
store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
if walk_forward:
try:
rep = walk_forward_vrp(store, cost_bps=cost_bps, window_days=window_days,
target_ann_vol=target_ann_vol)
finally:
store.close()
_print_vrp_walk_forward(rep, cost_bps=cost_bps)
return
if verify:
grid = tuple(float(x) for x in cost_grid.split(",") if x.strip())
try:
rep = verify_vrp_edge(store, cost_bps=cost_bps, cost_grid=grid, target_ann_vol=target_ann_vol)
finally:
store.close()
_print_vrp_verify(rep, cost_bps=cost_bps)
return
try:
m = vrp_metrics_from_store(store, cost_bps=cost_bps, direction=direction,
target_ann_vol=target_ann_vol)
finally:
store.close()
typer.echo("\n=== VRP (volatility risk premium) — TAIL-HONEST, READ-ONLY ===")
typer.echo(f"direction: {direction.upper()} ({'SELL variance — harvest' if direction == 'short_vol' else 'BUY variance — negation'})")
typer.echo("signal: Deribit DVOL (IV) vs BTC/ETH realized vol | book: BTC+ETH equal-weight, vol-targeted")
typer.echo(f"construction: rolling 30d variance swap K=(DVOL/100)^2 vs RV^2=(365/n)*sum(ret^2) (daily "
f"ladder) | target vol={target_ann_vol*100:.0f}%/yr")
if m.get("available"):
typer.echo(f" CAGR={m['cagr']*100:7.1f}% Sharpe={m['sharpe']:6.2f} maxDD={m['max_dd']*100:7.1f}% "
f" total={m['total_return']*100:7.1f}% days={m['days']} span={m['first_day']}..{m['last_day']}")
typer.echo(f" WORST DAY (vol-spike tail loss): {m['worst_day']*100:+.2f}% | avg leverage (k): {m['avg_leverage']:.2f}")
else:
typer.echo(f" N/A — {m.get('reason', '')}")
typer.echo(" (run `fxhnt ingest-dvol` + `fxhnt bybit-ingest-klines` first)")
typer.echo("\nCROSS-VENUE CAVEAT: the SIGNAL is Deribit's DVOL; live EXECUTION would be short Bybit option "
"straddles (delta-hedged).")
typer.echo("NOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched.")
@app.command("vrp-defined-risk-eval")
def vrp_defined_risk_eval(
direction: str = typer.Option(
"short_vol", "--direction",
help="short_vol (DEFAULT: short ATM straddle + long OTM wings — harvest the IV-over-RV premium with a "
"BOUNDED tail) or long_vol (the negation)."),
cost_bps: float = typer.Option(
5.5, "--cost-bps",
help="Per-day cost (bps, scaled by the vol-target leverage). Selling OPTIONS is NOT 5.5bp — use "
"--cost-bps 50 (or --verify) for the realistic option-cost deploy gate."),
target_ann_vol: float = typer.Option(
0.15, "--target-vol", help="Vol-target the defined-risk book to this annual vol (default 0.15)."),
wing_width: float = typer.Option(
0.15, "--wing-width",
help="OTM wing distance as a FRACTION of spot (default 0.15 = ±15%). The max loss is BOUNDED at "
"wing_width − credit. Wider wing => cheaper protection => larger credit but larger max loss "
"(naked limit as wing_width -> 1)."),
wing_in_atm_std: float = typer.Option(
0.0, "--wing-in-atm-std",
help="If > 0, place the wings at this many ATM std-moves (wing = N·IV·sqrt(T)) instead of the fixed "
"--wing-width. 0 (default) uses --wing-width."),
put_skew_vol: float = typer.Option(
5.0, "--put-skew-vol",
help="Crypto put-skew bump (vol POINTS) added to the put-leg IV (default +5). 0 = flat-ATM = under-"
"prices the put wing => mildly OPTIMISTIC (documented caveat)."),
verify: bool = typer.Option(
False, "--verify",
help="ADVERSARIAL verification: both directions net-of-cost across the cost grid (incl 50/100bp option "
"costs), worst-day, maxDD, per-year, and corr with each existing edge."),
walk_forward: bool = typer.Option(
False, "--walk-forward",
help="OOS / WALK-FORWARD validation (the TAIL-AWARE DEPLOY GATE): TRAIN picks direction+k, scored OOS, "
"rolling windows, look-ahead audit, PASS/FAIL verdict. The wings are meant to PASS the tail "
"check the naked VRP (−74% maxDD) failed."),
window_days: int = typer.Option(
180, "--window-days", help="Width (days) of each rolling non-overlapping OOS window in --walk-forward."),
cost_grid: str = typer.Option(
"5.5,11,22,50,100", "--cost-grid",
help="comma-separated cost levels (bps) for --verify (cheap perp-style + realistic option costs)."),
) -> None:
"""READ-ONLY, TAIL-HONEST: evaluate the DEFINED-RISK VRP — the deployable form of the vol-risk premium.
The naked short-variance VRP has a real gross edge but FAILS the deploy gate on an UNBOUNDED left tail
(modelled maxDD ~−74% on vol spikes + realistic option costs). This caps it: SELL the ATM straddle but BUY
OTM wings (an iron condor) so the max loss is BOUNDED at wing_width − credit. The legs are priced from DVOL
via BLACK-SCHOLES (we have no historical option chains — a documented BS + put-skew approximation, so the
modelled credit is an approximation, not a real chain). Re-runs the SAME tail-aware harness as the naked
evaluator (vol-target, cost grid, per-year, worst-day, maxDD, corr-vs-4-edges, look-ahead, PASS/FAIL) so
the two are directly comparable — the headline is the maxDD the wings buy back. Writes NOTHING."""
if direction not in ("short_vol", "long_vol"):
raise typer.BadParameter("must be 'short_vol' or 'long_vol'", param_hint="--direction")
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
from fxhnt.application.vrp_defined_risk_eval import (
defined_risk_metrics_from_store,
verify_defined_risk_edge,
walk_forward_defined_risk,
)
settings = get_settings()
store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
wstd = wing_in_atm_std if wing_in_atm_std > 0.0 else None
if walk_forward:
try:
rep = walk_forward_defined_risk(store, cost_bps=cost_bps, window_days=window_days,
target_ann_vol=target_ann_vol, wing_width=wing_width,
put_skew_vol=put_skew_vol, wing_in_atm_std=wstd)
finally:
store.close()
_print_vrp_walk_forward(rep, cost_bps=cost_bps)
typer.echo(f"\nDEFINED-RISK: wing_width=±{wing_width*100:.0f}% put_skew_vol=+{put_skew_vol:.1f}vp "
"(short ATM straddle + long OTM wings; max loss BOUNDED at wing_width − credit)")
typer.echo("MODELLING CAVEAT: legs are BLACK-SCHOLES-priced from the ATM DVOL (no real option chain); "
"put_skew_vol=0 under-prices the put wing => optimistic.")
return
if verify:
grid = tuple(float(x) for x in cost_grid.split(",") if x.strip())
try:
rep = verify_defined_risk_edge(store, cost_bps=cost_bps, cost_grid=grid,
target_ann_vol=target_ann_vol, wing_width=wing_width,
put_skew_vol=put_skew_vol, wing_in_atm_std=wstd)
finally:
store.close()
_print_vrp_verify(rep, cost_bps=cost_bps)
typer.echo(f"\nDEFINED-RISK: wing_width=±{wing_width*100:.0f}% put_skew_vol=+{put_skew_vol:.1f}vp "
"(BS-priced from DVOL; max loss BOUNDED at wing_width − credit)")
return
try:
m = defined_risk_metrics_from_store(store, cost_bps=cost_bps, direction=direction,
target_ann_vol=target_ann_vol, wing_width=wing_width,
put_skew_vol=put_skew_vol, wing_in_atm_std=wstd)
finally:
store.close()
typer.echo("\n=== DEFINED-RISK VRP (short straddle + protective wings) — TAIL-HONEST, READ-ONLY ===")
typer.echo(f"direction: {direction.upper()} | wing_width: ±{wing_width*100:.0f}% | put_skew_vol: "
f"+{put_skew_vol:.1f}vp | target vol={target_ann_vol*100:.0f}%/yr")
typer.echo("construction: SELL ATM straddle, BUY OTM wings (BS-priced from DVOL); max loss BOUNDED at "
"wing_width − credit")
if m.get("available"):
typer.echo(f" CAGR={m['cagr']*100:7.1f}% Sharpe={m['sharpe']:6.2f} maxDD={m['max_dd']*100:7.1f}% "
f" total={m['total_return']*100:7.1f}% days={m['days']} span={m['first_day']}..{m['last_day']}")
typer.echo(f" WORST DAY (capped vol-spike loss): {m['worst_day']*100:+.2f}% | avg leverage (k): {m['avg_leverage']:.2f}")
else:
typer.echo(f" N/A — {m.get('reason', '')}")
typer.echo(" (run `fxhnt ingest-dvol` + `fxhnt bybit-ingest-klines` first)")
typer.echo("\nMODELLING CAVEAT: legs are BLACK-SCHOLES-priced from the ATM DVOL (no real option chain); a "
"put-skew bump models richer crypto puts (put_skew_vol=0 under-prices the put wing => optimistic).")
typer.echo("CROSS-VENUE CAVEAT: SIGNAL = Deribit DVOL; live EXECUTION = short Bybit option condors.")
typer.echo("NOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched.")
def _print_vrp_verify(rep: dict, *, cost_bps: float) -> None:
"""Pretty-print the adversarial VRP verification report (both directions, tail-honest)."""
typer.echo("\n=== VRP (volatility risk premium) — ADVERSARIAL VERIFICATION (TAIL-HONEST, READ-ONLY) ===")
typer.echo("signal: Deribit DVOL (IV) vs BTC/ETH realized vol | book: BTC+ETH equal-weight, vol-targeted")
if not rep.get("available"):
typer.echo(f" N/A — {rep.get('reason', '')}")
typer.echo(" (run `fxhnt ingest-dvol` + `fxhnt bybit-ingest-klines` first)")
return
typer.echo("\n-- NET-OF-COST, BOTH DIRECTIONS (incl REALISTIC OPTION costs 50/100bp) --")
for d in ("short_vol", "long_vol"):
typer.echo(f" [{d.upper()}] avg leverage (k)={rep['avg_leverage'].get(d, 0.0):.2f}")
for bps in rep["cost_grid"]:
m = rep["net_cost"][d][bps]
typer.echo(f" cost={bps:6.1f}bps Sharpe={m['sharpe']:6.2f} CAGR={m['cagr']*100:7.1f}% "
f"maxDD={m['max_dd']*100:7.1f}% total={m['total_return']*100:8.1f}% days={m['days']}")
typer.echo("\n-- THE TAIL (single worst day = the vol spike; maxDD @ headline cost) --")
for d in ("short_vol", "long_vol"):
typer.echo(f" [{d.upper()}] WORST DAY={rep['worst_day'][d]*100:+.2f}% maxDD={rep['max_dd'][d]*100:+.1f}%")
typer.echo("\n-- PER-YEAR Sharpe + maxDD (net @ headline cost; all-from-one-year => overfit) --")
for d in ("short_vol", "long_vol"):
years = rep["per_year"][d]
ys = " ".join(f"{yr}:Sh={y['sharpe']:+5.2f}/DD={y['max_dd']*100:+4.0f}%(n={y['days']})"
for yr, y in sorted(years.items()))
typer.echo(f" [{d.upper()}] {ys}")
typer.echo(f"\n-- CORRELATION with EACH existing edge (net @ {cost_bps:.1f}bps; ~0 => additive) --")
for d in ("short_vol", "long_vol"):
corrs = rep["corr_edges"][d]
parts = [f"{edge}={'N/A' if corrs.get(edge) is None else f'{corrs[edge]:+.3f}'}"
for edge in ("tstrend", "unlock", "xsfunding", "positioning")]
typer.echo(f" [{d.upper()}] " + " ".join(parts))
typer.echo("\nVERDICT GUIDE: a deployable VRP edge stays Sharpe>0 at REALISTIC OPTION cost (50/100bp), is "
"positive in MORE THAN ONE year, has a SURVIVABLE tail (maxDD not catastrophic), AND is "
"uncorrelated with all four existing edges. VRP looks great until a spike — judge it NET of "
"the worst day.")
typer.echo(f"\nCROSS-VENUE CAVEAT: {rep.get('cross_venue_note', '')}")
typer.echo("NOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched.")
def _print_vrp_walk_forward(rep: dict, *, cost_bps: float) -> None:
"""Pretty-print the tail-aware OOS / walk-forward validation report (the deploy gate)."""
typer.echo("\n=== VRP — OOS / WALK-FORWARD VALIDATION (TAIL-AWARE DEPLOY GATE, READ-ONLY) ===")
typer.echo(f"signal: Deribit DVOL (IV) vs BTC/ETH realized vol | cost: {cost_bps:.1f}bps")
if not rep.get("available"):
typer.echo(f" N/A — {rep.get('reason', '')}")
typer.echo(" (run `fxhnt ingest-dvol` + `fxhnt bybit-ingest-klines` first)")
return
th = rep["train_holdout"]
rw = rep["rolling_windows"]
full = rep["full"]
typer.echo(f" train-chosen direction: {rep['direction'].upper()} | vol-target={rep['target_ann_vol']*100:.0f}%/yr "
f"| avg leverage (k)={rep['avg_leverage']:.2f} | maxDD floor={rep['max_dd_floor']*100:.0f}%")
typer.echo(f" FULL-sample: Sharpe={full['sharpe']:6.2f} maxDD={full['max_dd']*100:+.1f}% "
f"worst day={full['worst_day']*100:+.2f}% total={full['total_return']*100:+.1f}%")
typer.echo("\n-- TRAIN/TEST (OOS) split: direction+k chosen on TRAIN, scored on held-out TEST --")
typer.echo(f" train Sharpe={th['train_sharpe']:+6.2f} -> TEST(OOS) Sharpe={th['test_sharpe']:+6.2f} "
f"TEST maxDD={th.get('test_max_dd', 0.0)*100:+.1f}% (split@{th.get('split_at', 'n/a')})")
typer.echo(f"\n-- ROLLING non-overlapping {rep['window_days']}d windows ({rw['n_windows']}); "
f"fraction positive={rw['fraction_positive']*100:.0f}% --")
for w in rw["windows"]:
typer.echo(f" {w['start']}..{w['end']} Sharpe={w['sharpe']:+6.2f} total={w['total_return']*100:+7.1f}% "
f"maxDD={w['max_dd']*100:+5.1f}% (n={w['days']})")
typer.echo(f"\n-- LOOK-AHEAD audit: causal={rep['look_ahead']['causal']} "
f"(leak_days={rep['look_ahead']['leak_days']}) --")
v = rep["verdict"]
typer.echo("\n-- VERDICT (tail-aware) --")
for name, ok in v["checks"].items():
typer.echo(f" [{'PASS' if ok else 'FAIL'}] {name}")
typer.echo(f"\n ==> {'PASS — DEPLOYABLE' if v['deployable'] else 'FAIL — NOT deployable'} "
f"(a clean 'no / too-tail-heavy' is a valid outcome)")
typer.echo(f"\nCROSS-VENUE CAVEAT: {rep.get('cross_venue_note', '')}")
typer.echo("NOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched.")
def _print_oi_deleverage_verify(rep: dict, *, scope: str, direction_note: str, cost_bps: float) -> None:
"""Pretty-print the adversarial OI-deleveraging verification report (both directions)."""
typer.echo("\n=== Bybit OI-deleveraging reversion — ADVERSARIAL VERIFICATION (READ-ONLY) ===")