"""CLI — the composition root. The ONLY place that wires concrete adapters to the application services.
Swap an adapter here (Yahoo→Databento, SQLite→Postgres) without touching any other layer."""
from __future__ import annotations

import os

import typer

from fxhnt.adapters.broker import IbkrBroker
from fxhnt.adapters.data import DatabentoDataProvider, YahooDataProvider
from fxhnt.adapters.data.futures_fetch import fetch_futures
from fxhnt.adapters.llm import build_chat_model
from fxhnt.adapters.persistence import DuckDbAnalyticalStore, SqlOperationalRepository
from fxhnt.application import (
    DiscoveryService,
    ExecutionService,
    GridGenerator,
    PortfolioEvaluationService,
    ResearchService,
)
from fxhnt.application.bybit_book_persist import _BYBIT_BOOK_SIDS, _persist_bybit_book
from fxhnt.config import Settings, get_settings
from fxhnt.domain.models import AssetClass, Market, StrategySpec
from fxhnt.domain.portfolio import Book, StrategyAllocation, book_from_runs
from fxhnt.domain.strategies import available

app = typer.Typer(help="fxhnt — agentic strategy research & multi-strategy execution", no_args_is_help=True)

fund_config_app = typer.Typer(help="Fund-level knobs (capital + risk appetite) — the allocation SSOT.")
app.add_typer(fund_config_app, name="fund-config")


@fund_config_app.command("show")
def fund_config_show() -> None:
    """Print the effective fund knobs (capital, target_vol, kelly_fraction, max_venue_weight)."""
    from fxhnt.application.fund_config import read_fund_config
    cfg = read_fund_config(get_settings().operational_dsn)
    typer.echo(f"capital={cfg.capital:.2f} target_vol={cfg.target_vol:.4f} kelly_fraction={cfg.kelly_fraction:.4f} "
              f"max_venue_weight={cfg.max_venue_weight:.4f}")


@fund_config_app.command("set")
def fund_config_set(
    capital: float = typer.Option(None, help="Fund capital (equity the allocation sizes against). > 0."),
    target_vol: float = typer.Option(None, "--target-vol", help="Annualised vol target (0, 1.0]."),
    kelly: float = typer.Option(None, "--kelly", help="Kelly fraction (0, 1.0]."),
    max_venue_weight: float = typer.Option(
        None, "--max-venue-weight",
        help="Max fraction of equity deployed to any single venue (0,1]. 1.0 = no cap."),
) -> None:
    """Set one or more fund knobs (partial override; unspecified fields unchanged)."""
    from fxhnt.application.fund_config import write_fund_config
    try:
        cfg = write_fund_config(get_settings().operational_dsn, capital=capital,
                                target_vol=target_vol, kelly_fraction=kelly,
                                max_venue_weight=max_venue_weight)
    except ValueError as e:
        typer.echo(f"error: {e}"); raise typer.Exit(code=1) from e
    typer.echo(f"set: capital={cfg.capital:.2f} target_vol={cfg.target_vol:.4f} kelly_fraction={cfg.kelly_fraction:.4f} "
              f"max_venue_weight={cfg.max_venue_weight:.4f}")


def _data_provider(data_source: str, settings: Settings) -> object:
    if data_source == "warehouse":
        from fxhnt.adapters.warehouse import get_feature_store
        from fxhnt.adapters.warehouse.price_provider import WarehousePriceProvider
        return WarehousePriceProvider(get_feature_store(settings))
    if data_source == "databento":
        d = settings.databento
        return DatabentoDataProvider(max_cost_usd=d.max_cost_usd, equity_dataset=d.equity_dataset,
                                     future_dataset=d.future_dataset, default_start=d.default_start)
    return YahooDataProvider()


def _build_service(settings: Settings) -> ResearchService:
    return ResearchService(
        data=YahooDataProvider(),
        operational=SqlOperationalRepository(settings.operational_dsn),
        analytical=DuckDbAnalyticalStore(settings.analytical_path),
        settings=settings,
    )


@app.command()
def research(
    symbol: str = typer.Argument(..., help="ticker, e.g. SPY"),
    kind: str = typer.Option("trend", help="strategy kind"),
    asset_class: AssetClass = typer.Option(AssetClass.ETF),
    window: int = typer.Option(200, help="trend MA window"),
    n_trials: int = typer.Option(1, help="full search size (for the deflated-Sharpe correction)"),
) -> None:
    """Run one (market × strategy) candidate through the gauntlet and persist the verdict."""
    settings = get_settings()
    svc = _build_service(settings)
    market = Market(symbol=symbol.upper(), asset_class=asset_class)
    spec = StrategySpec(kind=kind, params={"window": float(window)})
    run = svc.evaluate_candidate(market, spec, n_trials=n_trials)
    s, v = run.stats, run.verdict
    typer.echo(f"\n{market} | {spec.key()}")
    typer.echo(f"  backtest: CAGR {100*s.cagr:+.1f}%  vol {100*s.ann_vol:.0f}%  "
               f"Sharpe {s.sharpe:+.2f}  maxDD {100*s.max_drawdown:+.0f}%  (n={s.n_obs})")
    typer.echo(f"  gauntlet: {v.summary()}")
    for r in v.reasons:
        typer.echo(f"     - {r}")
    typer.echo(f"  -> {'PASS ✅' if v.passed else 'REJECT ❌'}  (run_id {run.run_id})\n")


@app.command("list")
def list_runs(passed_only: bool = typer.Option(False, "--passed-only")) -> None:
    """List persisted research runs (the survivor library when --passed-only)."""
    svc = _build_service(get_settings())
    runs = svc._op.list_runs(passed_only=passed_only)  # noqa: SLF001 (composition root may reach in)
    if not runs:
        typer.echo("no runs yet.")
        return
    for run in runs:
        flag = "PASS" if run.verdict.passed else "REJECT"
        typer.echo(f"  [{flag}] {run.market} {run.spec.key()} | "
                   f"Sharpe {run.stats.sharpe:+.2f} DSR {run.verdict.dsr:.2f} | {run.run_id}")


@app.command()
def strategies() -> None:
    """List available strategy kinds."""
    typer.echo("available strategies: " + ", ".join(available()))


@app.command()
def ingest(
    symbols: str = typer.Option(..., help="comma-separated tickers to fetch + cache"),
    asset_class: AssetClass = typer.Option(AssetClass.ETF),
    data_source: str = typer.Option("yahoo", help="yahoo | databento | warehouse"),
) -> None:
    """Pre-fetch + cache market data into the analytical store. Decouples slow ingestion from fast search."""
    settings = get_settings()
    provider = _data_provider(data_source, settings)
    store = DuckDbAnalyticalStore(settings.analytical_path)
    for s in (x.strip().upper() for x in symbols.split(",") if x.strip()):
        market = Market(symbol=s, asset_class=asset_class)
        ps = provider.fetch(market)  # type: ignore[attr-defined]
        store.save_prices(ps)
        typer.echo(f"cached {market}: {len(ps)} bars  {ps.dates[0]}..{ps.dates[-1]}")


@app.command()
def discover(
    symbols: str = typer.Option("SPY,QQQ,IEF,GLD", help="comma-separated tickers to search"),
    asset_class: AssetClass = typer.Option(AssetClass.ETF),
    data_source: str = typer.Option("yahoo", help="data provider: yahoo | databento"),
) -> None:
    """Sweep strategies × markets through the gauntlet (deflated over the FULL search) and persist survivors."""
    settings = get_settings()
    markets = [Market(symbol=s.strip().upper(), asset_class=asset_class) for s in symbols.split(",") if s.strip()]
    grids: dict[str, list[dict[str, float]]] = {
        "trend": [{"window": float(w)} for w in (100, 150, 200, 250)],
        "mean_reversion": [{"window": float(w), "z": 1.0} for w in (10, 20, 40)],
        "buy_hold": [{}],
    }
    svc = DiscoveryService(
        data=_data_provider(data_source, settings),  # type: ignore[arg-type]
        operational=SqlOperationalRepository(settings.operational_dsn),
        analytical=DuckDbAnalyticalStore(settings.analytical_path),
        settings=settings,
    )
    report = svc.search(GridGenerator(markets, grids))
    typer.echo(f"searched {report.n_candidates} candidates  (effective independent trials "
               f"{report.n_effective_trials:.1f}, sr_var={report.sr_variance:.5f})")
    typer.echo(f"passed the gauntlet: {report.n_passed_gauntlet}  →  "
               f"survivors after FDR over the search: {report.n_survivors}")
    for run in sorted(report.survivors, key=lambda r: -r.verdict.dsr):
        typer.echo(f"  [PASS] {run.market} {run.spec.key():28} | Sharpe {run.stats.sharpe:+.2f}  "
                   f"DSR {run.verdict.dsr:.3f}  OOS {run.verdict.oos_sharpe:+.2f}")


@app.command()
def agent(
    goal: str = typer.Option("Find the most robust diversified futures book you can (try CL, GC, ZN, 6E, ES, NQ).",
                             help="the research goal for the agent"),
    data_source: str = typer.Option("databento", help="yahoo | databento | warehouse"),
    mode: str = typer.Option("structured", help="structured (robust proposal-loop, local-friendly) | react"),
) -> None:
    """Run the research agent: it proposes diversified books, tests them via the gauntlet, and reports the
    most robust survivor. The LLM proposes; the deterministic gauntlet disposes."""
    from fxhnt.application.agent import ResearchAgent, StructuredResearchAgent, build_tools

    settings = get_settings()
    provider = _data_provider(data_source, settings)
    analytical = DuckDbAnalyticalStore(settings.analytical_path)  # ONE shared store (DuckDB single-writer)
    research = ResearchService(provider, SqlOperationalRepository(settings.operational_dsn), analytical, settings)  # type: ignore[arg-type]
    portfolio = PortfolioEvaluationService(provider, settings, analytical)  # type: ignore[arg-type]
    llm = build_chat_model(settings)
    if mode == "react":
        researcher: object = ResearchAgent(
            llm, build_tools(research, portfolio), recursion_limit=settings.agent.recursion_limit)
    else:
        researcher = StructuredResearchAgent(llm, portfolio, settings)
    typer.echo(f"agent [{settings.agent.provider}:{settings.agent.model}, mode={mode}] researching:\n  {goal}\n")
    typer.echo(researcher.run(goal))  # type: ignore[attr-defined]


@app.command("validate-book")
def validate_book(
    symbols: str = typer.Option(..., help="markets in the book (comma-separated)"),
    kind: str = typer.Option("trend", help="strategy kind for every sleeve"),
    window: int = typer.Option(200),
    asset_class: AssetClass = typer.Option(AssetClass.FUTURE),
    data_source: str = typer.Option("databento", help="yahoo | databento | warehouse"),
    max_gross: float = typer.Option(1.0, help="book gross-leverage budget"),
    n_book_trials: int = typer.Option(1, help="how many book RECIPES you tried (1 if pre-specified)"),
) -> None:
    """Evaluate a diversified BOOK (combined sleeves) through the gauntlet — the unit of alpha."""
    settings = get_settings()
    markets = [Market(symbol=s.strip().upper(), asset_class=asset_class) for s in symbols.split(",") if s.strip()]
    weight = max_gross / len(markets)
    allocations = [StrategyAllocation(spec=StrategySpec(kind=kind, params={"window": float(window)}),
                                      market=m, weight=weight) for m in markets]
    book = Book(name=f"{kind}-book", allocations=allocations, max_gross_leverage=max_gross)
    svc = PortfolioEvaluationService(_data_provider(data_source, settings), settings,  # type: ignore[arg-type]
                                     DuckDbAnalyticalStore(settings.analytical_path))
    ev = svc.evaluate_book(book, n_book_trials=n_book_trials)
    s, v = ev.stats, ev.verdict
    typer.echo(f"book '{book.name}' — {len(markets)} markets × {kind}(window={window}), gross {max_gross}")
    typer.echo(f"  combined: CAGR {100*s.cagr:+.1f}%  vol {100*s.ann_vol:.0f}%  "
               f"Sharpe {s.sharpe:+.2f}  maxDD {100*s.max_drawdown:+.0f}%  (n={ev.n_obs})")
    typer.echo(f"  gauntlet: {v.summary()}")
    for r in v.reasons:
        typer.echo(f"     - {r}")
    typer.echo(f"  -> {'PASS ✅' if v.passed else 'REJECT ❌'}")


@app.command()
def execute(
    book_name: str = typer.Option("survivors", help="name for the built book"),
    do_execute: bool = typer.Option(False, "--execute", help="place orders (default: dry-run plan only)"),
    live: bool = typer.Option(False, "--live", help="allow a non-paper account (requires a DU-free account)"),
    max_gross: float = typer.Option(1.0, help="book gross-leverage budget"),
) -> None:
    """Build a book from the survivor library and rebalance it on IBKR (dry-run by default)."""
    settings = get_settings()
    if live:
        settings.execution.allow_live = True
    runs = SqlOperationalRepository(settings.operational_dsn).list_runs(passed_only=True)
    try:
        book = book_from_runs(runs, book_name, max_gross_leverage=max_gross)
    except ValueError as e:
        typer.echo(f"cannot build book: {e}  (run `fxhnt research ...` first)")
        raise typer.Exit(1) from e
    typer.echo(f"book '{book.name}': {len(book.allocations)} sleeves over {len(book.markets())} markets")
    broker_cm: Any = IbkrBroker(settings.ibkr.host, settings.ibkr.port, settings.ibkr.client_id,
                                settings.ibkr.timeout)
    where = f"IB Gateway/TWS at {settings.ibkr.host}:{settings.ibkr.port}"
    try:
        with broker_cm as brk:
            plan = ExecutionService(YahooDataProvider(), brk, settings).rebalance(book, execute=do_execute)
    except Exception as e:  # connection / runtime — keep the message clean
        typer.echo(f"broker error: {type(e).__name__}: {str(e)[:100]}")
        typer.echo(f"  reachable? {where} (creds set?)")
        raise typer.Exit(1) from e
    typer.echo(f"NLV ${plan.nlv:,.0f}  | targets {plan.target_weights}")
    for n in plan.notes:
        typer.echo(f"  - {n}")
    for o in plan.orders:
        typer.echo(f"  {o.side} {o.quantity:g} {o.symbol}  (~${o.est_value:,.0f})")
    status = "BLOCKED" if plan.blocked else ("EXECUTED" if plan.executed else "DRY-RUN")
    typer.echo(f"-> {status}")


def _exec_venue_string(broker_name: str, is_paper: bool) -> str:
    return f"{broker_name}-{'paper' if is_paper else 'live'}"


def _resolve_paper_envelope(flag: float, config_default: float) -> float:
    """Explicit flag wins; otherwise the shared settings default. Both default 0.0 = off (B2a behavior).
    A negative flag is an operator typo, not "unset" — reject it rather than silently falling back to the
    config default (which would hide the typo behind a seemingly-valid run)."""
    if flag < 0.0:
        raise typer.BadParameter("--paper-envelope must be >= 0")
    return flag if flag > 0.0 else config_default


@app.command("execute-multistrat")
def execute_multistrat(
    do_execute: bool = typer.Option(False, "--execute", help="place orders (default: dry-run plan only)"),
    live: bool = typer.Option(False, "--live", help="allow a non-paper account"),
    max_gross: float = typer.Option(2.0, help="gross budget (risk-parity multistrat can sit modestly above 1)"),
    paper_envelope: float = typer.Option(
        0.0, "--paper-envelope",
        help="PAPER-only validation envelope in $ (bypasses B2a; refused on live). 0 = B2a-gated."),
    venue_kind: str = typer.Option("us", "--venue", help="'us' (US ETFs, default) or 'ucits' (Euronext UCITS)"),
) -> None:
    """Compute the adaptive multi-strat book's CURRENT target weights and rebalance them on IBKR — the 5
    pure-equity streams (SPY/IEF/GLD/PDBC/DBMF) as whole shares. BTC-USD is DROPPED from the fund book
    (crypto exposure comes from the Bybit book; BTC only added correlation) so this is a pure-equity leg,
    no crypto order. Dry-run by default. Same ExecutionService safety gates as `execute`."""
    import datetime as _dt
    import logging

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.adapters.persistence.ibkr_account_repo import IbkrAccountRepo
    from fxhnt.application.multistrat_exec_record import plan_and_record
    from fxhnt.domain.strategies import multistrat

    if venue_kind not in ("us", "ucits"):
        raise typer.BadParameter("--venue must be 'us' or 'ucits'")

    settings = get_settings()
    if live:
        settings.execution.allow_live = True
    data = YahooDataProvider()
    closes: dict[str, dict[str, float]] = {}
    price_now: dict[str, float] = {}
    for sym in multistrat.FUND_INSTRUMENTS:                  # 5 pure-equity ETFs (no BTC in the fund book)
        ps = data.fetch(Market(symbol=sym, asset_class=AssetClass.ETF))
        closes[sym] = {d: float(c) for d, c in zip(ps.dates, ps.close)}
        price_now[sym] = float(ps.close[-1])
    weights = multistrat.target_weights(closes, multistrat.FUND_INSTRUMENTS)
    if not weights:
        typer.echo("multistrat: insufficient history for target weights — no action")
        raise typer.Exit(1)

    # all 5 are equities (Yahoo ticker == IBKR equity symbol) — no crypto symbol mapping needed. For the US
    # venue these are the broker symbols + prices directly; the UCITS venue re-routes them (symbols + REAL
    # UCITS prices) once the broker connection is up, below.
    bweights = dict(weights)
    bprices = dict(price_now)
    contract_specs: dict[str, dict[str, str]] = {}
    typer.echo("multistrat target weights: " + ", ".join(f"{s} {w:+.3f}" for s, w in bweights.items()))

    if venue_kind == "ucits":
        # Validate the UCITS map BEFORE connecting to the broker so a config bug (two sleeves -> one UCITS
        # ticker) surfaces as a clear "UCITS map collision", not the broker block's "paper-envelope refused".
        from fxhnt.application.ucits_map import apply_ucits_map
        try:
            apply_ucits_map(weights, settings.ucits.map)
        except ValueError as e:
            typer.echo(f"UCITS map error: {e}")
            raise typer.Exit(1) from e

    repo = ForwardNavRepo(settings.operational_dsn)
    repo.migrate()
    # D2.1 real-account capture (fills + NAV snapshot) — best-effort observability, NOT the trading path.
    # Guard the migration too: a broken capture migration (DB unreachable / ALTER lock) must NEVER prevent a
    # real rebalance. On failure, drop the capture repo (None) so plan_and_record simply skips the capture.
    account_repo: Any = IbkrAccountRepo(settings.operational_dsn)
    try:
        account_repo.migrate()
    except Exception as e:
        logging.getLogger(__name__).warning(
            "IBKR account-capture migration failed — capture disabled this run (best-effort): %s", e)
        account_repo = None
    today = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%d")
    at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)

    # execution venue: IBKR (whole-share, via ib-gateway) — the sole Broker port implementation.
    broker: Any = IbkrBroker(settings.ibkr.host, settings.ibkr.port,
                             settings.ibkr.client_id, settings.ibkr.timeout)
    where = f"IBKR ({settings.ibkr.host}:{settings.ibkr.port})"
    try:
        with broker as brk:
            svc = ExecutionService(data, brk, settings)
            acct = brk.account_state()
            paper_env = _resolve_paper_envelope(paper_envelope, settings.paper_validation.envelope)
            venue = _exec_venue_string(brk.name, acct.is_paper)
            if venue_kind == "ucits":
                # Re-route to UCITS symbols + REAL UCITS USD prices from IBKR (a UCITS ETF's price != its US
                # cousin's). Sleeves that don't map / resolve / price are logged + skipped here; survivors
                # keep their own weight (the book is NOT rescaled). contract_specs carry the ISIN so the order
                # resolves in place_order.
                from fxhnt.application.ucits_map import route_ucits
                venue = "ibkr-ucits"
                bweights, bprices, contract_specs, skipped = route_ucits(
                    weights, settings.ucits.map,
                    price_fn=lambda lst: brk.market_price(lst.ticker, lst.exchange, lst.currency,
                                                          "ISIN", lst.isin))
                for us_sym, w in skipped.items():
                    typer.echo(f"UCITS routing: SKIPPED sleeve {us_sym} (weight {w:+.3f}) — unmapped or "
                               "unresolved/unpriced on IBKR; not routed, book NOT rescaled")
                if not bweights:
                    typer.echo("UCITS routing: no sleeve resolved — nothing to route, no action")
                    raise typer.Exit(1)
                if skipped:
                    dropped = sum(abs(w) for w in skipped.values())
                    typer.echo(f"UCITS routing: {len(skipped)} sleeve(s) skipped, {dropped:.1%} of gross "
                               "weight un-deployed")
                typer.echo("UCITS routing: " + ", ".join(
                    f"{s} {w:+.3f} @ ${bprices[s]:,.2f}" for s, w in bweights.items()))
            plan, envelope = plan_and_record(
                repo=repo, exec_svc=svc, within_weights=bweights, prices=bprices, nlv=acct.nlv,
                today=today, at=at, max_gross=max_gross, execute=do_execute,
                paper_envelope=paper_env, account_is_paper=acct.is_paper, venue=venue,
                account_repo=account_repo, broker=brk, contract_specs=contract_specs)
    except typer.Exit:  # a deliberate early exit (e.g. UCITS: no sleeve resolved) — not a broker error
        raise
    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 error, keep it that way
        raise typer.Exit(1) from e
    except Exception as e:  # connection / runtime
        typer.echo(f"broker error: {type(e).__name__}: {str(e)[:100]}")
        typer.echo(f"  reachable? {where} — creds/gateway set?")
        raise typer.Exit(1) from e
    typer.echo(f"NLV ${plan.nlv:,.0f}  | envelope ${envelope:,.0f} | orders {len(plan.orders)}")
    for n in plan.notes:
        typer.echo(f"  - {n}")
    for o in plan.orders:
        typer.echo(f"  {o.side} {o.quantity:g} {o.symbol}  (~${o.est_value:,.0f})")
    # A DEGRADED run: execution was requested and orders were attempted, but a leg did not fill (Error
    # 478/10349, or a broker raise) so the plan is NOT executed and NOT a clean dry-run. It MUST exit non-zero
    # — a silent exit 0 makes the CronJob report Complete, and the executed forward track + reconciliation gate
    # would book a rebalance that only partially happened (real money errs false-WAIT). Live 2026-07-20.
    degraded = do_execute and bool(plan.orders) and not plan.executed and not plan.blocked
    status = "BLOCKED" if plan.blocked else ("EXECUTED" if plan.executed else ("DEGRADED" if degraded else "DRY-RUN"))
    typer.echo(f"-> {status}")
    if degraded:
        raise typer.Exit(1)


@app.command("execute-bybit")
def execute_bybit(
    do_execute: bool = typer.Option(False, "--execute", help="place orders (default: dry-run plan only)"),
    paper_envelope: float = typer.Option(
        0.0, "--paper-envelope",
        help="PAPER-only validation envelope in $ (bypasses B2a; refused on live). 0 = B2a-gated."),
) -> None:
    """Reconcile + record the bybit_4edge book on Bybit TESTNET (ported off Dagster). Dry-run by default;
    same safety guards (kill-switch, NAV-marker idempotency, open-orders guard). Records bybit_4edge_exec."""
    from fxhnt.application.bybit_testnet_run import run_bybit_testnet

    settings = get_settings()
    paper_env = _resolve_paper_envelope(paper_envelope, settings.paper_validation.envelope)
    try:
        out = run_bybit_testnet(settings, paper_envelope=paper_env, execute=do_execute, log=typer.echo)
    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 raw traceback
        raise typer.Exit(1) from e
    typer.echo(f"-> placed {out.get('placed', 0)} | envelope ${out.get('envelope', 0.0):,.0f} "
               f"| exec_return {out.get('exec_return')}")


@app.command("migrate-cockpit")
def migrate_cockpit() -> None:
    """Create the cockpit tables (+ TimescaleDB hypertable on Postgres) and seed the strategy registry."""
    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.application.fund_config import seed_fund_config
    dsn = get_settings().operational_dsn
    ForwardNavRepo(dsn).migrate()
    seed_fund_config(dsn)
    typer.echo(f"cockpit migrated: {dsn.split('@')[-1]}")


@app.command("migrate-forward-anchors")
def migrate_forward_anchors() -> None:
    """One-time: seed forward_anchor rows from the on-disk *_state.json trackers (spec sub-project A §6).
    Recompute tracks reconcile (divergence => drift, logged); observed tracks import their booked log."""
    import datetime as _dt
    import json
    from pathlib import Path

    from fxhnt.adapters.orchestration.assets import _data_dir
    from fxhnt.adapters.orchestration.migration_builders import track_builders
    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.application.forward_migration import migrate_track
    from fxhnt.config import get_settings
    from fxhnt.registry import STRATEGY_REGISTRY

    s = get_settings()
    data_dir = _data_dir()                  # the trackers' state directory (reuse the asset helper)
    repo = ForwardNavRepo(s.operational_dsn)
    repo.migrate()
    today = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%d")
    at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)

    # strategy_id -> (state_file basename, build_strategy) for the recompute reconcile. The build callables
    # mirror each *_nav asset. Observed tracks pass a no-op strategy (their record is imported, not recomputed).
    builders = track_builders(s, data_dir)

    for sid in STRATEGY_REGISTRY:
        sf, build = builders.get(sid, (None, None))
        if sf is None:
            typer.echo(f"{sid}: no state-file mapping, skipped")
            continue
        path = Path(data_dir) / f"{sf}.json"
        if not path.exists():
            typer.echo(f"{sid}: no {path.name}, skipped (fresh track)")
            continue
        state = json.loads(path.read_text())
        out = migrate_track(repo, sid, state, build(), today=today, at=at)
        typer.echo(f"{sid}: {out}")


@app.command("unlock-calendar-import")
def unlock_calendar_import(
    json_path: str = typer.Option(
        None, "--json",
        help="Path to the existing unlock_calendar.json. Default: $FXHNT_SURFER_DATA_DIR/unlock_calendar.json."),
) -> None:
    """One-time migration: load the existing `unlock_calendar.json` into the operational `unlock_events`
    table, so the unlock sleeve / evals / Jobs read the cliff calendar over the network (no surfer-data PVC).
    Idempotent (upsert keyed on (sym, cliff_day)) — safe to re-run."""
    import json
    from pathlib import Path

    from fxhnt.adapters.orchestration.assets import _data_dir
    from fxhnt.adapters.persistence.unlock_calendar_repo import UnlockCalendarRepo

    path = json_path or f"{_data_dir()}/unlock_calendar.json"
    events = json.loads(Path(path).read_text())
    dsn = get_settings().operational_dsn
    repo = UnlockCalendarRepo(dsn)
    repo.migrate()
    repo.write_events(events)
    typer.echo(f"unlock-calendar imported: {len(events)} events from {path} -> "
               f"unlock_events @ {dsn.split('@')[-1]} (table now has {len(repo.read_events())})")


@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
    (25y range) into `yahoo_pit_volume` — the ADV input the equity-leg cost/capacity model consumes.
    PIT first-seen-frozen (mirrors `snapshot_yahoo_closes`): a re-run never overwrites an already-stored
    (symbol, date). A per-symbol fetch failure is logged and skipped, never aborts the whole backfill."""
    import datetime as _dt

    from fxhnt.adapters.data.yahoo_daily import YahooDailyClient
    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.domain.strategies.multistrat import FUND_INSTRUMENTS

    dsn = get_settings().operational_dsn
    repo = ForwardNavRepo(dsn)
    repo.migrate()
    at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)
    live = YahooDailyClient(rng="25y")
    for sym in FUND_INSTRUMENTS:
        try:
            repo.snapshot_yahoo_volumes(sym, live.volumes(sym), at)
            typer.echo(f"backfill-etf-volume: {sym} -> {len(repo.yahoo_pit_volumes(sym))} days frozen")
        except Exception as e:  # noqa: BLE001 — a bad symbol (network/Yahoo) must not abort the whole backfill
            typer.echo(f"backfill-etf-volume: {sym} FAILED ({type(e).__name__}: {str(e)[:120]}) — skipped")

    # Also freeze each UCITS LSE line's raw daily volume (the ucits-regime ADV input) — a thin/absent line
    # (e.g. DBMF.L, new+small) is logged + skipped, never fatal to the rest of the backfill.
    for listing in get_settings().ucits.map.values():
        try:
            repo.snapshot_yahoo_volumes(listing.yahoo_ticker, live.volumes(listing.yahoo_ticker), at)
            typer.echo(f"backfill-etf-volume: {listing.yahoo_ticker} -> "
                       f"{len(repo.yahoo_pit_volumes(listing.yahoo_ticker))} days frozen")
        except Exception as e:  # noqa: BLE001 — a thin/absent UCITS line must not abort the whole backfill
            typer.echo(f"backfill-etf-volume: {listing.yahoo_ticker} FAILED "
                       f"({type(e).__name__}: {str(e)[:120]}) — skipped")


@app.command("backfill-ucits-volume-ibkr")
def backfill_ucits_volume_ibkr() -> None:
    """One-time/resumable historical freeze of each `settings.ucits.map` line's REAL on-exchange daily volume
    (`IbkrBroker.historical_daily_volume`, `reqHistoricalData TRADES`) into `ibkr_pit_volume` — the alternate
    ADV input `equity_allocation_inputs` reads when `FXHNT_ALLOCATION_UCITS_VOLUME_SOURCE=ibkr` (see
    `AllocationSettings.ucits_volume_source`). Network-gated: only runs where the NetworkPolicy allows
    `ib-gateway:4004` (the `multistrat-rebalancer`-labeled CronJob) — the daemon cannot reach it.

    PIT first-seen-frozen (mirrors `backfill-etf-volume`): a re-run never overwrites an already-stored
    (symbol, date). A per-line resolve-fail/thin-line/network error is logged and skipped, never aborts the
    rest of the backfill. Reuses ONE IBKR connection across every line (`with broker as brk:`) — the
    documented lesson that reconnecting per call causes clientId churn (see `ibkr.py`'s module docstring)."""
    import datetime as _dt

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo

    s = get_settings()
    repo = ForwardNavRepo(s.operational_dsn)
    repo.migrate()
    at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)

    broker = IbkrBroker(host=s.ibkr.host, port=s.ibkr.port, client_id=s.ibkr.client_id)
    with broker as brk:
        for listing in s.ucits.map.values():
            try:
                series = brk.historical_daily_volume(
                    ticker=listing.ticker, isin=listing.isin, exchange=listing.exchange,
                    currency=listing.currency, days=90)
                repo.snapshot_ibkr_volumes(listing.yahoo_ticker, series, at)
                typer.echo(f"backfill-ucits-volume-ibkr: {listing.yahoo_ticker} -> "
                           f"{len(repo.ibkr_pit_volumes(listing.yahoo_ticker))} days frozen")
            except Exception as e:  # noqa: BLE001 — a thin/unresolvable line must not abort the backfill
                typer.echo(f"backfill-ucits-volume-ibkr: {listing.yahoo_ticker} FAILED "
                           f"({type(e).__name__}: {str(e)[:120]}) — skipped")


@app.command("probe-ucits-symbols")
def probe_ucits_symbols() -> None:
    """READ-ONLY: for every `settings.ucits.map` line, resolve it on IBKR by ISIN and print the SYMBOL IBKR
    actually assigns it (`symbol` / `localSymbol` / `conId`) vs the config `ticker`. Detects where the cosmetic
    display ticker diverges from what IBKR REPORTS a holding under — e.g. IEF's ticker `CBU0` resolves to the
    canonical `CSBGU0`, which then flags a legitimate book position as a stray in the NLV view. Run it (via the
    multistrat-rebalancer-labeled path that can reach ib-gateway:4004) to populate each line's `ibkr_symbol`
    config field. Places NO orders; resolves + prints only."""
    s = get_settings()
    broker = IbkrBroker(host=s.ibkr.host, port=s.ibkr.port, client_id=s.ibkr.client_id)
    typer.echo(f"{'ticker':<8}{'ibkr_symbol':<14}{'local_symbol':<14}{'conId':<12}{'DIVERGES?'}")
    typer.echo("-" * 62)
    with broker as brk:
        for listing in s.ucits.map.values():
            info = brk.resolve_symbol_info(ticker=listing.ticker, isin=listing.isin,
                                           exchange=listing.exchange, currency=listing.currency)
            if info is None:
                typer.echo(f"{listing.ticker:<8}{'(unresolved)':<40}")
                continue
            reported = info["ibkr_symbol"] or info["local_symbol"]
            diverges = "YES -> set ibkr_symbol" if reported and reported != listing.ticker else "no"
            typer.echo(f"{listing.ticker:<8}{info['ibkr_symbol']:<14}{info['local_symbol']:<14}"
                       f"{info['con_id']:<12}{diverges}")


@app.command("ingest-forward")
def ingest_forward() -> None:
    """Evaluate + persist the forward gate for every track the engine has recorded in the cockpit DB."""
    from fxhnt.application.forward_ingest import ingest_forward_state
    ingest_forward_state(get_settings().operational_dsn)
    typer.echo("ingested tracker(s) from the DB record")


@app.command()
def serve(
    host: str = typer.Option("0.0.0.0", "--host"),
    port: int = typer.Option(8080, "--port"),
) -> None:
    """Run the cockpit web app (uvicorn). Reads the operational DSN from settings."""
    import uvicorn

    from fxhnt.adapters.web.app import create_app_from_settings
    uvicorn.run(create_app_from_settings(), host=host, port=port)


@app.command()
def fetch(
    data_dir: str = typer.Option(..., "--data-dir", help="dir for <root>.dbn"),
) -> None:
    """Fetch futures data (Databento) into data_dir. (The crypto_pit leg was retired — Phase 0b Task 7e;
    Bybit is fxhnt's sole crypto venue and its data is ingested straight to the warehouse, not via a
    local npz fetch.)"""
    if not os.environ.get("DATABENTO_API_KEY"):
        typer.echo("error: fetch requires DATABENTO_API_KEY in the environment", err=True)
        raise typer.Exit(1)
    s = get_settings().databento
    import datetime as _dt
    start = (_dt.date.today() - _dt.timedelta(days=365 * s.fetch_years)).isoformat()   # rolling window
    recs, ok, spent = fetch_futures(data_dir, start=start, max_cost_usd=s.max_cost_usd)
    typer.echo(f"futures: {ok} roots, {recs} records, ${spent:.2f} (since {start}) -> {data_dir}")


@app.command("warehouse-ingest")
def warehouse_ingest(
    data_dir: str = typer.Option(..., "--data-dir", help="dir with <symbol>.dbn bronze data"),
    source: str = typer.Option(..., "--source", help="futures (the only supported source)"),
    symbol: str = typer.Option("", "--symbol", help="root for <data_dir>/<symbol>.dbn (required)"),
) -> None:
    """Bronze -> silver -> gold: normalize raw futures bars and write them into the warehouse SSOT.
    (The crypto_pit bulk-ingest leg was retired — Phase 0b Task 7e; Bybit ingest is its own dedicated
    CLI (`bybit-ingest-*`), not routed through this futures-only command.)"""
    import os

    from fxhnt.adapters.warehouse import get_feature_store
    from fxhnt.adapters.warehouse.silver_futures import normalize_futures
    from fxhnt.application.warehouse_ingest import WarehouseIngest

    settings = get_settings()
    settings.ensure_dirs()

    if source != "futures":
        typer.echo("error: --source must be 'futures'", err=True)
        raise typer.Exit(1)
    if not symbol:
        typer.echo("error: --symbol is required for --source futures", err=True)
        raise typer.Exit(1)

    store = get_feature_store(settings)
    ingest = WarehouseIngest(store)
    try:
        from fxhnt.adapters.data.dbn_local import read_front_arrays
        days, inst, close, high, low = read_front_arrays(os.path.join(data_dir, f"{symbol}.dbn"), symbol)
        n = ingest.ingest_bars(symbol, normalize_futures(symbol, days, inst, close, high, low))
        typer.echo(f"futures {symbol}: {n} bars -> warehouse")
    finally:
        store.close()


@app.command("ingest-historical")
def ingest_historical(
    limit: int = typer.Option(
        0, "--limit",
        help="ingest only the first N survivorship-free tickers (0 = full set; use a small N to validate)",
    ),
) -> None:
    """B3a: resumable, survivorship-free historical daily-EOD ingest into the DEDICATED backtest warehouse
    (settings.backtest_warehouse_path — its OWN PVC, never the live cockpit warehouse). Re-running is safe
    and cheap (already-processed tickers are skipped). Pass --limit N for a small-batch validation run."""
    from pathlib import Path

    from fxhnt.adapters.data.tiingo_daily import TiingoDailyClient
    from fxhnt.adapters.data.tiingo_universe import TiingoUniverseSource
    from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
    from fxhnt.application.historical_eod_ingest import HistoricalEodIngest

    settings = get_settings()
    Path(settings.backtest_warehouse_path).parent.mkdir(parents=True, exist_ok=True)

    tickers: list[str] | None = None
    if limit > 0:
        tickers = [t[0] for t in TiingoUniverseSource().survivorship_free()[:limit]]
        typer.echo(f"ingest-historical: small-batch validation — first {len(tickers)} tickers")

    store = DuckDbFeatureStore(settings.backtest_warehouse_path)
    try:
        ingest = HistoricalEodIngest(TiingoDailyClient(), store, TiingoUniverseSource())
        counts = ingest.ingest(tickers=tickers)
    finally:
        store.close()
    typer.echo(
        f"ingest-historical -> {settings.backtest_warehouse_path}: "
        f"{counts['tickers']} ingested, {counts['skipped']} skipped, {counts['empty']} empty, "
        f"{counts['rows']} rows, {counts['members']} membership rows"
    )


@app.command("backtest-equity")
def backtest_equity(
    n: int = typer.Option(150, "--n", help="PIT universe size (top-N by trailing dollar-volume)."),
    cost_bps: float = typer.Option(15.0, "--cost-bps", help="Cost in bps per unit (two-way) turnover."),
    borrow_annual: float = typer.Option(0.0, "--borrow-annual", help="Annual borrow cost on short notional."),
    out: str = typer.Option("/backtest-data/equity_backtest_report.json", "--out", help="Report JSON path."),
    pit: bool = typer.Option(
        True, "--pit/--no-pit",
        help="POINT-IN-TIME universe: rank each month by TRAILING dollar-volume (data <= rebalance "
             "date). Removes peak-full-history lookahead. Default. --no-pit reverts to the legacy "
             "candidate/membership path."),
    trailing_window_months: int = typer.Option(
        12, "--trailing-window-months",
        help="PIT trailing window (months) for the dollar-volume rank/alive gate."),
    candidate_top_k: int = typer.Option(
        2000, help="Bounded candidate universe: top-K most-liquid names (ONLY used with --no-pit)."),
    candidate_min_history: int = typer.Option(
        300,
        help="Min daily bars (close+volume) a name needs to be a candidate (ONLY used with --no-pit)."),
    delisting_return: float = typer.Option(
        -0.30, "--delisting-return",
        help="Terminal return booked when a held name delists (data series ends mid-hold). -0.30 = "
             "Shumway canonical; 0.0 = none (biased); -1.0 = worst case."),
    persist_cockpit: bool = typer.Option(
        False, "--persist-cockpit",
        help="Also upsert each construction's verdict into the cockpit DB (operational_dsn) for the "
             "dashboard."),
) -> None:
    """B3b: walk-forward backtest of the 3 price-only equity-factor constructions against the DEDICATED
    survivorship-free backtest warehouse (settings.backtest_warehouse_path) -> gauntlet verdict + JSON report."""
    import json

    from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
    from fxhnt.application.equity_backtest_runner import EquityBacktestRunner, evaluate_constructions

    s = get_settings()
    store = DuckDbFeatureStore(s.backtest_warehouse_path)
    try:
        result = EquityBacktestRunner(
            store, n=n, cost_bps_per_turnover=cost_bps, borrow_annual=borrow_annual,
            pit=pit, trailing_window_months=trailing_window_months,
            candidate_top_k=candidate_top_k if not pit else None,
            candidate_min_history=candidate_min_history,
            delisting_return=delisting_return,
        ).run()
        report = evaluate_constructions(
            result,
            oos_fraction=s.gauntlet.oos_fraction, dsr_min=s.gauntlet.dsr_min,
            oos_min_sharpe=s.gauntlet.oos_min_sharpe, max_is_oos_decay=s.gauntlet.max_is_oos_decay,
        )
    finally:
        store.close()
    with open(out, "w") as fh:
        json.dump(report, fh, indent=2)
    if persist_cockpit:
        import datetime as dt

        from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
        from fxhnt.application.backtest_ingest import backtest_summaries_from_report
        at = dt.datetime.now(dt.UTC).replace(tzinfo=None)  # naive-UTC, matches forward ingest
        repo = ForwardNavRepo(s.operational_dsn)
        repo.migrate()
        for sm in backtest_summaries_from_report(report):
            repo.upsert_backtest_summary(sm, at=at)
        typer.echo(f"backtest-equity: persisted {len(report['constructions'])} verdict(s) to cockpit")
    for c, block in report["constructions"].items():
        v = block["verdict"]
        typer.echo(f"backtest-equity {c}: passed={v['passed']} dsr={v['dsr']:.3f} "
                   f"is_sharpe={v['is_sharpe']:.2f} oos_sharpe={v['oos_sharpe']:.2f} "
                   f"sharpe={block['stats']['sharpe']:.2f} -> {out}")


# backtest-funding (the survivorship-free crypto_pit funding-dispersion sweep) RETIRED — Phase 0b Task 7e:
# its only data source was CryptoPitPanelSource (the crypto_pit fetch/panel surface), now gone. The generic
# `FundingBacktestRunner`/`evaluate_funding_matrix` engine in funding_backtest_runner.py is venue-agnostic
# (takes a plain Panel, no crypto_pit import) and stays — a future Bybit-fed caller can still use it.


@app.command("bybit-ingest-funding")
def bybit_ingest_funding(
    from_month: str = typer.Option(
        "", "--from", help="floor for the backward pagination, YYYY-MM (default: the 2020-01 backstop)."),
    symbols: str = typer.Option(
        "all", "--symbols",
        help="'all' (build the universe) | comma-separated | @path-to-file (one symbol per line)."),
    include_delisted: bool = typer.Option(
        False, "--include-delisted/--active-only",
        help="union the active universe with the public.bybit.com/trading delisted dirs "
             "(survivorship-free). Default: active only."),
    resume: bool = typer.Option(
        True, "--resume/--no-resume",
        help="skip symbols already carrying 'funding' in bybit_features (resume a partial/OOM-killed run). "
             "--no-resume re-fetches every symbol (same final state; idempotent upsert)."),
) -> None:
    """Phase A: ingest Bybit USDT-perp FUNDING (the carry signal), survivorship-free, into the Bybit-namespaced
    TimescaleDB table `bybit_features` (operational DSN — permanent, networked; NOT a PVC/parquet side-store,
    because live execution will be on Bybit). Daily funding = SUM of each day's 8h rates. Resumable/idempotent."""
    import datetime as _dt

    from fxhnt.adapters.data.bybit_funding import BybitFunding
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_funding_ingest import ingest_bybit_funding

    settings = get_settings()
    bybit = BybitFunding()

    # Resolve the floor (YYYY-MM -> ms) for the backward pagination.
    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

    # Resolve the symbol universe.
    if symbols == "all":
        universe = bybit.ever_listed_universe() if include_delisted else bybit.active_universe()
        kind = "ever-listed" if include_delisted else "active"
    elif symbols.startswith("@"):
        with open(symbols[1:]) as fh:
            universe = [ln.strip().upper() for ln in fh if ln.strip()]
        kind = "explicit"
    else:
        universe = [s.strip().upper() for s in symbols.split(",") if s.strip()]
        kind = "explicit"
    if not universe:
        typer.echo("error: empty symbol universe", err=True)
        raise typer.Exit(1)
    typer.echo(f"bybit-ingest-funding: {len(universe)} symbols ({kind}), "
               f"floor={from_month or '2020-01 (default)'} "
               f"-> bybit_features @ {settings.operational_dsn.split('@')[-1]}")

    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    try:
        summary = ingest_bybit_funding(
            store, bybit, universe, start_ms=start_ms, resume=resume, progress=typer.echo)
    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"\nbybit-ingest-funding done: {summary['symbols']} symbols, "
               f"{summary['day_rows']} day-rows, oldest date {oldest_date}")


@app.command("bybit-ingest-long-short")
def bybit_ingest_long_short(
    symbols: str = typer.Option(
        "liquid", "--symbols",
        help="'liquid' (the liquid_universe subset) | 'all' (every active USDT perp) | comma-separated | "
             "@path-to-file (one symbol per line)."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) used by --symbols liquid."),
    from_month: str = typer.Option(
        "", "--from", help="floor for the backward pagination, YYYY-MM (default: the 2023-01 backstop)."),
    workers: int = typer.Option(
        12, "--workers", help="bounded concurrent FETCH workers (writes stay serialized in the main thread)."),
    resume: bool = typer.Option(
        True, "--resume/--no-resume",
        help="skip symbols already carrying 'long_ratio' in bybit_features (resume a partial/OOM-killed run). "
             "--no-resume re-fetches every symbol (same final state; idempotent upsert)."),
) -> None:
    """Ingest the Bybit long/short ACCOUNT ratio (retail positioning) into the Bybit-namespaced TimescaleDB
    table `bybit_features` as feature 'long_ratio' (buyRatio = fraction of accounts LONG). period=1d → one
    value/day; the bet built off it (bybit-positioning-eval) is on PRICE. Resumable/idempotent; ONLY the
    long_ratio feature is written — funding/close are the other ingests' job.

    'liquid' resolves the symbol set via `liquid_universe(bybit_features)` (needs the kline/turnover ingest
    to have populated bybit_features first); 'all' builds the active USDT-perp universe from instruments-info."""
    import datetime as _dt

    from fxhnt.adapters.data.bybit_account_ratio import BybitAccountRatio
    from fxhnt.adapters.data.bybit_funding import BybitFunding
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.bybit_long_short_ingest import ingest_bybit_long_short

    settings = get_settings()
    bybit = BybitAccountRatio()

    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

    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    if symbols == "liquid":
        universe = sorted(liquid_universe(store, min_dollar_vol=min_dollar_vol))
        kind = f"liquid (median turnover >= ${min_dollar_vol:,.0f})"
    elif symbols == "all":
        universe = BybitFunding().active_universe()
        kind = "all active USDT perps"
    elif symbols.startswith("@"):
        with open(symbols[1:]) as fh:
            universe = [ln.strip().upper() for ln in fh if ln.strip()]
        kind = "explicit"
    else:
        universe = [s.strip().upper() for s in symbols.split(",") if s.strip()]
        kind = "explicit"
    if not universe:
        typer.echo("error: empty symbol universe (for 'liquid', run the Bybit kline/turnover ingest first)",
                   err=True)
        store.close()
        raise typer.Exit(1)
    typer.echo(f"bybit-ingest-long-short: {len(universe)} symbols ({kind}), "
               f"floor={from_month or '2023-01 (default)'}, workers={workers} "
               f"-> bybit_features @ {settings.operational_dsn.split('@')[-1]}")

    try:
        summary = ingest_bybit_long_short(
            store, bybit, universe, start_ms=start_ms, resume=resume, workers=workers, progress=typer.echo)
    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"\nbybit-ingest-long-short done: {summary['symbols']} symbols, "
               f"{summary['day_rows']} day-rows, oldest date {oldest_date}")


@app.command("bybit-ingest-oi")
def bybit_ingest_oi(
    symbols: str = typer.Option(
        "liquid", "--symbols",
        help="'liquid' (the liquid_universe subset) | 'all' (every active USDT perp) | comma-separated | "
             "@path-to-file (one symbol per line)."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) used by --symbols liquid."),
    from_month: str = typer.Option(
        "", "--from", help="floor for the backward pagination, YYYY-MM (default: the 2023-01 backstop)."),
    workers: int = typer.Option(
        12, "--workers", help="bounded concurrent FETCH workers (writes stay serialized in the main thread)."),
    resume: bool = typer.Option(
        True, "--resume/--no-resume",
        help="skip symbols already carrying 'open_interest' in bybit_features (resume a partial/OOM-killed "
             "run). --no-resume re-fetches every symbol (same final state; idempotent upsert)."),
) -> None:
    """Ingest the Bybit OPEN INTEREST (deleveraging-flow vehicle) into the Bybit-namespaced TimescaleDB table
    `bybit_features` as feature 'open_interest'. intervalTime=1d → one value/day; the bet built off it
    (bybit-oi-deleverage-eval) is on PRICE reversion after a forced OI unwind. Resumable/idempotent; ONLY the
    open_interest feature is written — funding/close/long_ratio are the other ingests' job.

    'liquid' resolves the symbol set via `liquid_universe(bybit_features)` (needs the kline/turnover ingest to
    have populated bybit_features first); 'all' builds the active USDT-perp universe from instruments-info."""
    import datetime as _dt

    from fxhnt.adapters.data.bybit_funding import BybitFunding
    from fxhnt.adapters.data.bybit_open_interest import BybitOpenInterest
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.bybit_oi_ingest import ingest_bybit_oi

    settings = get_settings()
    bybit = BybitOpenInterest()

    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

    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    if symbols == "liquid":
        universe = sorted(liquid_universe(store, min_dollar_vol=min_dollar_vol))
        kind = f"liquid (median turnover >= ${min_dollar_vol:,.0f})"
    elif symbols == "all":
        universe = BybitFunding().active_universe()
        kind = "all active USDT perps"
    elif symbols.startswith("@"):
        with open(symbols[1:]) as fh:
            universe = [ln.strip().upper() for ln in fh if ln.strip()]
        kind = "explicit"
    else:
        universe = [s.strip().upper() for s in symbols.split(",") if s.strip()]
        kind = "explicit"
    if not universe:
        typer.echo("error: empty symbol universe (for 'liquid', run the Bybit kline/turnover ingest first)",
                   err=True)
        store.close()
        raise typer.Exit(1)
    typer.echo(f"bybit-ingest-oi: {len(universe)} symbols ({kind}), "
               f"floor={from_month or '2023-01 (default)'}, workers={workers} "
               f"-> bybit_features @ {settings.operational_dsn.split('@')[-1]}")

    try:
        summary = ingest_bybit_oi(
            store, bybit, universe, start_ms=start_ms, resume=resume, workers=workers, progress=typer.echo)
    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"\nbybit-ingest-oi done: {summary['symbols']} symbols, "
               f"{summary['day_rows']} day-rows, oldest date {oldest_date}")


@app.command("bybit-ingest-klines")
def bybit_ingest_klines(
    from_month: str = typer.Option(
        "", "--from", help="floor for the backward pagination, YYYY-MM (default: the 2020-01 backstop)."),
    symbols: str = typer.Option(
        "all", "--symbols",
        help="'all' (build the universe) | comma-separated | @path-to-file (one symbol per line)."),
    include_delisted: bool = typer.Option(
        False, "--include-delisted/--active-only",
        help="union the active universe with the public.bybit.com/trading delisted dirs "
             "(survivorship-free). Default: active only."),
    resume: bool = typer.Option(
        True, "--resume/--no-resume",
        help="DATE-RESUME each symbol from its last persisted perp day (fetch only new days; every symbol "
             "advances). A symbol with no 'turnover' yet full back-fills from the floor. --no-resume re-fetches "
             "every symbol from the floor (same final state; idempotent upsert)."),
) -> None:
    """Phase B: ingest Bybit daily PERP close ('close') + SPOT close ('spot_close') into the Bybit-namespaced
    TimescaleDB table `bybit_features` (operational DSN — the same store Phase A's funding lands in), so the
    carry re-validation can compute the delta-neutral basis (spot_ret − perp_ret) on Bybit. Resumable/idempotent.
    The universe is built with BybitFunding's helpers (active / ever-listed)."""
    import datetime as _dt

    from fxhnt.adapters.data.bybit_funding import BybitFunding
    from fxhnt.adapters.data.bybit_klines import BybitKlines
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_klines_ingest import ingest_bybit_klines

    settings = get_settings()
    klines = BybitKlines()

    # Resolve the floor (YYYY-MM -> ms) for the backward pagination.
    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

    # Resolve the symbol universe (reuse BybitFunding's active / ever-listed helpers).
    if symbols == "all":
        bybit = BybitFunding()
        universe = bybit.ever_listed_universe() if include_delisted else bybit.active_universe()
        kind = "ever-listed" if include_delisted else "active"
    elif symbols.startswith("@"):
        with open(symbols[1:]) as fh:
            universe = [ln.strip().upper() for ln in fh if ln.strip()]
        kind = "explicit"
    else:
        universe = [s.strip().upper() for s in symbols.split(",") if s.strip()]
        kind = "explicit"
    if not universe:
        typer.echo("error: empty symbol universe", err=True)
        raise typer.Exit(1)
    typer.echo(f"bybit-ingest-klines: {len(universe)} symbols ({kind}), "
               f"floor={from_month or '2020-01 (default)'} "
               f"-> bybit_features @ {settings.operational_dsn.split('@')[-1]}")

    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    try:
        summary = ingest_bybit_klines(
            store, klines, universe, start_ms=start_ms, resume=resume, progress=typer.echo)
    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"\nbybit-ingest-klines done: {summary['symbols']} symbols, "
               f"{summary['perp_rows']} perp-rows + {summary['spot_rows']} spot-rows, "
               f"oldest date {oldest_date}")


@app.command("ingest-deribit-funding")
def ingest_deribit_funding_cmd(
    instruments: str = typer.Option(
        "BTC-PERPETUAL,ETH-PERPETUAL,SOL_USDC-PERPETUAL", "--instruments",
        help="comma-separated Deribit perp instruments to ingest. Default = the majors that overlap Bybit "
             "(BTC/ETH inverse perps + SOL USDC-margined perp). NB: only BTC/ETH are bare '<BASE>-PERPETUAL'; "
             "alts are '<BASE>_USDC-PERPETUAL' (a bare SOL-PERPETUAL 400s). Each is mapped to its Bybit symbol "
             "(BTC-PERPETUAL -> BTCUSDT, SOL_USDC-PERPETUAL -> SOLUSDT) and written as 'deribit_funding'."),
    from_month: str = typer.Option(
        "", "--from", help="floor for the backward pagination, YYYY-MM (default: the 2021-01 backstop)."),
) -> None:
    """Ingest the Deribit perp DAILY funding (the Deribit leg of the cross-venue carry edge) into the
    Bybit-namespaced TimescaleDB table `bybit_features` as feature 'deribit_funding', keyed by the Bybit perp
    symbol (BTC-PERPETUAL -> BTCUSDT, ...). DAILY = SUM of the UTC day's hourly interest_1h rows (matches the
    Bybit daily-sum funding convention so `bybit_funding − deribit_funding` is a like-for-like spread). ONLY
    'deribit_funding' is written (the Bybit funding/close are the other ingests' job); idempotent/resumable.
    The cross-venue-carry-eval reads this `deribit_funding` panel alongside the Bybit `funding` panel. No live
    paper_* tables are touched."""
    import datetime as _dt

    from fxhnt.adapters.data.deribit_funding import DeribitFunding
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.deribit_funding_ingest import ingest_deribit_funding

    settings = get_settings()
    deribit = DeribitFunding()

    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

    insts = [s.strip().upper() for s in instruments.split(",") if s.strip()]
    if not insts:
        typer.echo("error: empty --instruments", err=True)
        raise typer.Exit(1)
    typer.echo(f"ingest-deribit-funding: {insts}, floor={from_month or '2021-01 (default)'} "
               f"-> bybit_features feature='deribit_funding' @ {settings.operational_dsn.split('@')[-1]}")

    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    try:
        summary = ingest_deribit_funding(store, deribit, insts, 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-deribit-funding done: {summary['symbols']} instruments, "
               f"{summary['day_rows']} day-rows, oldest date {oldest_date}")


@app.command("bybit-ingest-worst-basis")
def bybit_ingest_worst_basis_cmd(
    from_month: str = typer.Option(
        "", "--from", help="floor for the backward pagination, YYYY-MM (default: the 2020-01 backstop)."),
    symbols: str = typer.Option(
        "liquid", "--symbols",
        help="'all' (build the universe) | 'liquid' (the tradeable subset via liquid_universe — all we need "
             "worst_basis for) | comma-separated | @path-to-file (one symbol per line)."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) for --symbols liquid."),
    include_delisted: bool = typer.Option(
        False, "--include-delisted/--active-only",
        help="for --symbols all: union the active universe with the delisted dirs (survivorship-free)."),
    workers: int = typer.Option(12, "--workers", help="bounded fetch concurrency (default 12)."),
    resume: bool = typer.Option(
        True, "--resume/--no-resume",
        help="skip symbols already carrying 'worst_basis' in bybit_features (resume a partial/OOM-killed "
             "run). --no-resume re-fetches every symbol (same final state; idempotent upsert)."),
) -> None:
    """Ingest the REAL intraday 1m `worst_basis` series into bybit_features — the squeeze-tail input the
    TRUSTWORTHY carry accounting OVERRIDES the daily-close basis with (same definition as the legacy combined
    book's). Per symbol, the 1m
    perp (linear) + spot klines are aligned per UTC day and reduced to the deepest adverse intraday basis
    (capped at −1.0); only days with BOTH legs produce a value (perp-only days fall back to daily-close).
    Resumable/idempotent/concurrent/bounded-memory. Default `--symbols liquid` (the only subset we trade)."""
    import datetime as _dt

    from fxhnt.adapters.data.bybit_funding import BybitFunding
    from fxhnt.adapters.data.bybit_klines import BybitKlines
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.bybit_worst_basis_ingest import ingest_bybit_worst_basis

    settings = get_settings()
    klines = BybitKlines()

    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

    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    try:
        if symbols == "all":
            bybit = BybitFunding()
            universe = bybit.ever_listed_universe() if include_delisted else bybit.active_universe()
            kind = "ever-listed" if include_delisted else "active"
        elif symbols == "liquid":
            universe = sorted(liquid_universe(store, min_dollar_vol=min_dollar_vol))
            kind = "liquid"
        elif symbols.startswith("@"):
            with open(symbols[1:]) as fh:
                universe = [ln.strip().upper() for ln in fh if ln.strip()]
            kind = "explicit"
        else:
            universe = [s.strip().upper() for s in symbols.split(",") if s.strip()]
            kind = "explicit"
        if not universe:
            typer.echo("error: empty symbol universe", err=True)
            raise typer.Exit(1)
        typer.echo(f"bybit-ingest-worst-basis: {len(universe)} symbols ({kind}), "
                   f"floor={from_month or '2020-01 (default)'} "
                   f"-> bybit_features @ {settings.operational_dsn.split('@')[-1]}")

        summary = ingest_bybit_worst_basis(
            store, klines, universe, start_ms=start_ms, resume=resume, workers=workers,
            progress=typer.echo)
    finally:
        store.close()
    typer.echo(f"\nbybit-ingest-worst-basis done: {summary['symbols']} symbols, "
               f"{summary['days']} worst_basis day-rows, worst {summary['worst']:.4f}")


@app.command("bybit-carry-revalidate")
def bybit_carry_revalidate(
    bybit_cost_bps: float = typer.Option(
        5.5, "--bybit-cost-bps", help="Bybit taker round-trip cost (bps) applied to the Bybit carry curve."),
    from_month: str = typer.Option(
        "", "--from", help="Floor the carry window to YYYY-MM and later (default: all ingested days)."),
    trustworthy: bool = typer.Option(
        False, "--trustworthy/--naive",
        help="De-artifact the naive 13.92-Sharpe: print a 2-row table — Bybit-naive (all symbols, daily-"
             "close basis, flat fee) vs Bybit-trustworthy (liquid universe + REAL 1m worst_basis squeeze-tail "
             "OVERRIDE + per-coin cost); verdict on the trustworthy row. Needs the turnover feature (klines "
             "ingest) and the worst_basis feature (bybit-ingest-worst-basis)."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) for --trustworthy."),
) -> None:
    """Phase C (REPORT, read-only): re-validate the cross-sectional CARRY edge on BYBIT data — the venue we
    actually trade (Phase 0b venue consolidation retired the legacy-venue-baseline comparison this command
    used to also render — Bybit is fxhnt's sole crypto venue). Runs the SAME carry pipeline (XsFundingReplay +
    funding + daily basis) over the Bybit `bybit_features` table (funding + close + spot_close, ingested by
    Phases A/B). Writes NOTHING.

    Default (--naive) prints the Bybit metrics. --trustworthy prints the 2-row de-artifacted table (naive /
    trustworthy) and renders the verdict on the trustworthy row.

    Produces real numbers only once the Phase-A/B Bybit ingests have populated bybit_features; until then
    verdict 'insufficient data'."""
    import datetime as _dt

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_carry_revalidate import (
        carry_metrics_from_store,
        revalidate,
        trustworthy_revalidate,
    )

    settings = get_settings()
    start_day: int | None = None
    if from_month:
        try:
            y, m = (int(x) for x in from_month.split("-"))
            start_day = (_dt.date(y, m, 1) - _dt.date(1970, 1, 1)).days
        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

    bybit_store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    def _fmt(m: dict) -> str:
        span = (f"{m['first_day']}..{m['last_day']}" if m.get("first_day") else "n/a")
        return (f"CAGR={m['cagr']*100:7.1f}%  Sharpe={m['sharpe']:6.2f}  "
                f"maxDD={m['max_dd']*100:7.1f}%  totRet={m['total_return']*100:7.1f}%  "
                f"days={m['days']:5d}  symbols={m['symbols']:4d}  span={span}")

    if trustworthy:
        try:
            report = trustworthy_revalidate(
                bybit_store, bybit_cost_bps=bybit_cost_bps, start_day=start_day,
                min_dollar_vol=min_dollar_vol)
        finally:
            bybit_store.close()
        v = report["verdict"]
        typer.echo("\n=== Bybit carry re-validation — TRUSTWORTHY (READ-ONLY) ===")
        typer.echo(f"Bybit-naive       (all symbols, daily-close, flat {bybit_cost_bps}bp): "
                   f"{_fmt(report['bybit_naive'])}")
        typer.echo(f"Bybit-trustworthy (liquid≥${min_dollar_vol:,.0f} + squeeze-tail + per-coin cost): "
                   f"{_fmt(report['bybit_trustworthy'])}")
        typer.echo(f"\nLiquid universe size: {report['universe_size']} symbols "
                   f"(of {report['bybit_naive']['symbols']} naive)")
        holds = v["holds"]
        tag = "INSUFFICIENT DATA" if holds is None else ("HOLDS ✓" if holds else "DOES NOT HOLD ✗")
        typer.echo(f"VERDICT (trustworthy row): carry edge on Bybit -> {tag}  ({v['reason']})")
        if holds is None:
            typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` + "
                       "`fxhnt bybit-ingest-worst-basis` first; the trustworthy knobs need the turnover "
                       "feature and the real 1m worst_basis squeeze-tail override)")
        else:
            typer.echo(f"  rule: {v['rule']}")
        return

    try:
        if start_day is not None:
            # Honour --from, then re-derive the verdict on the windowed metrics.
            bybit = carry_metrics_from_store(bybit_store, cost_bps=bybit_cost_bps, start_day=start_day)
            if bybit["days"] < 2:
                verdict = {"holds": None, "reason": "insufficient data"}
            else:
                holds = bybit["sharpe"] > 0.0
                verdict = {"holds": holds,
                           "reason": "bybit carry edge is positive" if holds else
                                     "bybit carry edge is NOT positive",
                           "rule": "bybit.sharpe > 0", "bybit_sharpe": bybit["sharpe"]}
            report = {"bybit": bybit, "verdict": verdict}
        else:
            report = revalidate(bybit_store, bybit_cost_bps=bybit_cost_bps)
    finally:
        bybit_store.close()

    b, v = report["bybit"], report["verdict"]
    typer.echo("\n=== Bybit carry re-validation (READ-ONLY) ===")
    typer.echo(f"Bybit (fee {bybit_cost_bps}bp):  {_fmt(b)}")
    holds = v["holds"]
    tag = "INSUFFICIENT DATA" if holds is None else ("HOLDS ✓" if holds else "DOES NOT HOLD ✗")
    typer.echo(f"\nVERDICT: carry edge on Bybit -> {tag}  ({v['reason']})")
    if holds is None:
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first to populate "
                   "bybit_features)")
    else:
        typer.echo(f"  rule: {v['rule']}")
    typer.echo("NOTE: the Bybit data universe is GLOBAL; live Bybit-EU (MiCA) trades a SMALLER subset, so "
               "this is an upper bound on the deployable carry universe.")


@app.command("bybit-edges-eval")
def bybit_edges_eval(
    liquid: bool = typer.Option(
        False, "--liquid/--all",
        help="Restrict the crypto universe to the liquid tradeable subset (median daily turnover >= "
             "--min-dollar-vol), mirroring the carry liquid-subset discipline. Default --all reports the "
             "naive (over-diversified) universe."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) used by --liquid."),
    cost_bps: float = typer.Option(
        5.5, "--cost-bps", help="Per-edge turnover cost (bps) applied to each edge's curve."),
) -> None:
    """READ-ONLY: run ALL fxhnt edges (tstrend / unlock / stablecoin / xsfunding) on the BYBIT warehouse
    (`bybit_features`) and print a CAGR / Sharpe / maxDD / symbols table per edge, plus an equal-weight
    combined book. Writes NOTHING — the live paper book tables (paper_nav/paper_sleeve_ret/...) are
    untouched.

    Each edge reuses its OWN validated return engine driven off the Bybit store. tstrend + xsfunding are
    fully computable from the warehouse. unlock needs the DefiLlama unlock calendar (loaded live from
    `unlock_calendar.json`); stablecoin needs an EXTERNAL stablecoin spot panel — Phase 0b venue
    consolidation retired the legacy-venue spot fetch this command used to inject, so `stablecoin_rotation`
    now reports "N/A (reason)" here (use `fxhnt bybit-stablecoin-eval` for the Bybit-native stablecoin
    peg-reversion probe on `bybit_features`' own spot_close). With --liquid the crypto edges are restricted
    to the liquid subset so the over-diversification artifact (the naive all-universe number) is visible
    alongside it."""
    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_edges_eval import evaluate_all_edges
    from fxhnt.application.bybit_liquidity import liquid_universe

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    # Live external loader (run ONLY when this command is invoked — never in tests): the unlock calendar
    # (DefiLlama emissions cached at `unlock_calendar.json`).
    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        # DB-first (operational `unlock_events`, networked — no surfer-data PVC), JSON fallback for
        # back-compat. Absent calendar → unlock reports N/A, never a crash.
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock reports N/A, not a crash
            log.warning("bybit-edges-eval: could not load unlock calendar: %s", exc)
            return []

    universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if liquid else None

    try:
        report = evaluate_all_edges(
            store, universe=universe, cost_bps=cost_bps,
            load_unlock_events=_load_unlock_events,
            stablecoin_spot_panel=None)
    finally:
        store.close()

    scope = (f"LIQUID subset (median turnover >= ${min_dollar_vol:,.0f}, {len(universe)} symbols)"
             if liquid else "ALL symbols (naive / over-diversified)")
    typer.echo(f"\n=== Bybit multi-edge evaluation — READ-ONLY — {scope} ===")
    header = f"{'edge':<20} {'CAGR':>8} {'Sharpe':>8} {'maxDD':>8} {'symbols':>8}  detail"
    typer.echo(header)
    typer.echo("-" * len(header))
    for edge in ("tstrend", "unlock", "stablecoin_rotation", "xsfunding", "combined"):
        m = report[edge]
        if m.get("available"):
            typer.echo(f"{edge:<20} {m['cagr']*100:7.1f}% {m['sharpe']:8.2f} "
                       f"{m['max_dd']*100:7.1f}% {m['symbols']:8d}  days={m['days']}")
        else:
            typer.echo(f"{edge:<20} {'N/A':>8} {'N/A':>8} {'N/A':>8} {'N/A':>8}  ({m.get('reason','')})")
    typer.echo("\nNOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched. 'combined' = equal-weight "
               "blend of the AVAILABLE per-edge summary metrics (not a re-simulated vol-targeted book). "
               "Compare --all vs --liquid: the all-universe numbers are over-diversification artifacts.")


@app.command("bybit-stablecoin-eval")
def bybit_stablecoin_eval(
    cost_bps: float = typer.Option(
        5.5, "--cost-bps", help="Per-turnover cost (bps) applied to the peg-reversion curve."),
) -> None:
    """READ-ONLY Phase-1: does stablecoin PEG-REVERSION transfer to BYBIT's OWN near-$1 stablecoin spot pairs
    (USDCUSDT / USDEUSDT / TUSDT), using `spot_close` ALREADY in bybit_features?

    Auto-detects the near-$1 pairs (median spot_close ∈ [0.95, 1.05] — excludes gold PAXG/XAUT, dead pegs,
    and normal coins), runs the LIVE `StableReversionRunner` UNCHANGED (same short-rich edge) on that panel,
    and prints the detected pairs + CAGR/Sharpe/maxDD. Writes NOTHING — the live paper_* tables are
    untouched. Validate-cheap BEFORE ingesting USDC-margined perps (Phase 2)."""
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_stablecoin_eval import stablecoin_reversion_metrics_from_store

    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    try:
        m = stablecoin_reversion_metrics_from_store(store, cost_bps=cost_bps)
    finally:
        store.close()

    pairs = m.get("pairs", [])
    typer.echo("\n=== Bybit stablecoin peg-reversion (Phase-1) — READ-ONLY ===")
    typer.echo(f"detected near-$1 stablecoin spot pairs ({len(pairs)}): "
               f"{', '.join(pairs) if pairs else '(none)'}")
    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']}  "
                   f"span={m['first_day']}..{m['last_day']}")
    else:
        typer.echo(f"  N/A — {m.get('reason', '')}")

    typer.echo(
        "\nCAVEATS:\n"
        "  * Daily-close spot captures only SUSTAINED dislocations. Sharp INTRADAY venue-dislocations\n"
        "    (e.g. USDE traded $0.65 on 2025-10-10 and reverted within hours) are INVISIBLE at the daily\n"
        "    close → this is a CONSERVATIVE LOWER BOUND. Intraday reversion (needs 1m spot) is Phase-2+.\n"
        f"  * Only {len(pairs)} near-$1 pair(s) — that is THIN. Phase-1 tells us whether to invest in\n"
        "    Phase-2 breadth (USDC-margined perps), NOT a deployable book on its own.\n"
        "  * READ-ONLY: writes NOTHING — paper_nav/paper_sleeve_ret untouched.")




def _print_xvenue_carry_verify(rep: dict, *, cost_bps: float) -> None:
    """Pretty-print the adversarial cross-venue carry verification report (both modes)."""
    typer.echo("\n=== Bybit-vs-Deribit cross-venue funding CARRY — ADVERSARIAL VERIFICATION (READ-ONLY) ===")
    typer.echo("signal: bybit_funding − deribit_funding per coin/day  |  delta-neutral SHORT-Bybit/LONG-Deribit "
               "(price risk ~cancels) → harvest the funding SPREAD")
    if not rep.get("available"):
        typer.echo(f"  N/A — {rep.get('reason', '')}  (n-coins={rep.get('n_coins', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt ingest-deribit-funding` first)")
        return
    n = rep["n_coins"]
    typer.echo(f"  n-coins (tradeable overlap breadth): {n}"
               + ("   *** LOW BREADTH: a 2-coin Sharpe is NOISY ***" if n <= 2 else ""))

    typer.echo("\n-- NET-OF-COST, BOTH MODES (Sharpe / CAGR / maxDD / total / days) --")
    for mode in ("static", "dynamic"):
        typer.echo(f"  [{mode.upper()}]"
                   + ("  (always short-Bybit/long-Deribit — harvest the persistent level, low turnover)"
                      if mode == "static"
                      else "  (position by sign of the spread — per-flip turnover)"))
        for bps in rep["cost_grid"]:
            m = rep["net_cost"][mode][bps]
            typer.echo(f"    cost={bps:5.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(f"    avg daily turnover={rep['avg_turnover'][mode]:.4f}  "
                   f"top1-coin share={rep['top1_share'][mode]*100:.0f}%")

    typer.echo("\n-- PER-YEAR (net @ headline cost), BOTH MODES --")
    for mode in ("static", "dynamic"):
        typer.echo(f"  [{mode.upper()}]")
        for yr, y in sorted(rep["per_year"][mode].items()):
            typer.echo(f"    {yr}: Sharpe={y['sharpe']:6.2f}  sum_ret={y['sum_ret']*100:7.2f}%  days={y['days']}")

    typer.echo("\n-- CORRELATION vs the 4 existing edges (net @ headline cost) --")
    for mode in ("static", "dynamic"):
        parts = []
        for edge in ("tstrend", "unlock", "xsfunding", "positioning"):
            c = rep["corr_edges"][mode].get(edge)
            parts.append(f"{edge}={'N/A' if c is None else f'{c:+.3f}'}")
        typer.echo(f"  [{mode.upper()}]  " + "  ".join(parts))
    typer.echo("\nNOTE: writes NOTHING. Memory-bounded (only the overlap coins are READ). The funding-spread is "
               "the modelled return; a small perp-basis-divergence residual is omitted (no Deribit price panel).")


def _print_xvenue_carry_walk_forward(rep: dict, *, cost_bps: float) -> None:
    """Pretty-print the OOS / walk-forward deploy-gate report for the cross-venue carry edge."""
    typer.echo("\n=== Bybit-vs-Deribit cross-venue carry — OOS / WALK-FORWARD (DEPLOY GATE, READ-ONLY) ===")
    if not rep.get("available"):
        typer.echo(f"  N/A — {rep.get('reason', '')}  (n-coins={rep.get('n_coins', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt ingest-deribit-funding` first)")
        return
    n = rep["n_coins"]
    typer.echo(f"  n-coins={n}   train-chosen mode={rep['mode'].upper()}   "
               f"full-sample Sharpe (net @ {cost_bps}bps)={rep['full_sharpe']:.2f}")
    if n <= 2:
        typer.echo(f"  *** LOW BREADTH: {rep['verdict'].get('low_breadth_note', '')} ***")

    rw = rep["rolling_windows"]
    typer.echo(f"\n-- ROLLING WINDOWS ({rep['window_days']}d, non-overlapping) --")
    typer.echo(f"  n_windows={rw['n_windows']}  fraction_positive={rw['fraction_positive']*100:.0f}%")
    for w in rw["windows"]:
        typer.echo(f"    {w['start']}..{w['end']}  Sharpe={w['sharpe']:6.2f}  "
                   f"total={w['total_return']*100:7.2f}%  days={w['days']}")

    th = rep["train_holdout"]
    typer.echo("\n-- TRAIN/TEST MODE HOLDOUT --")
    typer.echo(f"  train_mode={th['train_mode']}  train_sharpe={th['train_sharpe']:.2f}  "
               f"TEST(OOS)_sharpe={th['test_sharpe']:.2f}  split_at={th.get('split_at')}  "
               f"(train {th['train_days']}d / test {th['test_days']}d)")

    typer.echo("\n-- CORRELATION vs the 4 existing edges (full-sample, train-mode) --")
    parts = []
    for edge in ("tstrend", "unlock", "xsfunding", "positioning"):
        c = rep["corr_edges"].get(edge)
        parts.append(f"{edge}={'N/A' if c is None else f'{c:+.3f}'}")
    typer.echo("  " + "  ".join(parts))

    la = rep["look_ahead"]
    typer.echo(f"\n-- LOOK-AHEAD AUDIT --  causal={la['causal']}  leak_days={la['leak_days']}/"
               f"{la['n_days_compared']}")

    v = rep["verdict"]
    typer.echo("\n-- VERDICT --")
    for name, ok in v["checks"].items():
        typer.echo(f"    [{'PASS' if ok else 'FAIL'}] {name}")
    typer.echo(f"  => {'DEPLOYABLE' if v['deployable'] else 'NOT DEPLOYABLE'}"
               + (f"  ({v['low_breadth_note']})" if v.get("low_breadth_note") else ""))
    typer.echo("\nNOTE: writes NOTHING. A clean 'too thin / arbed / no edge' is a VALID outcome (low-breadth prior).")


def _print_xvenue_carry_gated_walk_forward(rep: dict, *, cost_bps: float) -> None:
    """Pretty-print the GATED-STATIC OOS / walk-forward deploy-gate report (disciplined coin selection)."""
    typer.echo("\n=== Bybit-vs-Deribit cross-venue carry — GATED-STATIC OOS / WALK-FORWARD (READ-ONLY) ===")
    typer.echo("  gate: include a coin ONLY when its CAUSAL trailing-mean spread over [t-L, t-1] >= threshold "
               "(a smarter STATIC — picks WHICH coins, never flips long-Bybit)")
    if not rep.get("available"):
        typer.echo(f"  N/A — {rep.get('reason', '')}  (n-coins={rep.get('n_coins', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt ingest-deribit-funding` first)")
        return
    n = rep["n_coins"]
    g = rep["grid"]
    ch = rep["chosen"]
    typer.echo(f"  n-coins={n}   grid: L∈{g['L']} × threshold∈{g['threshold']}  (SMALL by design — a huge "
               "grid = overfitting by search)")
    typer.echo(f"  TRAIN-chosen: L={ch['lookback']}  threshold={ch['threshold']:.2e}  "
               f"train_sharpe={ch['train_sharpe']:.2f}   split_at={rep['split_at']}  "
               f"(train {rep['train_days']}d / test {rep['test_days']}d)")
    if n <= 2:
        typer.echo(f"  *** LOW BREADTH: {rep['verdict'].get('low_breadth_note', '')} ***")

    cmp = rep["comparison"]
    bte = cmp["btceth_static_oos_sharpe"]
    typer.echo("\n-- HONEST OOS COMPARISON (same TEST split) --")
    typer.echo(f"  gated-OOS Sharpe        = {cmp['gated_oos_sharpe']:6.2f}")
    typer.echo(f"  plain-static-OOS Sharpe = {cmp['plain_static_oos_sharpe']:6.2f}   (all-coins static, the "
               "0.72-class dilution)")
    typer.echo(f"  BTC/ETH-static-OOS      = {('N/A' if bte is None else f'{bte:6.2f}')}   (the 2-coin clean "
               "edge, the 5.65-class)")
    typer.echo(f"  full-sample gated Sharpe (net @ {cost_bps}bps) = {rep['full_sharpe']:.2f}")

    typer.echo("\n-- SELECTED COINS over the TEST window (inclusion fraction) --")
    sel = rep["selected_coins"]
    if sel:
        for coin, frac in sorted(sel.items(), key=lambda kv: -kv[1]):
            typer.echo(f"    {coin:14s} included {frac*100:5.1f}% of test days")
        typer.echo(f"  retail-majors selected: {', '.join(rep['selected_retail_majors']) or 'none'}")
    else:
        typer.echo("    (none selected on TEST — the gate excluded every coin)")

    rw = rep["rolling_windows"]
    typer.echo(f"\n-- ROLLING WINDOWS ({rep['window_days']}d, non-overlapping) --  "
               f"n_windows={rw['n_windows']}  fraction_positive={rw['fraction_positive']*100:.0f}%")

    typer.echo("\n-- CORRELATION vs the 4 existing edges (full-sample, gated) --")
    parts = []
    for edge in ("tstrend", "unlock", "xsfunding", "positioning"):
        c = rep["corr_edges"].get(edge)
        parts.append(f"{edge}={'N/A' if c is None else f'{c:+.3f}'}")
    typer.echo("  " + "  ".join(parts))

    la = rep["look_ahead"]
    typer.echo(f"\n-- LOOK-AHEAD AUDIT --  causal={la['causal']}  leak_days={la['leak_days']}/"
               f"{la['n_days_compared']}  (leaked gate would differ on {la.get('leaked_differs_on_days', 0)} "
               "days → audit is non-vacuous)")

    v = rep["verdict"]
    typer.echo("\n-- VERDICT --")
    for name, ok in v["checks"].items():
        typer.echo(f"    [{'PASS' if ok else 'FAIL'}] {name}")
    typer.echo(f"  => {'DEPLOYABLE (genuinely broader edge)' if v['deployable'] else 'NOT DEPLOYABLE'}")
    typer.echo(f"  INTERPRETATION: {v['interpretation']}")
    typer.echo("\nNOTE: writes NOTHING. A 'decays to the plain-static (~0.72) → it's BTC/ETH-only' outcome is a "
               "VALID, valuable honest result (deploy the 2-coin static, not a wider basket).")


@app.command("cross-venue-carry-eval")
def cross_venue_carry_eval_cmd(
    mode: str = typer.Option(
        "static", "--mode",
        help="static (DEFAULT: always SHORT-Bybit/LONG-Deribit — harvest the persistent positive funding "
             "level, low turnover), dynamic (position by the SIGN of the spread — per-flip turnover), or gated "
             "(the disciplined causal persistence-gated STATIC — include a coin ONLY when its trailing-mean "
             "spread over [t-L, t-1] clears a threshold; with --walk-forward (L, threshold) are picked on "
             "TRAIN and scored OOS). The P&L is RE-RUN per mode so the per-turnover cost is correctly "
             "subtracted."),
    cost_bps: float = typer.Option(
        11.0, "--cost-bps",
        help="Per-turnover cost (bps). Default 11 = the realistic perp round-trip across BOTH venues "
             "(~5.5bp each). Static charges entry only (low turnover); dynamic charges each flip."),
    liquid: bool = typer.Option(
        False, "--liquid/--all",
        help="Restrict to the liquid Bybit subset on top of the Bybit∩Deribit overlap. Default --all uses "
             "every overlap coin (the overlap is already only the majors)."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) used by --liquid."),
    verify: bool = typer.Option(
        False, "--verify",
        help="ADVERSARIAL verification: BOTH modes' net-of-cost metrics across a cost grid (11/22/44 bps), "
             "per-coin attribution, per-year Sharpe, turnover, and the correlation with EACH of the 4 "
             "existing edges (tstrend/unlock/xsfunding/positioning) — to judge a robust additive carry vs "
             "an artifact."),
    walk_forward: bool = typer.Option(
        False, "--walk-forward",
        help="OOS / WALK-FORWARD validation (the DEPLOY GATE): rolling non-overlapping windows + "
             "fraction-positive, a TRAIN/TEST MODE split (pick the mode on TRAIN, score it OOS on the "
             "held-out TEST), corr-vs-4-edges, a look-ahead audit, and a PASS/FAIL deploy verdict."),
    window_days: int = typer.Option(
        180, "--window-days",
        help="Width (days) of each rolling non-overlapping OOS window in --walk-forward."),
) -> None:
    """READ-ONLY: evaluate the NEW Bybit-vs-Deribit CROSS-VENUE FUNDING-CARRY edge (delta-neutral level harvest).

    Thesis: Bybit perp funding runs PERSISTENTLY higher than Deribit on the liquid majors (retail-heavy Bybit
    pays more). A delta-neutral SHORT-Bybit / LONG-Deribit position on the SAME coin harvests the funding
    SPREAD (bybit − deribit) with ~cancelled price risk → the return ≈ the spread minus cost. Distinct from
    the failed cross-VENUE cross-sectional dispersion probe (illiquid alts = fee-trap) and from xsfunding
    (cross-section WITHIN Bybit) — this is a BETWEEN-venue LEVEL carry on the LIQUID MAJORS.

    Prints the edge metrics (CAGR/Sharpe/maxDD) + the n-overlap-coins breadth (intrinsically LOW: 2-3 coins —
    a 2-coin Sharpe is flagged as noisy). With --verify it stress-tests both modes; with --walk-forward it runs
    the OOS deploy gate. Memory-bounded (only the overlap coins are READ); writes NOTHING. Real numbers appear
    only once BOTH the Bybit funding and the Deribit funding ingests have populated bybit_features; before that
    the overlap is empty → N/A. A clean 'too thin / arbed / no edge' verdict is a VALID outcome."""
    if mode not in ("static", "dynamic", "gated"):
        raise typer.BadParameter("must be 'static', 'dynamic', or 'gated'", param_hint="--mode")

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.cross_venue_carry_eval import (
        cross_venue_carry_metrics_from_store,
        verify_cross_venue_carry,
        walk_forward_cross_venue_carry,
        walk_forward_cross_venue_carry_gated,
    )

    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if liquid else None

    if walk_forward:
        try:
            if mode == "gated":
                rep = walk_forward_cross_venue_carry_gated(
                    store, universe=universe, cost_bps=cost_bps, window_days=window_days)
            else:
                rep = walk_forward_cross_venue_carry(
                    store, universe=universe, cost_bps=cost_bps, window_days=window_days)
        finally:
            store.close()
        if mode == "gated":
            _print_xvenue_carry_gated_walk_forward(rep, cost_bps=cost_bps)
        else:
            _print_xvenue_carry_walk_forward(rep, cost_bps=cost_bps)
        return

    if verify:
        try:
            rep = verify_cross_venue_carry(store, universe=universe, cost_bps=cost_bps)
        finally:
            store.close()
        _print_xvenue_carry_verify(rep, cost_bps=cost_bps)
        return

    try:
        m = cross_venue_carry_metrics_from_store(store, universe=universe, cost_bps=cost_bps, mode=mode)
    finally:
        store.close()

    typer.echo("\n=== Bybit-vs-Deribit cross-venue funding CARRY — READ-ONLY ===")
    typer.echo(f"mode: {mode.upper()}  "
               + ("(always SHORT-Bybit/LONG-Deribit — harvest the persistent positive spread, low turnover)"
                  if mode == "static"
                  else "(position by the SIGN of the spread — per-flip turnover)"))
    typer.echo("signal: bybit_funding − deribit_funding per coin/day  |  delta-neutral (price risk ~cancels) "
               "→ harvest the funding SPREAD")
    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']}  "
                   f"span={m['first_day']}..{m['last_day']}")
        n = m["n_coins"]
        typer.echo(f"  n-overlap-coins (tradeable breadth): {n}"
                   + ("   *** LOW BREADTH: a 2-coin Sharpe is NOISY ***" if n <= 2 else ""))
    else:
        typer.echo(f"  N/A — {m.get('reason', '')}  (n-coins={m.get('n_coins', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt ingest-deribit-funding` first)")
    typer.echo("\nNOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched. Memory-bounded: only the overlap "
               "(and, with --liquid, liquid) coins are READ. The funding-spread is the modelled return; a small "
               "perp-basis-divergence residual is omitted (no Deribit price panel).")


def _print_positioning_verify(rep: dict, *, scope: str, direction_note: str, cost_bps: float) -> None:
    """Pretty-print the adversarial positioning verification report (both directions)."""
    typer.echo("\n=== Bybit positioning (long/short account-ratio) — ADVERSARIAL VERIFICATION (READ-ONLY) ===")
    typer.echo(f"signal: Bybit long/short ACCOUNT ratio (retail positioning)  |  scope: {scope}  |  "
               "bet: PRICE (close-to-close), cross-sectional, market-neutral")
    typer.echo(f"  {direction_note}")
    if not rep.get("available"):
        typer.echo(f"  N/A — {rep.get('reason', '')}  (n-coins={rep.get('n_coins', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-long-short` + `fxhnt bybit-ingest-klines` first)")
        return
    typer.echo(f"  n-coins (tradeable breadth): {rep['n_coins']}")

    typer.echo("\n-- NET-OF-COST, BOTH DIRECTIONS (Sharpe / CAGR / maxDD / total / days) --")
    for d in ("contrarian", "momentum"):
        typer.echo(f"  [{d.upper()}]")
        for bps in rep["cost_grid"]:
            m = rep["net_cost"][d][bps]
            typer.echo(f"    cost={bps:5.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-- PER-COIN ATTRIBUTION (gross; is the edge broad or 1-2-coin fragile?) --")
    for d in ("contrarian", "momentum"):
        attr = rep["per_coin"][d]
        top = sorted(attr.items(), key=lambda kv: abs(kv[1]), reverse=True)[:5]
        top_str = ", ".join(f"{c}={v*100:+.2f}%" for c, v in top)
        typer.echo(f"  [{d.upper()}] top-5 |share| of gross = {rep['top5_share'][d]*100:5.1f}%  | "
                   f"top: {top_str}")

    typer.echo("\n-- PER-YEAR Sharpe (net @ headline cost; all-from-one-year => overfit) --")
    for d in ("contrarian", "momentum"):
        years = rep["per_year"][d]
        ys = "  ".join(f"{yr}:Sh={y['sharpe']:+5.2f}(n={y['days']})" for yr, y in sorted(years.items()))
        typer.echo(f"  [{d.upper()}] {ys}")

    typer.echo("\n-- AVERAGE DAILY TURNOVER (Σ|Δw|; high => costs bite + lower capacity) --")
    for d in ("contrarian", "momentum"):
        typer.echo(f"  [{d.upper()}] {rep['avg_turnover'][d]:.3f}")

    typer.echo(f"\n-- CORRELATION with EACH existing edge (net @ {cost_bps:.1f}bps; ~0 => additive) --")
    for d in ("contrarian", "momentum"):
        corrs = rep["corr_edges"][d]
        parts = []
        for edge in ("tstrend", "unlock", "xsfunding"):
            c = corrs.get(edge)
            parts.append(f"{edge}={'N/A' if c is None else f'{c:+.3f}'}")
        typer.echo(f"  [{d.upper()}] " + "  ".join(parts))

    typer.echo("\nVERDICT GUIDE: the winning direction is a ROBUST uncorrelated edge only if it stays Sharpe>0 "
               "at 22bps, is NOT driven by 1-2 coins (top-5 share well below ~100% on a broad universe), is "
               "positive in MORE THAN ONE year, AND is uncorrelated with ALL THREE existing edges (tstrend / "
               "unlock / xsfunding). Otherwise it is an overfit / low-cost / single-regime artifact.")
    typer.echo("\nNOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched. Memory-bounded.")


def _print_positioning_walk_forward(rep: dict, *, scope: str, cost_bps: float) -> None:
    """Pretty-print the OOS / walk-forward validation report (the deploy gate)."""
    typer.echo("\n=== Bybit positioning — OOS / WALK-FORWARD VALIDATION (DEPLOY GATE, READ-ONLY) ===")
    typer.echo(f"signal: Bybit long/short ACCOUNT ratio (retail positioning)  |  scope: {scope}  |  "
               f"cost: {cost_bps:.1f}bps")
    if not rep.get("available"):
        typer.echo(f"  N/A — {rep.get('reason', '')}  (n-coins={rep.get('n_coins', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-long-short` + `fxhnt bybit-ingest-klines` first)")
        return
    typer.echo(f"  n-coins (tradeable breadth): {rep['n_coins']}  |  direction (TRAIN-chosen): "
               f"{rep['direction'].upper()}  |  full-sample Sharpe: {rep['full_sharpe']:.2f}")

    rw = rep["rolling_windows"]
    typer.echo(f"\n-- ROLLING NON-OVERLAPPING WINDOWS ({rep['window_days']}d; a real edge is positive in "
               "MOST windows) --")
    typer.echo(f"  windows={rw['n_windows']}  fraction-positive={rw['fraction_positive']*100:5.1f}%")
    for w in rw["windows"]:
        typer.echo(f"    {w['start']}..{w['end']}  Sharpe={w['sharpe']:+6.2f}  "
                   f"total={w['total_return']*100:+7.1f}%  days={w['days']}")

    th = rep["train_holdout"]
    typer.echo("\n-- TRAIN / HOLDOUT DIRECTION (pick the sign on TRAIN, score it OOS on never-seen TEST) --")
    if th.get("available"):
        typer.echo(f"  TRAIN-chosen direction = {th['train_direction'].upper()}  |  "
                   f"split@ {th.get('split_at')}  (train {th['train_days']}d / test {th['test_days']}d)")
        typer.echo(f"  TRAIN Sharpe = {th['train_sharpe']:+6.2f}   ->   "
                   f"TEST (OOS) Sharpe = {th['test_sharpe']:+6.2f}   "
                   f"{'[GENERALISES]' if th['test_sharpe'] > 0 else '[OVERFIT — OOS NEGATIVE]'}")
    else:
        typer.echo("  N/A — too few days to split")

    dt = rep["drop_top_n"]
    typer.echo("\n-- DROP-TOP-N CONTRIBUTORS (ranked CAUSALLY on TRAIN; collapse => concentration-driven) --")
    typer.echo(f"  full Sharpe = {dt['full_sharpe']:+6.2f}")
    for n in (5, 10):
        d = dt[f"drop{n}"]
        typer.echo(f"  drop-top-{n:<2d} Sharpe = {d['sharpe']:+6.2f}  "
                   f"total={d['total_return']*100:+7.1f}%  (dropped {d['n_dropped']}: "
                   f"{', '.join(d['dropped'][:6])}{'…' if len(d['dropped']) > 6 else ''})")

    typer.echo("\n-- LIQUIDITY-THRESHOLD SWEEP (min daily $vol; does the edge need a specific universe?) --")
    for thr, m in sorted(rep["liquidity_sweep"].items()):
        typer.echo(f"  min_dollar_vol >= ${thr:>12,.0f}  Sharpe={m['sharpe']:+6.2f}  "
                   f"total={m['total_return']*100:+7.1f}%  n-coins={m['n_coins']}")

    typer.echo("\n-- SIGNAL-VARIANT SWEEP (construction robustness; one-only => construction artifact) --")
    for name, m in rep["signal_variants"].items():
        typer.echo(f"  {name:<8s} Sharpe={m['sharpe']:+6.2f}  total={m['total_return']*100:+7.1f}%  "
                   f"days={m['days']}")

    la = rep["look_ahead"]
    typer.echo("\n-- LOOK-AHEAD AUDIT (signal at d -> d->d+1 return only; no same-day/future leak) --")
    typer.echo(f"  causal={la['causal']}  leak-days={la['leak_days']}  "
               f"(compared {la['n_days_compared']} days)")

    v = rep["verdict"]
    typer.echo("\n=== VERDICT ===")
    labels = {
        "rolling_windows_positive": ">70% of rolling windows positive",
        "oos_direction_positive": "TEST/OOS Sharpe > 0 with the TRAIN-chosen direction",
        "survives_drop_top5": "survives drop-top-5 (Sharpe still clearly positive)",
        "liquidity_robust": "positive across the liquidity sweep",
        "look_ahead_clean": "look-ahead audit clean (strictly causal)",
    }
    for key, label in labels.items():
        ok = v["checks"].get(key, False)
        typer.echo(f"  [{'PASS' if ok else 'FAIL'}] {label}")
    typer.echo(f"\n  >>> {'PASS — DEPLOYABLE' if v['deployable'] else 'FAIL — DO NOT DEPLOY'} <<<")
    typer.echo("\nNOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched. Memory-bounded.")


def _print_positioning_metalabel(rep: dict, *, scope: str, shuffle_label: bool) -> None:
    """Pretty-print the de Prado meta-labeling experiment (purged-CV comparison + skeptical verdict)."""
    typer.echo("\n=== Bybit positioning — META-LABELING EXPERIMENT (de Prado, PURGED CV, READ-ONLY) ===")
    typer.echo("primary (SIDE) = contrarian positioning fade  |  meta (SIZE/FILTER) = LogisticRegression "
               "P(profitable)")
    typer.echo(f"scope: {scope}" + ("   |   SHUFFLE-LABEL CONTROL (leakage tripwire: expect NO LIFT)"
                                     if shuffle_label else ""))
    if not rep.get("available"):
        typer.echo(f"  N/A — {rep.get('reason', '')}  (n-rows={rep.get('n_rows', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-long-short` + `fxhnt bybit-ingest-klines` first)")
        return

    typer.echo(f"  coin-day rows={rep['n_rows']}  OOS rows={rep['n_oos_rows']}  "
               f"purged {rep['n_splits']}-fold (embargo {rep['embargo_days']}d, "
               f"label_horizon {rep['label_horizon']}d)  threshold(P>)={rep['threshold']:.2f}")

    typer.echo("\n-- FEW META-FEATURES (anti-overfit) + averaged FEATURE IMPORTANCE (|coef|) --")
    for name in rep["feature_names"]:
        typer.echo(f"    {name:<18s} importance={rep['feature_importances'][name]:+.4f}")

    o = rep["oos"]
    typer.echo("\n-- PURGED OUT-OF-SAMPLE: RAW (side-only) vs META-LABELED (filter P>thr, size ∝ P) --")
    typer.echo(f"    RAW   Sharpe={o['raw_sharpe']:+6.2f}  CAGR={o['raw_cagr']*100:+7.1f}%  "
               f"maxDD={o['raw_max_dd']*100:+6.1f}%  days={o['raw_days']}")
    typer.echo(f"    META  Sharpe={o['meta_sharpe']:+6.2f}  CAGR={o['meta_cagr']*100:+7.1f}%  "
               f"maxDD={o['meta_max_dd']*100:+6.1f}%  days={o['meta_days']}")
    typer.echo(f"    Sharpe LIFT (meta-raw) = {o['sharpe_lift']:+6.2f}   "
               f"(adopt only if > {rep['verdict']['add_margin']:.2f})")
    typer.echo(f"    OOS precision (kept→profitable) = {o['meta_precision']*100:5.1f}%  "
               f"vs base-rate {o['base_rate']*100:5.1f}%  (lift {o['precision_lift']*100:+5.1f}pp)  |  "
               f"trades kept = {o['frac_trades_kept']*100:5.1f}%")

    v = rep["verdict"]
    typer.echo("\n=== VERDICT (PURGED-OOS only — train metrics are reference, never the verdict) ===")
    typer.echo(f"  >>> {v['label']} <<<")
    if shuffle_label and not v["adds"]:
        typer.echo("  [OK] shuffled-label control shows NO lift → pipeline is leakage-free.")
    elif shuffle_label and v["adds"]:
        typer.echo("  [LEAK] shuffled-label control LIFTED — pipeline LEAKS; do NOT trust the real run.")
    typer.echo("\nNOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched. Memory-bounded.")


@app.command("bybit-positioning-eval")
def bybit_positioning_eval(
    liquid: bool = typer.Option(
        False, "--liquid/--all",
        help="Restrict the positioning universe to the liquid Bybit subset (median daily turnover >= "
             "--min-dollar-vol). Default --all uses every coin with a long_ratio panel (the naive breadth)."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) used by --liquid."),
    direction: str = typer.Option(
        "contrarian", "--direction",
        help="contrarian (DEFAULT: SHORT over-long / LONG over-short — fade retail) or momentum (LONG "
             "over-long / SHORT over-short — informed-flow continuation). Both are RE-RUN so the cost is "
             "correctly subtracted in EACH."),
    cost_bps: float = typer.Option(
        5.5, "--cost-bps", help="Per-turnover cost (bps) applied to the positioning curve."),
    verify: bool = typer.Option(
        False, "--verify",
        help="ADVERSARIAL verification mode: print BOTH directions' net-of-cost metrics across a cost grid "
             "(5.5/11/22 bps), per-coin attribution (top-5 share), per-year Sharpe, average daily turnover, "
             "AND the correlation with EACH existing edge (tstrend/unlock/xsfunding) — to judge whether the "
             "winning direction is a robust additive edge or an artifact."),
    walk_forward: bool = typer.Option(
        False, "--walk-forward",
        help="OOS / WALK-FORWARD validation (the DEPLOY GATE): rolling non-overlapping windows + "
             "fraction-positive, a TRAIN/TEST direction split (pick the sign on TRAIN, score it OOS on the "
             "held-out TEST), drop-top-N robustness, a liquidity-threshold sweep, a signal-variant sweep, a "
             "look-ahead audit, and a PASS/FAIL deploy verdict. Use with --liquid on live data."),
    window_days: int = typer.Option(
        180, "--window-days",
        help="Width (days) of each rolling non-overlapping OOS window in --walk-forward."),
) -> None:
    """READ-ONLY: evaluate the NEW Bybit POSITIONING edge off the long/short ACCOUNT ratio (retail positioning).

    Signal = the fraction of accounts LONG (buyRatio) per coin/day. CONTRARIAN (default) SHORTS coins where
    retail is over-LONG (high buyRatio) and LONGS the over-SHORT — betting on PRICE (close-to-close) reversion,
    cross-sectional + market-neutral on LIQUID coins. MOMENTUM is the sign-flip (positioning = informed flow).
    Both directions are tested, like the cross-venue funding edge.

    Prints the edge metrics (CAGR/Sharpe/maxDD) + the n-coins breadth. With --verify it additionally stress-
    tests both directions across a cost grid, attributes per-coin, slices per-year, reports turnover, and shows
    the correlation with each existing edge so additivity is visible.

    Memory-bounded: with --liquid only the liquid coins are READ. Writes NOTHING — the live paper_* tables are
    untouched. Real numbers appear only once the long-short + kline ingests have populated bybit_features."""
    if direction not in ("contrarian", "momentum"):
        raise typer.BadParameter("must be 'contrarian' or 'momentum'", param_hint="--direction")

    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.bybit_positioning_eval import (
        positioning_metrics_from_store,
        verify_positioning_edge,
        walk_forward_positioning,
    )

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if liquid else None
    scope = (f"LIQUID subset (median turnover >= ${min_dollar_vol:,.0f})" if liquid
             else "ALL coins with a long_ratio panel (naive breadth)")
    direction_note = ("direction: " + (
        "CONTRARIAN (SHORT over-long / LONG over-short — fade retail)" if direction == "contrarian"
        else "MOMENTUM (LONG over-long / SHORT over-short — informed-flow continuation)"))

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock corr simply N/A, not a crash
            log.warning("bybit-positioning-eval: could not load unlock calendar: %s", exc)
            return []

    if walk_forward:
        try:
            rep = walk_forward_positioning(
                store, universe=universe, cost_bps=cost_bps, window_days=window_days)
        finally:
            store.close()
        _print_positioning_walk_forward(rep, scope=scope, cost_bps=cost_bps)
        return

    if verify:
        try:
            rep = verify_positioning_edge(
                store, universe=universe, cost_bps=cost_bps, unlock_events=_load_unlock_events())
        finally:
            store.close()
        _print_positioning_verify(rep, scope=scope, direction_note=direction_note, cost_bps=cost_bps)
        return

    try:
        m = positioning_metrics_from_store(
            store, universe=universe, cost_bps=cost_bps, direction=direction)
    finally:
        store.close()

    typer.echo("\n=== Bybit positioning (long/short account-ratio) — READ-ONLY ===")
    typer.echo(f"{direction_note}")
    typer.echo(f"scope: {scope}  |  bet: PRICE (close-to-close), cross-sectional, market-neutral")
    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']}  "
                   f"span={m['first_day']}..{m['last_day']}")
        typer.echo(f"  n-coins (tradeable breadth): {m['n_coins']}")
    else:
        typer.echo(f"  N/A — {m.get('reason', '')}  (n-coins={m.get('n_coins', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-long-short` + `fxhnt bybit-ingest-klines` first)")
    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.")

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) ===")
    typer.echo(f"signal: forced OI-unwind overshoot (-sign(ret)·|ΔOI|·|ret| on OI-DROP days)  |  scope: "
               f"{scope}  |  bet: PRICE (next-day), cross-sectional, market-neutral")
    typer.echo(f"  {direction_note}")
    if not rep.get("available"):
        typer.echo(f"  N/A — {rep.get('reason', '')}  (n-coins={rep.get('n_coins', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-oi` + `fxhnt bybit-ingest-klines` first)")
        return
    typer.echo(f"  n-coins (tradeable breadth): {rep['n_coins']}")

    typer.echo("\n-- NET-OF-COST, BOTH DIRECTIONS (Sharpe / CAGR / maxDD / total / days) --")
    for d in ("reversion", "momentum"):
        typer.echo(f"  [{d.upper()}]")
        for bps in rep["cost_grid"]:
            m = rep["net_cost"][d][bps]
            typer.echo(f"    cost={bps:5.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-- PER-COIN ATTRIBUTION (gross; is the edge broad or 1-2-coin fragile?) --")
    for d in ("reversion", "momentum"):
        attr = rep["per_coin"][d]
        top = sorted(attr.items(), key=lambda kv: abs(kv[1]), reverse=True)[:5]
        top_str = ", ".join(f"{c}={v*100:+.2f}%" for c, v in top)
        typer.echo(f"  [{d.upper()}] top-5 |share| of gross = {rep['top5_share'][d]*100:5.1f}%  | "
                   f"top: {top_str}")

    typer.echo("\n-- PER-YEAR Sharpe (net @ headline cost; all-from-one-year => overfit) --")
    for d in ("reversion", "momentum"):
        years = rep["per_year"][d]
        ys = "  ".join(f"{yr}:Sh={y['sharpe']:+5.2f}(n={y['days']})" for yr, y in sorted(years.items()))
        typer.echo(f"  [{d.upper()}] {ys}")

    typer.echo("\n-- AVERAGE DAILY TURNOVER (Σ|Δw|; high => costs bite + lower capacity) --")
    for d in ("reversion", "momentum"):
        typer.echo(f"  [{d.upper()}] {rep['avg_turnover'][d]:.3f}")

    typer.echo(f"\n-- CORRELATION with EACH existing edge (net @ {cost_bps:.1f}bps; ~0 => additive) --")
    for d in ("reversion", "momentum"):
        corrs = rep["corr_edges"][d]
        parts = []
        for edge in ("tstrend", "unlock", "xsfunding", "positioning"):
            c = corrs.get(edge)
            parts.append(f"{edge}={'N/A' if c is None else f'{c:+.3f}'}")
        typer.echo(f"  [{d.upper()}] " + "  ".join(parts))

    typer.echo("\nVERDICT GUIDE: the winning direction is a ROBUST uncorrelated edge only if it stays Sharpe>0 "
               "at 22bps, is NOT driven by 1-2 coins (top-5 share well below ~100% on a broad universe), is "
               "positive in MORE THAN ONE year, AND is uncorrelated with ALL FOUR existing edges (tstrend / "
               "unlock / xsfunding / positioning — ESPECIALLY positioning, since a ΔOI signal must not just "
               "re-express the positioning LEVEL). Otherwise it is an overfit / low-cost / single-regime / "
               "non-additive artifact. A CLEAN REJECTION here is a valid, valuable outcome.")
    typer.echo("\nNOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched. Memory-bounded.")


def _print_oi_deleverage_walk_forward(rep: dict, *, scope: str, cost_bps: float) -> None:
    """Pretty-print the OI-deleveraging OOS / walk-forward validation report (the deploy gate)."""
    typer.echo("\n=== Bybit OI-deleveraging — OOS / WALK-FORWARD VALIDATION (DEPLOY GATE, READ-ONLY) ===")
    typer.echo(f"signal: forced OI-unwind overshoot  |  scope: {scope}  |  cost: {cost_bps:.1f}bps")
    if not rep.get("available"):
        typer.echo(f"  N/A — {rep.get('reason', '')}  (n-coins={rep.get('n_coins', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-oi` + `fxhnt bybit-ingest-klines` first)")
        return
    typer.echo(f"  n-coins (tradeable breadth): {rep['n_coins']}  |  direction (TRAIN-chosen): "
               f"{rep['direction'].upper()}  |  full-sample Sharpe: {rep['full_sharpe']:.2f}")

    rw = rep["rolling_windows"]
    typer.echo(f"\n-- ROLLING NON-OVERLAPPING WINDOWS ({rep['window_days']}d; positive in MOST windows) --")
    typer.echo(f"  windows={rw['n_windows']}  fraction-positive={rw['fraction_positive']*100:5.1f}%")
    for w in rw["windows"]:
        typer.echo(f"    {w['start']}..{w['end']}  Sharpe={w['sharpe']:+6.2f}  "
                   f"total={w['total_return']*100:+7.1f}%  days={w['days']}")

    th = rep["train_holdout"]
    typer.echo("\n-- TRAIN / HOLDOUT DIRECTION (pick the sign on TRAIN, score it OOS on never-seen TEST) --")
    if th.get("available"):
        typer.echo(f"  TRAIN-chosen direction = {th['train_direction'].upper()}  |  "
                   f"split@ {th.get('split_at')}  (train {th['train_days']}d / test {th['test_days']}d)")
        typer.echo(f"  TRAIN Sharpe = {th['train_sharpe']:+6.2f}   ->   "
                   f"TEST (OOS) Sharpe = {th['test_sharpe']:+6.2f}   "
                   f"{'[GENERALISES]' if th['test_sharpe'] > 0 else '[OVERFIT — OOS NEGATIVE]'}")
    else:
        typer.echo("  N/A — too few days to split")

    dt = rep["drop_top_n"]
    typer.echo("\n-- DROP-TOP-N CONTRIBUTORS (ranked CAUSALLY on TRAIN; collapse => concentration-driven) --")
    typer.echo(f"  full Sharpe = {dt['full_sharpe']:+6.2f}")
    for n in (5, 10):
        d = dt[f"drop{n}"]
        typer.echo(f"  drop-top-{n:<2d} Sharpe = {d['sharpe']:+6.2f}  "
                   f"total={d['total_return']*100:+7.1f}%  (dropped {d['n_dropped']}: "
                   f"{', '.join(d['dropped'][:6])}{'…' if len(d['dropped']) > 6 else ''})")

    typer.echo("\n-- LIQUIDITY-THRESHOLD SWEEP (min daily $vol; does the edge need a specific universe?) --")
    for thr, m in sorted(rep["liquidity_sweep"].items()):
        typer.echo(f"  min_dollar_vol >= ${thr:>12,.0f}  Sharpe={m['sharpe']:+6.2f}  "
                   f"total={m['total_return']*100:+7.1f}%  n-coins={m['n_coins']}")

    la = rep["look_ahead"]
    typer.echo("\n-- LOOK-AHEAD AUDIT (signal at d -> d->d+1 return only; no same-day/future leak) --")
    typer.echo(f"  causal={la['causal']}  leak-days={la['leak_days']}  "
               f"(compared {la['n_days_compared']} days)")

    v = rep["verdict"]
    typer.echo("\n=== VERDICT ===")
    labels = {
        "rolling_windows_positive": ">70% of rolling windows positive",
        "oos_direction_positive": "TEST/OOS Sharpe > 0 with the TRAIN-chosen direction",
        "survives_drop_top5": "survives drop-top-5 (Sharpe still clearly positive)",
        "liquidity_robust": "positive across the liquidity sweep",
        "look_ahead_clean": "look-ahead audit clean (strictly causal)",
    }
    for key, label in labels.items():
        ok = v["checks"].get(key, False)
        typer.echo(f"  [{'PASS' if ok else 'FAIL'}] {label}")
    typer.echo(f"\n  >>> {'PASS — DEPLOYABLE' if v['deployable'] else 'FAIL — DO NOT DEPLOY'} <<<")
    typer.echo("  (a clean FAIL/rejection is a valid, valuable outcome — like cross-venue funding)")
    typer.echo("\nNOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched. Memory-bounded.")


@app.command("bybit-oi-deleverage-eval")
def bybit_oi_deleverage_eval(
    liquid: bool = typer.Option(
        False, "--liquid/--all",
        help="Restrict the universe to the liquid Bybit subset (median daily turnover >= --min-dollar-vol). "
             "Default --all uses every coin with an open_interest panel (the naive breadth)."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) used by --liquid."),
    direction: str = typer.Option(
        "reversion", "--direction",
        help="reversion (DEFAULT: fade the forced OI-unwind move) or momentum (the forced move continues — "
             "the exact sign-flip). Both are RE-RUN so the cost is correctly subtracted in EACH."),
    cost_bps: float = typer.Option(
        5.5, "--cost-bps", help="Per-turnover cost (bps) applied to the OI-deleveraging curve."),
    verify: bool = typer.Option(
        False, "--verify",
        help="ADVERSARIAL verification mode: BOTH directions net-of-cost across a cost grid (5.5/11/22 bps), "
             "per-coin attribution (top-5 share), per-year Sharpe, average daily turnover, AND the correlation "
             "with EACH existing edge (tstrend/unlock/xsfunding/positioning) — to judge whether the winning "
             "direction is a robust ADDITIVE edge or an artifact."),
    walk_forward: bool = typer.Option(
        False, "--walk-forward",
        help="OOS / WALK-FORWARD validation (the DEPLOY GATE): rolling non-overlapping windows + "
             "fraction-positive, a TRAIN/TEST direction split, drop-top-N robustness, a liquidity sweep, a "
             "look-ahead audit, and a PASS/FAIL deploy verdict. Use with --liquid on live data."),
    window_days: int = typer.Option(
        180, "--window-days",
        help="Width (days) of each rolling non-overlapping OOS window in --walk-forward."),
) -> None:
    """READ-ONLY: evaluate the NEW Bybit OI-DELEVERAGING REVERSION edge off open interest + price.

    Signal = on an OI-DROP day (forced unwind: doi<0), FADE that day's move with magnitude ∝ |ΔOI|·|ret|;
    OI-build days carry no signal. The bet is on PRICE (next-day, cross-sectional, market-neutral) on LIQUID
    coins. MOMENTUM is the sign-flip (the forced move continues). Both directions are tested.

    Prints the edge metrics + n-coins breadth. With --verify it stress-tests both directions across a cost
    grid, attributes per-coin, slices per-year, reports turnover, and shows the correlation with each existing
    edge so additivity (esp. vs positioning) is visible. With --walk-forward it runs the OOS deploy gate.

    Memory-bounded: with --liquid only the liquid coins are READ. Writes NOTHING. Real numbers appear only once
    the OI + kline ingests have populated bybit_features."""
    if direction not in ("reversion", "momentum"):
        raise typer.BadParameter("must be 'reversion' or 'momentum'", param_hint="--direction")

    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.bybit_oi_deleverage_eval import (
        oi_deleverage_metrics_from_store,
        verify_oi_deleverage_edge,
        walk_forward_oi_deleverage,
    )

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if liquid else None
    scope = (f"LIQUID subset (median turnover >= ${min_dollar_vol:,.0f})" if liquid
             else "ALL coins with an open_interest panel (naive breadth)")
    direction_note = ("direction: " + (
        "REVERSION (fade the forced OI-unwind move)" if direction == "reversion"
        else "MOMENTUM (the forced move continues — sign-flip)"))

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock corr simply N/A, not a crash
            log.warning("bybit-oi-deleverage-eval: could not load unlock calendar: %s", exc)
            return []

    if walk_forward:
        try:
            rep = walk_forward_oi_deleverage(
                store, universe=universe, cost_bps=cost_bps, window_days=window_days)
        finally:
            store.close()
        _print_oi_deleverage_walk_forward(rep, scope=scope, cost_bps=cost_bps)
        return

    if verify:
        try:
            rep = verify_oi_deleverage_edge(
                store, universe=universe, cost_bps=cost_bps, unlock_events=_load_unlock_events())
        finally:
            store.close()
        _print_oi_deleverage_verify(rep, scope=scope, direction_note=direction_note, cost_bps=cost_bps)
        return

    try:
        m = oi_deleverage_metrics_from_store(
            store, universe=universe, cost_bps=cost_bps, direction=direction)
    finally:
        store.close()

    typer.echo("\n=== Bybit OI-deleveraging reversion — READ-ONLY ===")
    typer.echo(f"{direction_note}")
    typer.echo(f"scope: {scope}  |  bet: PRICE (next-day), cross-sectional, market-neutral")
    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']}  "
                   f"span={m['first_day']}..{m['last_day']}")
        typer.echo(f"  n-coins (tradeable breadth): {m['n_coins']}")
    else:
        typer.echo(f"  N/A — {m.get('reason', '')}  (n-coins={m.get('n_coins', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-oi` + `fxhnt bybit-ingest-klines` first)")
    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("positioning-metalabel")
def positioning_metalabel(
    liquid: bool = typer.Option(
        False, "--liquid/--all",
        help="Restrict the meta-labeling universe to the liquid Bybit subset (median daily turnover >= "
             "--min-dollar-vol). Default --all uses every coin with a long_ratio panel."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) used by --liquid."),
    label_horizon: int = typer.Option(
        1, "--label-horizon",
        help="Forward horizon (days) over which the primary position's profitability is labeled (causal)."),
    n_splits: int = typer.Option(
        5, "--n-splits", help="Number of PURGED k-folds for the leakage-free OOF cross-validation."),
    embargo_days: int = typer.Option(
        1, "--embargo-days", help="Embargo gap (days) around each test fold in the purged CV."),
    shuffle_label: bool = typer.Option(
        False, "--shuffle-label",
        help="LEAKAGE TRIPWIRE: shuffle the labels and re-run. A clean pipeline shows NO OOS lift with "
             "shuffled labels; if it lifts, the pipeline leaks (fail loudly)."),
    seed: int = typer.Option(0, "--seed", help="RNG seed (label shuffle + classifier determinism)."),
) -> None:
    """READ-ONLY: de Prado META-LABELING experiment on the Bybit POSITIONING edge (purged CV, skeptical OOS).

    The positioning edge (contrarian fade of the long/short account ratio) is the PRIMARY model — it sets the
    SIDE per coin-day. A simple, regularized META-MODEL (LogisticRegression, L2) learns WHEN to trust each
    coin-day signal (filter false positives) + SIZE by confidence P(profitable). Separating side from
    size/filter can raise precision → higher OOS Sharpe — IF it survives PURGED+EMBARGOED cross-validation
    (it often does NOT; ML overfits finance). "Meta-labeling doesn't add" is a VALID, honest outcome.

    Prints: the few documented meta-features + their importances, the PURGED out-of-sample RAW vs META-LABELED
    comparison (Sharpe / CAGR / maxDD / precision / trades-kept / Sharpe-lift), and the conservative verdict
    "META-LABELING ADDS (deploy)" vs "NO LIFT — STAY RAW". Use --shuffle-label as the leakage control.

    Memory-bounded (with --liquid only liquid coins are READ). Writes NOTHING — the live paper_* tables are
    untouched. Real numbers appear only once the long-short + kline ingests have populated bybit_features."""
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.positioning_metalabel import metalabel_experiment

    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if liquid else None
    scope = (f"LIQUID subset (median turnover >= ${min_dollar_vol:,.0f})" if liquid
             else "ALL coins with a long_ratio panel")
    try:
        rep = metalabel_experiment(
            store, universe=universe, label_horizon=label_horizon, n_splits=n_splits,
            embargo_days=embargo_days, shuffle_label=shuffle_label, seed=seed)
    finally:
        store.close()
    _print_positioning_metalabel(rep, scope=scope, shuffle_label=shuffle_label)


def _print_carry_metalabel(rep: dict, *, scope: str, shuffle_label: bool) -> None:
    """Pretty-print the de Prado CARRY meta-labeling experiment (purged-CV comparison + skeptical verdict)."""
    typer.echo("\n=== Bybit CARRY (xsfunding) — META-LABELING EXPERIMENT (de Prado, PURGED CV, READ-ONLY) ===")
    typer.echo("primary (SIDE) = cross-sectional funding/basis carry  |  meta (SIZE/FILTER) = "
               "LogisticRegression P(profitable)")
    typer.echo(f"scope: {scope}" + ("   |   SHUFFLE-LABEL CONTROL (leakage tripwire: expect NO LIFT)"
                                     if shuffle_label else ""))
    if not rep.get("available"):
        typer.echo(f"  N/A — {rep.get('reason', '')}  (n-rows={rep.get('n_rows', 0)})")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first)")
        return

    typer.echo(f"  coin-day rows={rep['n_rows']}  OOS rows={rep['n_oos_rows']}  "
               f"purged {rep['n_splits']}-fold (embargo {rep['embargo_days']}d, "
               f"label_horizon {rep['label_horizon']}d)  threshold(P>)={rep['threshold']:.2f}")

    typer.echo("\n-- FEW META-FEATURES (anti-overfit) + averaged FEATURE IMPORTANCE (|coef|) --")
    for name in rep["feature_names"]:
        typer.echo(f"    {name:<18s} importance={rep['feature_importances'][name]:+.4f}")

    o = rep["oos"]
    typer.echo("\n-- PURGED OUT-OF-SAMPLE: RAW (side-only) vs META-LABELED (filter P>thr, size ∝ P) --")
    typer.echo(f"    RAW   Sharpe={o['raw_sharpe']:+6.2f}  CAGR={o['raw_cagr']*100:+7.1f}%  "
               f"maxDD={o['raw_max_dd']*100:+6.1f}%  days={o['raw_days']}")
    typer.echo(f"    META  Sharpe={o['meta_sharpe']:+6.2f}  CAGR={o['meta_cagr']*100:+7.1f}%  "
               f"maxDD={o['meta_max_dd']*100:+6.1f}%  days={o['meta_days']}")
    typer.echo(f"    Sharpe LIFT (meta-raw) = {o['sharpe_lift']:+6.2f}   "
               f"(adopt only if > {rep['verdict']['add_margin']:.2f})")
    typer.echo(f"    OOS precision (kept→profitable) = {o['meta_precision']*100:5.1f}%  "
               f"vs base-rate {o['base_rate']*100:5.1f}%  (lift {o['precision_lift']*100:+5.1f}pp)  |  "
               f"trades kept = {o['frac_trades_kept']*100:5.1f}%")

    v = rep["verdict"]
    typer.echo("\n=== VERDICT (PURGED-OOS only — train metrics are reference, never the verdict) ===")
    typer.echo(f"  >>> {v['label']} <<<")
    if shuffle_label and not v["adds"]:
        typer.echo("  [OK] shuffled-label control shows NO lift → pipeline is leakage-free.")
    elif shuffle_label and v["adds"]:
        typer.echo("  [LEAK] shuffled-label control LIFTED — pipeline LEAKS; do NOT trust the real run.")
    typer.echo("\nNOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched. Memory-bounded.")


@app.command("carry-metalabel")
def carry_metalabel(
    liquid: bool = typer.Option(
        False, "--liquid/--all",
        help="Restrict the meta-labeling universe to the liquid Bybit subset (median daily turnover >= "
             "--min-dollar-vol). Default --all uses every coin with a funding/close panel."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) used by --liquid."),
    label_horizon: int = typer.Option(
        1, "--label-horizon",
        help="Forward horizon (days) over which the primary carry position's profitability is labeled "
             "(causal: forward carry-returns only)."),
    n_splits: int = typer.Option(
        5, "--n-splits", help="Number of PURGED k-folds for the leakage-free OOF cross-validation."),
    embargo_days: int = typer.Option(
        1, "--embargo-days", help="Embargo gap (days) around each test fold in the purged CV."),
    shuffle_label: bool = typer.Option(
        False, "--shuffle-label",
        help="LEAKAGE TRIPWIRE: shuffle the labels and re-run. A clean pipeline shows NO OOS lift with "
             "shuffled labels; if it lifts, the pipeline leaks (fail loudly)."),
    seed: int = typer.Option(0, "--seed", help="RNG seed (label shuffle + classifier determinism)."),
) -> None:
    """READ-ONLY: de Prado META-LABELING experiment on the Bybit CARRY (xsfunding) edge (purged CV, skeptical OOS).

    The cross-sectional CARRY book (funding + delta-neutral basis) is the PRIMARY model — it sets the SIDE per
    coin-day (LONG high-carry / SHORT low-carry). A simple, regularized META-MODEL (LogisticRegression, L2)
    learns WHEN to trust each coin-day signal (filter false positives) + SIZE by confidence P(profitable).
    Carry is a DIFFERENT mechanism than retail positioning, so it MIGHT carry filterable structure positioning
    lacked — but it often does NOT; "meta-labeling doesn't add" is a VALID, honest outcome (positioning's was).

    Reuses the positioning meta-label FRAMEWORK unchanged (PurgedKFold + embargo, the LogisticRegression
    meta-model, the bet-sizing apply, the shuffled-label leakage control, the _ADD_MARGIN=0.25 verdict) — only
    the per-coin-day dataset (carry primary signal + forward carry-return) is carry-specific.

    Prints: the few documented meta-features + their importances, the PURGED out-of-sample RAW vs META-LABELED
    comparison (Sharpe / CAGR / maxDD / precision / trades-kept / Sharpe-lift), and the conservative verdict
    "META-LABELING ADDS (deploy)" vs "NO LIFT — STAY RAW". Use --shuffle-label as the leakage control.

    Memory-bounded (with --liquid only liquid coins are READ). Writes NOTHING — the live paper_* tables are
    untouched. Real numbers appear only once the funding + kline ingests have populated bybit_features."""
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_carry_metalabel import carry_metalabel_experiment
    from fxhnt.application.bybit_liquidity import liquid_universe

    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if liquid else None
    scope = (f"LIQUID subset (median turnover >= ${min_dollar_vol:,.0f})" if liquid
             else "ALL coins with a funding/close panel")
    try:
        rep = carry_metalabel_experiment(
            store, universe=universe, label_horizon=label_horizon, n_splits=n_splits,
            embargo_days=embargo_days, shuffle_label=shuffle_label, seed=seed)
    finally:
        store.close()
    _print_carry_metalabel(rep, scope=scope, shuffle_label=shuffle_label)


@app.command("bybit-book-eval")
def bybit_book_eval(
    liquid: bool = typer.Option(
        True, "--liquid/--all",
        help="Restrict the crypto universe to the liquid tradeable subset (median daily turnover >= "
             "--min-dollar-vol). DEFAULT (and the memory-bounded path): only liquid symbols are READ. "
             "--all reads the FULL universe — OOM-risk; use the Job, never `kubectl exec` in the daemon."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) used by --liquid."),
    capital: float = typer.Option(
        100_000.0, "--capital", help="Starting equity for the risk-managed book curve."),
    cost_bps: float = typer.Option(
        5.5, "--cost-bps", help="Per-turnover cost (bps) applied to each sleeve + the book."),
    controller: str = typer.Option(
        "killswitch", "--controller",
        help="Risk-overlay controller (killswitch=live default / combined / edge_health)."),
) -> None:
    """READ-ONLY: run the LIVE risk overlay on the BYBIT liquid sleeves and print the deployable RISK-MANAGED
    book — the true analog of the Sharpe ~2.74 legacy combined book. For each Bybit-computable sleeve (tstrend +
    unlock + xsfunding + positioning) it derives the per-day return series off `bybit_features` via the SAME
    validated edge runners, then runs `simulate_book` — the EXACT live overlay (`IncrementalRiskOverlay` ==
    `paper_risk_overlay`: ISV-adaptive weights + adaptive vol-target + DrawdownKillSwitch; `--controller
    combined` adds the structural-retirement gate). Prints each sleeve's STANDALONE Sharpe, the risk-managed
    book (Sharpe / CAGR / maxDD / avg-leverage / killed-days), the naive EQUAL-WEIGHT comparison, and the
    per-sleeve correlation matrix (the diversification view).

    READ-ONLY — no PaperRepo, no store.write_*; the live paper_* tables are untouched. MEMORY-BOUNDED — with
    --liquid (default) only the liquid subset is read (`read_panel(liquid_symbols, ...)`), never the whole
    ~794-coin table. Heavy compute should run as the Job (own memory limit), NOT via `kubectl exec` in the
    dagster daemon — that shared-memory exec is what OOM'd the --all eval.

    unlock needs the DefiLlama unlock calendar (loaded live from `unlock_calendar.json`); absent → the unlock
    sleeve is simply omitted (an empty series), never fabricated. positioning needs the `long_ratio` feature
    (ingested for the liquid universe); absent → the positioning sleeve is omitted, never fabricated. Real
    numbers appear only once the Phase-A/B Bybit ingests have populated bybit_features; before that the book
    reports 'insufficient data'."""
    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_book_eval import risk_managed_book_from_store
    from fxhnt.application.bybit_liquidity import liquid_universe

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        # DB-first (operational `unlock_events`, networked — no surfer-data PVC), JSON fallback for
        # back-compat. Absent calendar → unlock sleeve omitted, never a crash.
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock omitted, not a crash
            log.warning("bybit-book-eval: could not load unlock calendar: %s", exc)
            return []

    universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if liquid else None
    try:
        book = risk_managed_book_from_store(
            store, universe=universe, capital=capital, cost_bps=cost_bps, controller=controller,
            unlock_events=_load_unlock_events())
    finally:
        store.close()

    scope = (f"LIQUID subset (median turnover >= ${min_dollar_vol:,.0f}, {len(universe)} symbols)"
             if liquid else "ALL symbols (naive / over-diversified — OOM-risk)")
    typer.echo(f"\n=== Bybit RISK-MANAGED book — LIVE overlay — READ-ONLY — {scope} ===")

    if book["days"] == 0:
        typer.echo(f"INSUFFICIENT DATA: {book.get('reason', 'no sleeve has >= 2 days of returns')}")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first to populate "
                   "bybit_features)")
        return

    typer.echo("\nStandalone per-sleeve Sharpe (no overlay):")
    for sleeve in book["sleeves"]:
        typer.echo(f"  {sleeve:<18} Sharpe={book['per_sleeve_standalone_sharpe'][sleeve]:6.2f}")

    ls = book["leverage_stats"]
    typer.echo(f"\nRISK-MANAGED book ({book['controller']} overlay, kelly={book['kelly_fraction']:.2f}, "
               f"capital=${book['capital']:,.0f}):")
    typer.echo(f"  Sharpe={book['sharpe']:6.2f}  CAGR={book['cagr']*100:7.1f}%  "
               f"maxDD={book['max_dd']*100:7.1f}%  totRet={book['total_return']*100:8.1f}%")
    typer.echo(f"  avg-leverage={ls['avg']:5.2f} (min {ls['min']:.2f} / max {ls['max']:.2f})  "
               f"killed-days={book['killed_days']}  days={book['days']}  sleeves={len(book['sleeves'])}")

    nw = book["naive_equal_weight"]
    typer.echo("\nNaive EQUAL-WEIGHT (no overlay) baseline:")
    typer.echo(f"  Sharpe={nw['sharpe']:6.2f}  CAGR={nw['cagr']*100:7.1f}%  "
               f"maxDD={nw['max_dd']*100:7.1f}%  totRet={nw['total_return']*100:8.1f}%  days={nw['days']}")
    typer.echo(f"\nOverlay delta: Sharpe {nw['sharpe']:.2f} -> {book['sharpe']:.2f}  "
               f"(the live vol-target + killswitch + ISV contribution)")

    corr = book.get("correlation_matrix") or {}
    if corr:
        sleeves = book["sleeves"]
        typer.echo("\nPer-sleeve correlation matrix (common-day Pearson — the diversification view):")
        header = "  " + " " * 18 + "".join(f"{s[:10]:>11}" for s in sleeves)
        typer.echo(header)
        for si in sleeves:
            cells = []
            for sj in sleeves:
                v = corr.get(si, {}).get(sj)
                cells.append(f"{v:>11.2f}" if v is not None else f"{'n/a':>11}")
            typer.echo(f"  {si:<18}" + "".join(cells))

    typer.echo("NOTE: writes NOTHING — paper_nav/paper_sleeve_ret untouched. Overlay = the SAME "
               "IncrementalRiskOverlay (== paper_risk_overlay) the live combined book uses. The Bybit data "
               "universe is GLOBAL; live Bybit-EU (MiCA) trades a SMALLER subset, so this is an upper bound.")


@app.command("honest-reference-report")
def honest_reference_report(
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD): only this liquid subset of bybit_features "
             "is READ, keeping the multi-sleeve compute memory-bounded (same seam the sleeve refs use)."),
) -> None:
    """READ-ONLY crypto honest-reference RESEARCH report (spec 9): the per-sleeve capacity curve (AUM × impact-k),
    the risk-parity base book + leverage ceiling, the three falsification gates, candidate edge-4 (turnover-pump
    short) through the honest-cost pipeline, and the survivorship guard — emitted as JSON. Writes NOTHING (no
    cockpit / backtest_summary / recon-gate / prod tables); construct-the-store-and-print only.

    The store, liquid universe and unlock calendar are built the SAME way `_persist_bybit_book` builds them; the
    per-coin real L1 spread FLOOR is read from the `bybit_real_spread` SSOT via `PaperRepo.read_real_spread_by_coin`."""
    import json

    from fxhnt.adapters.orchestration.assets import _data_dir
    from fxhnt.adapters.persistence.paper_repo import PaperRepo
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_book_eval import _bounded_store
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.honest_report import build_honest_report
    from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo

    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    try:
        # Memory bound: wrap the store in the SAME `_bounded_store` seam the sleeve runners use, so the panel
        # reads inside build_honest_report (crypto_close_panel + read_panel) load ONLY the liquid subset, not
        # the full ~794-coin table. `universe is None` (no liquid subset) → raw store (documented OOM-risk).
        universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) or None
        bounded = _bounded_store(store, universe)
        try:
            unlock_events = load_unlock_events(operational_unlock_repo(settings),
                                               f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock sleeve omitted, not a crash
            typer.echo(f"# warning: could not load unlock calendar: {exc}", err=True)
            unlock_events = []

        spread_bps = PaperRepo(settings.operational_dsn).read_real_spread_by_coin()
        result = build_honest_report(
            bounded, spread_bps, unlock_events,
            sleeves=["xsfunding", "unlock", "positioning"],
            aums=[35_000, 100_000, 350_000, 1_000_000, 10_000_000],
            ks=[1.0, 2.0, 4.0])
    finally:
        store.close()
    typer.echo(json.dumps(result, indent=2, default=str))


@app.command("bybit-persist-sleeve-ret")
def bybit_persist_sleeve_ret(
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD): only this liquid subset of bybit_features "
             "is READ, which is what keeps the 4-sleeve compute memory-bounded."),
    cost_bps: float | None = typer.Option(
        None, "--cost-bps",
        help="Per-turnover cost (bps) applied to each sleeve's daily return series. Default: "
             "Settings.bybit_cost_bps (the single-sourced, measured Bybit liquid-book cost — the same "
             "cost the forward tracks charge, so this ref can never silently diverge from them)."),
) -> None:
    """ON-DEMAND WRITER of the Bybit 4-edge BACKTEST data: compute the full per-sleeve daily return series
    (tstrend + unlock + xsfunding + positioning) on the LIQUID `bybit_features` universe via the SAME live
    edge runners, and idempotently upsert them into `bybit_sleeve_ret` — the table the cockpit sim reads so it
    NEVER recomputes the heavy 4-edge book on a web request. Also upserts the `_BYBIT_BOOK_SIDS` report-kind
    reconciliation-gate refs (bybit_4edge / bybit_4edge_levered / positioning / xsfunding / unlock) from the same
    computed series.

    Thin wrapper over the shared `_persist_bybit_book` helper — the SAME compute-once logic the scheduled
    `infra/k8s/jobs/fxhnt-backtest-refs.yaml` CronJob (`fxhnt backtest-refs --all`) runs weekly. Use this
    command for an on-demand / out-of-band run; the CronJob is the standing schedule.

    MEMORY-BOUNDED: only the `--min-dollar-vol` liquid subset is READ (never the full ~794-coin table).
    IDEMPOTENT: (run_date, sleeve) primary-key overwrite — re-running on the same data rewrites the same rows.
    unlock needs the DefiLlama calendar (DB-first, JSON fallback); absent → the unlock sleeve is simply
    omitted, never fabricated."""
    n, scope = _persist_bybit_book(get_settings(), min_dollar_vol=min_dollar_vol, cost_bps=cost_bps)
    cost_desc = cost_bps if cost_bps is not None else "Settings.bybit_cost_bps"
    typer.echo(f"bybit-persist-sleeve-ret: persisted {n} bybit_sleeve_ret rows from {scope} "
               f"(min_dollar_vol=${min_dollar_vol:,.0f}, cost_bps={cost_desc}).")


@app.command("backtest-refs")
def backtest_refs_cmd(
    all_: bool = typer.Option(
        False, "--all",
        help="Refresh the reconciliation-gate backtest_summary ref for every HEAVY (`report`-kind) "
             "strategy in STRATEGY_REGISTRY. The bybit-book sids (bybit_4edge / bybit_4edge_levered / "
             "positioning / xsfunding / unlock — see `_BYBIT_BOOK_SIDS`) go through the shared compute-once "
             "`_persist_bybit_book` helper (ALSO writes bybit_sleeve_ret, the cockpit sim's precompute — the "
             "same table the retired manual `bybit-persist-sleeve-ret` Job used to write). Any OTHER "
             "report-kind sid goes through the unified `backtest_ref(sid, store)` dispatcher "
             "(fxhnt.application.backtest_refs). The cheap recompute/sleeve-kind refs are persisted inline "
             "by the nightly assets, NOT here."),
) -> None:
    """OWN-MEMORY scheduled step: recompute + upsert the `report`-kind bybit-book backtest references
    (bybit_4edge, bybit_4edge_levered, positioning, xsfunding, unlock) via the shared compute-once
    `_persist_bybit_book` helper — the SAME helper the on-demand `bybit-persist-sleeve-ret` command calls —
    so this CronJob is a full, no-regression replacement for the retired manual sleeve-ret Job: it writes
    `bybit_sleeve_ret` (the cockpit sim's precompute) AND the `_BYBIT_BOOK_SIDS` report-kind refs from ONE
    compute-once pass. Any OTHER (non-bybit-book) report-kind strategy is then covered generically via the
    registry-driven `backtest_ref` dispatch; a per-strategy failure logs and CONTINUES (never aborts the
    sweep) — a missing ref just means the gate keeps WAITing for that one strategy."""
    import datetime as dt
    import logging

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.backtest_refs import backtest_ref, registry_backtest_kind
    from fxhnt.registry import STRATEGY_REGISTRY

    log = logging.getLogger(__name__)
    if not all_:
        typer.echo("backtest-refs: pass --all to refresh every report-kind strategy's backtest_summary ref.")
        raise typer.Exit(code=1)

    settings = get_settings()

    # The bybit 4-edge book (bybit_4edge / bybit_4edge_levered / positioning / xsfunding): compute-once batch
    # via the SAME shared helper `bybit-persist-sleeve-ret` calls on-demand. This is what makes
    # `backtest-refs --all` a true, complete replacement for the retired manual sleeve-ret Job — it ALSO
    # writes `bybit_sleeve_ret` (the cockpit sim's precompute), not just the refs.
    n_rows, scope = _persist_bybit_book(settings)
    typer.echo(f"backtest-refs --all: bybit book -> {n_rows} bybit_sleeve_ret row(s) from {scope} "
               f"+ {len(_BYBIT_BOOK_SIDS)} report refs ({', '.join(_BYBIT_BOOK_SIDS)}) via _persist_bybit_book.")

    # Any OTHER report-kind sid (future non-bybit-book report strategies): generic per-sid dispatch. Today
    # this loop covers zero sids (all report-kind sids are the bybit book, handled above) — logged
    # explicitly so that coverage is never silent.
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    fwd_repo = ForwardNavRepo(settings.operational_dsn)
    fwd_repo.migrate()
    at = dt.datetime.now(dt.UTC).replace(tzinfo=None)
    n_ok, n_fail, n_skip = 0, 0, 0
    try:
        for sid in STRATEGY_REGISTRY:
            if registry_backtest_kind(sid) != "report" or sid in _BYBIT_BOOK_SIDS:
                continue   # bybit-book sids are handled above by _persist_bybit_book; cheap kinds run inline
                           # in the nightly assets (Task 6)
            try:
                sm = backtest_ref(sid, store)
                if sm is None:
                    n_skip += 1
                    typer.echo(f"{sid}: no ref produced (insufficient data), skipped")
                    continue
                fwd_repo.upsert_backtest_summary(sm, at=at)
                n_ok += 1
                typer.echo(f"{sid}: backtest ref cagr={sm.cagr * 100:.1f}% sharpe={sm.sharpe:.2f} "
                           "-> backtest_summary")
            except Exception as exc:  # noqa: BLE001 — best-effort per strategy; never abort the sweep
                n_fail += 1
                log.warning("backtest-refs --all: %s failed: %s", sid, exc)
                typer.echo(f"{sid}: FAILED ({exc}), continuing")
    finally:
        store.close()
    n_other = n_ok + n_fail + n_skip
    log.info("backtest-refs --all: %d non-bybit-book report-kind sid(s) handled generically "
             "(0 expected today — all report-kind sids are the bybit book, handled by _persist_bybit_book).",
             n_other)
    typer.echo(f"backtest-refs --all: {n_ok} additional ref(s) persisted, {n_skip} skipped, {n_fail} failed "
               f"(non-bybit-book report-kind sids: {n_other} handled, 0 expected today).")


@app.command("bybit-spread-refresh")
def bybit_spread_refresh(
    samples: int = typer.Option(
        5, "--samples",
        help="How many L1 top-of-book snapshots to sample (a few seconds apart) and take the MEDIAN of per "
             "coin — robust against one weird moment."),
    interval_s: float = typer.Option(
        3.0, "--interval-s", help="Seconds to wait between L1 snapshot samples."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD): restrict the refreshed coins to this liquid "
             "subset of bybit_features (0 = all symbols Bybit quotes)."),
) -> None:
    """WRITER: refresh the REAL per-coin L1 quoted spread for the Bybit liquid book and persist it to
    `bybit_real_spread`. SAMPLES Bybit's `/v5/market/tickers` top-of-book `--samples` times a few seconds
    apart and stores the MEDIAN spread (bps) per coin — the forward-realistic cost input the measured-cost
    precompute charges the liquid book INSTEAD of the daily-high/low Corwin–Schultz proxy (which over-states
    the true quoted spread ~104×: 1.4bp real vs 143.6bp CS).

    Needs egress to api.bybit.com (443) + the operational Postgres (5432). IDEMPOTENT: one row per coin,
    replaced on each run. A fetch failure is NON-FATAL (logged; nothing stored → the precompute keeps using CS
    for the un-refreshed coins), so chaining `bybit-spread-refresh && compare-measured-precompute` never blocks
    the precompute. READ-ONLY against bybit_features (only `bybit_real_spread` is written)."""
    import datetime as dt
    import logging

    from fxhnt.adapters.persistence.paper_repo import PaperRepo
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.bybit_spread_refresh import refresh_real_spreads

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    try:
        universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if min_dollar_vol > 0 else None
    finally:
        store.close()

    repo = PaperRepo(settings.operational_dsn)
    repo.migrate()
    from fxhnt.adapters.data.bybit_true_spread import BybitTrueSpread
    try:
        medians = refresh_real_spreads(repo, BybitTrueSpread(), samples=samples, interval_s=interval_s,
                                       at=dt.datetime.now(dt.UTC), universe=universe or None)
    except Exception as exc:  # noqa: BLE001 — no live quotes → nothing stored, precompute falls back to CS
        log.warning("bybit-spread-refresh: could not refresh live L1 spreads (kept prior/CS): %s", exc)
        typer.echo(f"bybit-spread-refresh: SKIPPED (live L1 fetch failed: {exc}); precompute will use CS.")
        return
    n = len(medians)
    import statistics as _st
    med = _st.median(medians.values()) if medians else 0.0
    typer.echo(f"bybit-spread-refresh: persisted {n} real L1 spreads to bybit_real_spread "
               f"(median {med:.2f}bp, samples={samples}) "
               f"@ {settings.operational_dsn.split('@')[-1]}")


@app.command("compare-measured-precompute")
def compare_measured_precompute(
    capital: float = typer.Option(
        0.0, "--capital",
        help="Start capital both compared books normalize to (0 = settings.paper_capital, the cockpit default)."),
    taker_fee_bps: float = typer.Option(
        5.5, "--taker-fee-bps", help="Published perp taker fee (bps); the floor of the per-coin measured cost."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Bybit liquid-universe floor (median daily turnover, USD) for the per-coin Bybit measured curve."),
) -> None:
    """OWN-MEMORY WRITER of the measured-cost artifacts: run the (heavy) per-COIN MEASURED-cost compute ONCE
    and idempotently persist the per-book SINGLE measured curves the Backtest's DEFAULT cost mode reads —
    `bybit_4edge` and `bybit_4edge_levered` (each the per-coin book × the `bybit_features` Corwin–Schultz
    spreads on the liquid subset) — into `compare_measured_cache`. The cockpit ONLY READS these rows.

    (Phase 0b venue consolidation: the legacy combined-book leg and the legacy-vs-Bybit COMPARE artifact
    this command used to also persist are retired — Bybit is fxhnt's sole crypto venue.)

    Why a CLI/Job, not the web handler: this compute OOM-killed the 4Gi `fxhnt-dashboard` pod when run live
    (exit 137 → empty/500). So it lives HERE (own-memory Job), run on-demand / weekly. The cockpit falls back
    to the LIGHT flat-cost view until these artifacts exist, so the page never OOMs.

    READ-ONLY against the warehouse (only `compare_measured_cache` is written, via the repo). IDEMPOTENT:
    one-row primary-key overwrite per key — re-running on the same data rewrites the same artifacts."""
    import datetime as dt
    import logging

    from fxhnt.adapters.orchestration.assets import _data_dir
    from fxhnt.adapters.persistence.paper_repo import PaperRepo
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.compare_measured_cost import precompute_single_book_measured
    from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo

    log = logging.getLogger(__name__)
    settings = get_settings()
    cap = capital if capital > 0 else settings.paper_capital
    repo = PaperRepo(settings.operational_dsn)
    repo.migrate()
    bybit_store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    now = dt.datetime.now(dt.UTC)
    try:
        # Bybit per-coin measured needs the liquid subset (memory bound) + the unlock calendar (absent →
        # the unlock sleeve is simply omitted, never fabricated).
        bybit_universe = liquid_universe(bybit_store, min_dollar_vol=min_dollar_vol) or None
        try:
            unlock_events = load_unlock_events(operational_unlock_repo(settings),
                                               f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock sleeve omitted, not a crash
            log.warning("compare-measured-precompute: could not load unlock calendar: %s", exc)
            unlock_events = []

        # The REAL per-coin L1 quoted spread (refreshed by `fxhnt bybit-spread-refresh`) — the forward-realistic
        # cost the Bybit liquid book is charged INSTEAD of the high/low CS proxy. Empty when no refresh has run yet
        # → every coin falls back to CS.
        from fxhnt.application.bybit_spread_refresh import read_real_spread_by_coin
        bybit_real_spread = read_real_spread_by_coin(repo) or None

        # The per-book SINGLE measured curves (the Backtest's DEFAULT cost mode).
        singles = precompute_single_book_measured(
            repo, bybit_store, capital=cap, taker_fee_bps=taker_fee_bps, at=now,
            bybit_universe=bybit_universe, unlock_events=unlock_events,
            bybit_real_spread_by_coin=bybit_real_spread)
    finally:
        bybit_store.close()
    by_r = singles.get("bybit_4edge", {})
    typer.echo(
        f"compare-measured-precompute: persisted single-book curves ({', '.join(sorted(singles)) or 'none'}) "
        f"capital ${cap:,.0f}. "
        f"Bybit median spread {by_r.get('median_spread_bps', 0.0):.1f}bp, "
        f"cost drag {by_r.get('cost_drag_bps', 0.0):.1f}bp.")


@app.command("bybit-paper-book")
def bybit_paper_book_cmd(
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD): only this liquid subset of bybit_features "
             "is READ, which keeps the per-symbol weight extraction memory-bounded."),
) -> None:
    """SEED the Bybit LIVE paper book ONCE (own-memory; the same work the nightly `bybit_paper_book` Dagster
    asset does): derive the naive-eq-wt 4-edge book's per-symbol positions on the latest Bybit close, diff vs
    the prior bybit book → trades, charge the 5.5bp turnover, and persist into the cockpit DB (bybit is
    `PaperRepo`'s sole venue). The cockpit then shows it at `/paper?venue=bybit` (mark-to-market, per-symbol).
    READ-ONLY against the warehouse. IDEMPOTENT per run_date.

    Run this to populate + verify the live book without waiting for the nightly asset."""
    import datetime as dt
    import logging

    from fxhnt.adapters.orchestration.assets import _data_dir, build_bybit_paper_book
    from fxhnt.adapters.persistence.paper_repo import PaperRepo
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    try:
        universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) or None
        try:
            unlock_events = load_unlock_events(operational_unlock_repo(settings),
                                               f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock sleeve omitted, not a crash
            log.warning("bybit-paper-book: could not load unlock calendar: %s", exc)
            unlock_events = []
        repo = PaperRepo(settings.operational_dsn)
        repo.migrate()
        n = build_bybit_paper_book(
            repo, store, capital=settings.paper_capital, at=dt.datetime.now(dt.UTC),
            universe=universe, unlock_events=unlock_events, enabled=True,
            positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP)
    finally:
        store.close()

    scope = f"{len(universe)} liquid symbols" if universe else "full universe (no liquid subset)"
    typer.echo(f"bybit-paper-book: persisted {n} open positions (venue=bybit) from {scope}. "
               f"View at /paper?venue=bybit.")


@app.command("bybit-paper-backfill")
def bybit_paper_backfill_cmd(
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD): only this liquid subset of bybit_features "
             "is READ, which keeps the per-symbol weight extraction memory-bounded (the measured-optimal "
             "default)."),
    capital: float | None = typer.Option(
        None, "--capital",
        help="Starting capital for the backfilled book (default: settings.paper_capital)."),
    rebuild: bool = typer.Option(
        False, "--rebuild",
        help="full rebuild of the whole Bybit history (the rare path). Default is INCREMENTAL: fill only "
             "days STRICTLY AFTER the last persisted venue=bybit paper_nav date (resumable — the backfill "
             "is idempotent per run_date, so resuming from the last date is exact)."),
) -> None:
    """BACKFILL the Bybit LIVE paper book over its FULL history (own-memory; the heavy ~1600-day counterpart
    of the single-day `bybit-paper-book` asset): for EVERY historical day on which the 4-edge weights exist,
    derive the naive-eq-wt 4-edge per-symbol book on THAT day's Bybit close, diff vs the prior bybit day →
    trades, charge the 5.5bp turnover (the SAME haircut as the live book + forward track, so backfill +
    forward form ONE continuous book), and compound the book MTM into nav (bybit is `PaperRepo`'s sole
    venue). The cockpit then shows a multi-year record at `/paper?venue=bybit`.

    READ-ONLY against the warehouse; only paper rows are written. IDEMPOTENT per run_date (re-runnable).
    Default INCREMENTAL: resumes after the last persisted bybit nav date; `--rebuild` recomputes the whole
    history."""
    import datetime as dt
    import logging

    from fxhnt.adapters.orchestration.assets import _data_dir
    from fxhnt.adapters.persistence.paper_repo import PaperRepo
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.bybit_paper_backfill import backfill_bybit_paper_book
    from fxhnt.application.bybit_spread_refresh import read_real_spread_by_coin
    from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo

    log = logging.getLogger(__name__)
    settings = get_settings()
    cap = settings.paper_capital if capital is None else capital
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    try:
        universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) or None
        try:
            unlock_events = load_unlock_events(operational_unlock_repo(settings),
                                               f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock sleeve omitted, not a crash
            log.warning("bybit-paper-backfill: could not load unlock calendar: %s", exc)
            unlock_events = []
        repo = PaperRepo(settings.operational_dsn)
        repo.migrate()
        # REAL per-coin measured cost (taker_fee + 0.5·real_spread) so /paper NET == the cockpit Backtest NET.
        # Empty (no spread refresh yet) → None → the flat 5.5bp fallback path.
        real_spread = read_real_spread_by_coin(repo) or None
        # INCREMENTAL (default): resume STRICTLY AFTER the last persisted bybit nav date. The per-day
        # computation depends only on strictly-prior state (positions/nav are strictly-before reads), so
        # resuming reproduces exactly what a full rebuild would for the new days. --rebuild forces the whole
        # history (first run / rare recompute).
        since_day: int | None = None
        if not rebuild:
            last_done = repo.latest_nav_run_date()
            if last_done is not None:
                since_day = (dt.date.fromisoformat(last_done) - dt.date(1970, 1, 1)).days
        n = backfill_bybit_paper_book(
            repo, store, capital=cap, at=dt.datetime.now(dt.UTC),
            universe=universe, unlock_events=unlock_events, since_day=since_day,
            real_spread_by_coin=real_spread,
            positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP)
        # PROACTIVE: re-materialize the headline nav-summary for the freshly-backfilled nav so / + /paper are
        # immediately fast (read 1 row, not the full nav). Guarded — the read-path self-heal is the safety net,
        # so a refresh failure must never fail the backfill.
        try:
            from fxhnt.application.nav_summary import refresh_nav_summary
            refresh_nav_summary(repo)
        except Exception as exc:  # noqa: BLE001
            log.warning("bybit-paper-backfill: nav-summary refresh failed: %s", exc)
    finally:
        store.close()

    scope = f"{len(universe)} liquid symbols" if universe else "full universe (no liquid subset)"
    mode = "rebuild" if rebuild else "incremental"
    typer.echo(f"bybit-paper-backfill: persisted {n} day(s) ({mode}) from {scope} "
               f"(min_dollar_vol=${min_dollar_vol:,.0f}, capital=${cap:,.0f}). "
               f"View the multi-year record at /paper?venue=bybit.")


@app.command("bybit-overlay-ab")
def bybit_overlay_ab(
    liquid: bool = typer.Option(
        True, "--liquid/--all",
        help="Restrict the crypto universe to the liquid tradeable subset (median daily turnover >= "
             "--min-dollar-vol). DEFAULT (memory-bounded). --all reads the FULL universe — OOM-risk."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) used by --liquid."),
    cost_bps: float = typer.Option(
        5.5, "--cost-bps", help="Per-turnover cost (bps) applied to each sleeve + the book."),
    train_frac: float = typer.Option(
        0.6, "--train-frac", help="Fraction of the book dates used for the TRAIN split (rest = held-out TEST)."),
) -> None:
    """READ-ONLY: OOS-disciplined overlay A/B for the Bybit 4-edge book. Sweeps a SMALL, PRINCIPLED config grid
    (naive eq-wt / inverse-vol / calibrated vol-target 15-20% / +killswitch / +marginal-gate / ISV-adaptive /
    the live legacy-tuned overlay) through the SAME live overlay primitives (reused, not reimplemented) and
    scores EACH on a TRAIN split AND a held-out TEST split.

    The live overlay HURTS the Bybit book (naive ~1.81 -> overlay ~0.98: it over-levers). This harness finds
    whether a RE-TUNED overlay ADDS value OUT-OF-SAMPLE — picking the config on TRAIN and scoring it on the
    never-looked-at TEST. A re-tuned overlay is adopted ONLY if the TRAIN winner beats naive eq-wt on TEST; a
    config that wins TRAIN but loses TEST is FLAGGED overfit (our memory: in-sample overlay tuning
    over-estimates Sharpe ~0.6). Prints the full train-vs-OOS table + a clear recommendation (deploy which
    overlay config, or STAY NAIVE EQ-WT if none beats it OOS).

    READ-ONLY — no PaperRepo, no store.write_*; the live paper_* tables are untouched. MEMORY-BOUNDED —
    --liquid (default) reads only the liquid subset, never the whole ~794-coin table."""
    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.bybit_overlay_ab import overlay_ab

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock omitted, not a crash
            log.warning("bybit-overlay-ab: could not load unlock calendar: %s", exc)
            return []

    universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if liquid else None
    try:
        rep = overlay_ab(store, universe=universe, cost_bps=cost_bps, train_frac=train_frac,
                         unlock_events=_load_unlock_events())
    finally:
        store.close()

    scope = (f"LIQUID subset (median turnover >= ${min_dollar_vol:,.0f}, {len(universe)} symbols)"
             if liquid else "ALL symbols (naive / over-diversified — OOM-risk)")
    typer.echo(f"\n=== Bybit OVERLAY A/B — OOS-disciplined (TRAIN/TEST) — READ-ONLY — {scope} ===")

    if not rep.get("available"):
        typer.echo(f"INSUFFICIENT DATA: {rep.get('reason', 'no usable book')}")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first)")
        return

    typer.echo(f"sleeves={rep['sleeves']}  days={rep['days']}  cost_bps={rep['cost_bps']}  "
               f"train_frac={rep['train_frac']}")
    typer.echo("\nConfig                  TRAIN Sharpe   TEST Sharpe   TEST CAGR   TEST maxDD   (full Sharpe)")
    typer.echo("  " + "-" * 84)
    table = rep["table"]
    for name in table:
        tr = table[name]["train"]
        te = table[name]["test"]
        fu = table[name]["full"]
        typer.echo(f"  {name:<22}{tr['sharpe']:>9.2f}     {te['sharpe']:>9.2f}    "
                   f"{te['cagr']*100:>7.1f}%   {te['max_dd']*100:>8.1f}%      {fu['sharpe']:>6.2f}")

    v = rep["verdict"]
    typer.echo(f"\nTRAIN winner:   {rep['train_winner']}  "
               f"(TEST Sharpe {table[rep['train_winner']]['test']['sharpe']:.2f})")
    typer.echo(f"Best TEST/OOS:  {rep['best_test']}  "
               f"(TEST Sharpe {table[rep['best_test']]['test']['sharpe']:.2f})")
    typer.echo(f"naive eq-wt TEST Sharpe: {v['naive_test_sharpe']:.2f}")
    typer.echo(f"train->test haircut (train winner): {v['train_to_test_haircut']:.2f}  "
               f"(memory expects ~0.6 — a large haircut = overfit)")
    if v["train_winner_overfit"]:
        typer.echo("OVERFIT FLAG: the TRAIN winner does NOT beat naive eq-wt OUT-OF-SAMPLE.")
    typer.echo(f"\nVERDICT / RECOMMENDATION: deploy '{v['recommendation']}'")
    typer.echo(f"  {v['note']}")
    typer.echo("\nNOTE: writes NOTHING. Overlay configs REUSE the live paper_book primitives "
               "(isv_adaptive_weights / inverse_vol / vol_target_leverage / DrawdownKillSwitch / "
               "marginal_gate_active); 'live_overlay' IS the deployed simulate_book path.")


@app.command("sleeve-weighting-ab")
def sleeve_weighting_ab_cmd(
    liquid: bool = typer.Option(
        True, "--liquid/--all",
        help="Restrict the crypto universe to the liquid tradeable subset (median daily turnover >= "
             "--min-dollar-vol). DEFAULT (memory-bounded). --all reads the FULL universe — OOM-risk."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Liquid-universe floor (median daily turnover, USD) used by --liquid."),
    cost_bps: float = typer.Option(
        11.0, "--cost-bps",
        help="Per-turnover round-trip cost (bps) charged to each scheme + the book. Risk-weighting adds "
             "turnover; sweep 5.5/11/22 to see if cost erases any OOS edge."),
    train_frac: float = typer.Option(
        0.6, "--train-frac",
        help="Fraction of the book dates used for the TRAIN split (rest = held-out TEST). Each scheme's "
             "hyperparam (lookback / cap) is chosen on TRAIN, scored on TEST."),
) -> None:
    """READ-ONLY: OOS-disciplined SLEEVE-WEIGHTING A/B for the Bybit 4-edge book — does risk-weighting beat
    naive equal-weight OUT-OF-SAMPLE, net of cost?

    The 4 sleeves have very different vols, so EQUAL CAPITAL != EQUAL RISK (xsfunding ~66% of book risk under
    naive 1/n). IN-SAMPLE, inverse-vol helps (full-sample Sharpe ~1.82 -> ~2.18) by reining in xsfunding —
    BUT that 2.18 is LOOK-AHEAD-OPTIMISTIC. This harness isolates ONLY the weighting question (NO leverage /
    kelly / vol-target / killswitch — those already LOST OOS in `bybit-overlay-ab`) across three CAUSAL
    schemes (weights for day t use only data <= t-1):
      * naive       — static 1/n equal weight (the current book).
      * slow_rp     — inverse-vol from a SLOW causal trailing vol (90d/180d, picked on TRAIN).
      * risk_capped — 1/n with any sleeve's RISK share capped (30%/40%, picked on TRAIN; redistributed).
    Each scheme's hyperparam is chosen on TRAIN, scored on the held-out TEST. A scheme is adopted over naive
    ONLY IF it beats naive's TEST/OOS Sharpe (net of the extra turnover) AND is positive in a majority of
    rolling windows. Otherwise STAY NAIVE — a valid, valuable verdict.

    Prints per-scheme TRAIN/TEST/FULL Sharpe + CAGR + vol + maxDD + turnover + the realized RISK composition
    (does it actually cut xsfunding's risk share?), then a PASS(switch)/STAY-NAIVE verdict. READ-ONLY — no
    PaperRepo, no store.write_*; the live paper_* tables are untouched. MEMORY-BOUNDED — --liquid reads only
    the liquid subset, never the whole ~794-coin table. Does NOT change the live book's weighting; decide on
    the verdict."""
    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.sleeve_weighting_ab import sleeve_weighting_ab

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock omitted, not a crash
            log.warning("sleeve-weighting-ab: could not load unlock calendar: %s", exc)
            return []

    universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if liquid else None
    try:
        rep = sleeve_weighting_ab(store, universe=universe, cost_bps=cost_bps, train_frac=train_frac,
                                  unlock_events=_load_unlock_events())
    finally:
        store.close()

    scope = (f"LIQUID subset (median turnover >= ${min_dollar_vol:,.0f}, {len(universe)} symbols)"
             if liquid else "ALL symbols (naive / over-diversified — OOM-risk)")
    typer.echo(f"\n=== Bybit SLEEVE-WEIGHTING A/B — OOS-disciplined (TRAIN/TEST) — READ-ONLY — {scope} ===")

    if not rep.get("available"):
        typer.echo(f"INSUFFICIENT DATA: {rep.get('reason', 'no usable book')}")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first)")
        return

    typer.echo(f"sleeves={rep['sleeves']}  days={rep['days']}  cost_bps={rep['cost_bps']}  "
               f"train_frac={rep['train_frac']}")
    typer.echo("\nScheme        param   TRAIN Sh   TEST Sh   TEST CAGR   TEST maxDD   FULL Sh   FULL vol   "
               "turnover   win%")
    typer.echo("  " + "-" * 100)
    table = rep["table"]
    for name in ("naive", "slow_rp", "risk_capped"):
        row = table[name]
        tr, te, fu = row["train"], row["test"], row["full"]
        param = "" if row["param"] is None else str(row["param"])
        typer.echo(f"  {name:<12}{param:>4}  {tr['sharpe']:>8.2f}  {te['sharpe']:>8.2f}  "
                   f"{te['cagr']*100:>8.1f}%  {te['max_dd']*100:>9.1f}%  {fu['sharpe']:>7.2f}  "
                   f"{fu['vol']*100:>7.1f}%  {row['turnover']:>8.1f}  "
                   f"{row['rolling_frac_positive']*100:>5.0f}%")

    # realized risk composition per scheme — does risk-weighting actually cut the dominant sleeve?
    typer.echo("\nRealized RISK composition (final-day causal risk shares, w*vol / Σ w*vol):")
    for name in ("naive", "slow_rp", "risk_capped"):
        comp = table[name]["risk_composition"]
        parts = "  ".join(f"{s}={comp.get(s, 0.0)*100:.0f}%" for s in rep["sleeves"])
        typer.echo(f"  {name:<12} {parts}")

    v = rep["verdict"]
    typer.echo(f"\nnaive eq-wt TEST/OOS Sharpe: {v['naive_test_sharpe']:.2f}   "
               f"(rolling-window bar: > {v['min_window_frac']*100:.0f}% positive)")
    if v["switch"]:
        typer.echo(f"\nVERDICT: PASS(switch) -> deploy '{v['recommendation']}'")
    else:
        typer.echo("\nVERDICT: STAY-NAIVE (no risk scheme beats naive OOS net of cost)")
    typer.echo(f"  {v['note']}")
    typer.echo("\nNOTE: writes NOTHING. Schemes are CAUSAL (weights for day t use only data <= t-1) and "
               "REUSE the live primitives (book_allocator._inverse_vol_weights, bybit_book_eval naive "
               "baseline). NO leverage/kelly/vol-target/killswitch (those already lost OOS). The full-sample "
               "inverse-vol Sharpe (~2.18) is LOOK-AHEAD-optimistic; only the TEST/OOS column is the verdict.")


@app.command("bybit-liquidity-sweep")
def bybit_liquidity_sweep_cmd(
    capital: float = typer.Option(
        100_000.0, "--capital",
        help="Book capital (USD). The MEASURED cost (taker fee + half the Corwin–Schultz spread) is "
             "capital-independent in bps, so capital affects only the notional, not the per-bp cost — the "
             "verdict is capital-robust. Kept for the turnover/notional reporting."),
    taker_fee_bps: float = typer.Option(
        5.5, "--taker-fee-bps",
        help="Bybit's PUBLISHED perp taker fee (bps). A KNOWN exchange fee (0.055%), not an assumption."),
    train_frac: float = typer.Option(
        0.6, "--train-frac",
        help="Fraction of the book dates used for TRAIN (rest = held-out TEST). The verdict uses TEST/OOS."),
) -> None:
    """READ-ONLY: sweep the LIQUIDITY BOUND (--min-dollar-vol) for the Bybit 4-edge book and measure the
    OPTIMAL bound NET of a cost built from ONLY measured/known quantities — NO assumed market-impact
    coefficient. Does the liquid-61 default actually beat trading broad once thin coins pay their MEASURED
    spread?

    THE MEASURED COST (no free parameters):
        cost_bps(coin) = taker_fee_bps + 0.5 · effective_spread_bps(coin)
      * taker_fee_bps = 5.5 — Bybit's PUBLISHED perp taker fee (0.055%). A KNOWN exchange fee.
      * effective_spread_bps — MEASURED from real OHLC via the Corwin–Schultz (2012) parameter-free high-low
        spread estimator (median of the daily estimates, negatives clamped to 0). You cross HALF per trade.
      * MARKET IMPACT IS DROPPED (not assumed-zero-with-a-guess): it can't be measured without live fills, and
        at this book size each per-coin order is a tiny fraction of daily volume so impact is negligible. A
        future larger book that needs an impact term must fit it from REAL fills, not a guessed coefficient.
    Thin coins have genuinely wider MEASURED spreads -> higher cost — the real, observed driver, not an
    assumption. Bonus: because the cost is capital-independent per-bp, the verdict is capital-robust.

    For each bound (0=broad/all, 5M, 10M=default/liquid-61, 25M, 50M): universe = the coins clearing it (same
    `liquid_universe` path the book uses), build the eq-wt 4-edge book's per-COIN returns (REUSING the live
    per-symbol weights `bybit_paper_weights.sleeve_weights_and_contributions`, so they reconcile with the
    edges), and report GROSS Sharpe (no cost), NET TEST Sharpe (measured cost), CAGR, vol, maxDD, turnover,
    n_coins, and the MEDIAN measured effective-spread (bps) of that bound's coins (the auditable thin-coin
    penalty). The GROSS column is the OVER-DIVERSIFICATION check (the carry showed 13.94 vs 0.63 = spurious
    thin-coin diversification). VERDICT = the bound with the best TEST/OOS NET Sharpe -> CONFIRM (the 10M
    default) / WIDEN / TIGHTEN. READ-ONLY — the live paper_* tables are untouched; does NOT change the live
    book's default bound (decide on the verdict)."""
    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity_sweep import liquidity_bound_sweep

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock omitted, not a crash
            log.warning("bybit-liquidity-sweep: could not load unlock calendar: %s", exc)
            return []

    unlock_events = _load_unlock_events()

    try:
        rep = liquidity_bound_sweep(store, capital=capital, taker_fee_bps=taker_fee_bps,
                                    train_frac=train_frac, unlock_events=unlock_events)
    finally:
        store.close()

    typer.echo(f"\n=== Bybit LIQUIDITY-BOUND sweep — OOS-disciplined — READ-ONLY — capital=${capital:,.0f} "
               f"taker_fee={taker_fee_bps}bp + 0.5·Corwin–Schultz MEASURED spread (NO assumed impact) ===")
    if not rep.get("available"):
        typer.echo(f"INSUFFICIENT DATA: {rep.get('reason', 'no usable book')}")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first — the high/low + "
                   "turnover features are needed for the measured cost model)")
        return

    typer.echo(f"\n  capital=${rep['capital']:,.0f}  taker_fee={rep['taker_fee_bps']}bp")
    typer.echo("  bound        n_coins   GROSS Sh   NET TEST Sh   TEST CAGR   TEST maxDD   turnover   "
               "cost drag   med spread")
    typer.echo("  " + "-" * 110)
    for row in rep["table"]:
        bnd = "broad/all" if row["bound"] <= 0.0 else f"${row['bound']/1e6:.0f}M"
        typer.echo(f"  {bnd:<10}{row['n_coins']:>8d}  {row['gross_sharpe']:>9.2f}  "
                   f"{row['net_test_sharpe']:>11.2f}  {row['cagr']*100:>8.1f}%  "
                   f"{row['max_dd']*100:>9.1f}%  {row['turnover']:>8.1f}  "
                   f"{row['avg_cost_drag_bps']:>7.1f}bp  {row['median_spread_bps']:>8.1f}bp")
    v = rep["verdict"]
    rec = "broad/all" if v["recommendation"] <= 0.0 else f"${v['recommendation']/1e6:.0f}M"
    typer.echo(f"  -> {v['direction']} (best OOS NET bound = {rec}; "
               f"NET TEST Sharpe {v['best_net_test_sharpe']:.2f} vs $10M default "
               f"{v['default_net_test_sharpe']:.2f})"
               + ("  [OVER-DIVERSIFICATION FLAG]" if v["over_diversification_flag"] else ""))

    typer.echo(f"\nVERDICT ({v['direction']}): {v['note']}")
    typer.echo("\nNOTE: writes NOTHING. There are NO trading assumptions — every cost input is a PUBLISHED "
               "fee (Bybit 5.5bp taker) or a DATA-MEASURED spread (Corwin–Schultz, parameter-free). Market "
               "impact is DROPPED, not guessed. The cost is capital-independent per-bp, so the verdict is "
               "capital-robust. The GROSS column is the over-diversification check; the NET TEST column + the "
               "median measured spread are the real, auditable verdict. Per-coin weights REUSE the live book "
               "(bybit_paper_weights). Does NOT change the live book's --min-dollar-vol default.")


@app.command("maker-recost")
def maker_recost_cmd(
    capital: float = typer.Option(
        100_000.0, "--capital",
        help="Book capital (USD). The cost is capital-independent in bps, so capital sets only the notional — "
             "the net-Sharpe lift is capital-robust."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Restrict to the liquid Bybit subset (median daily $-turnover >= this). DEFAULT 10M = the "
             "deployed liquid-61 book. --min-dollar-vol 0 reads the FULL universe (OOM-risk, broad book)."),
    taker_fee_bps: float = typer.Option(
        5.5, "--taker-fee-bps",
        help="Bybit's PUBLISHED perp TAKER fee (bps, 0.055%). The all-taker leg crosses the spread."),
    maker_fee_bps: float = typer.Option(
        2.0, "--maker-fee-bps",
        help="Bybit's PUBLISHED non-VIP perp MAKER fee (bps, 0.02%). The maker leg does NOT cross the spread."),
    real_spread: bool = typer.Option(
        False, "--real-spread/--no-real-spread",
        help="ALSO re-cost the grid with the REAL live L1 quoted spread (fetched from api.bybit.com tickers) "
             "in place of the high/low Corwin–Schultz estimate, and print BOTH grids side by side. Needs "
             "egress to api.bybit.com (the live quotes) + postgres (the book). The CS grid is a worst-case "
             "upper bound on cost; the REAL grid is the forward-realistic number (a CURRENT snapshot)."),
) -> None:
    """READ-ONLY: maker-vs-taker EXECUTION re-costing of the Bybit 4-edge deploy book — the net-Sharpe lift
    from filling rebalances PASSIVELY (maker) instead of crossing the spread (taker).

    THIS IS NOT A SIGNAL CHANGE AND NOT A FIT. The GROSS book return is UNCHANGED across the whole grid — the
    identical gross series + per-day per-coin turnover are reused from `per_coin_book_returns`; ONLY the
    per-turnover COST model varies. Nothing to overfit; a structural execution what-if.

    THE COST MODEL (per rebalance, per coin, |Δw| split by maker fill fraction f):
      * TAKER leg (1−f): cost = taker_fee(5.5) + 0.5·spread   ← crosses the spread (the CURRENT live model)
      * MAKER leg (f):   cost = maker_fee(2.0) + adverse_sel·spread   ← does NOT cross; adverse_sel ∈ [0,0.5]
        (0 optimistic = earns the spread, 0.25 conservative = quarter-spread to adverse selection, 0.5 = no
        spread edge over taker). Blended = f·maker + (1−f)·taker.

    Sweeps fill fraction f ∈ {0.0 (=ALL-TAKER BASELINE), 0.3, 0.5, 0.7, 1.0} × adverse_sel ∈ {0.0, 0.25} and
    reports per-cell net Sharpe / total return / CAGR / maxDD / cost-drag and the Δ Sharpe + Δ cost-drag vs
    the f=0 baseline. f=0 reproduces the currently-deployed all-taker number EXACTLY (sanity tie-out).
    READ-ONLY — the live book is untouched; this does NOT deploy a maker mode."""
    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.bybit_maker_recost import maker_recost_sweep

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock omitted, not a crash
            log.warning("maker-recost: could not load unlock calendar: %s", exc)
            return []

    unlock_events = _load_unlock_events()

    try:
        universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if min_dollar_vol > 0 else None
        real_spread_by_coin = None
        if real_spread:
            # The FORWARD-realistic cost input: the CURRENT live L1 quoted spread (one tickers call). Needs
            # egress to api.bybit.com. A fetch failure is non-fatal — fall back to the CS grid only.
            from fxhnt.adapters.data.bybit_true_spread import BybitTrueSpread
            try:
                real_spread_by_coin = BybitTrueSpread().real_spread_bps(universe=universe)
            except Exception as exc:  # noqa: BLE001 — no live quotes → CS grid only, not a crash
                log.warning("maker-recost: could not fetch live L1 spreads (CS grid only): %s", exc)
        rep = maker_recost_sweep(store, universe=universe, capital=capital, taker_fee_bps=taker_fee_bps,
                                 maker_fee_bps=maker_fee_bps, unlock_events=unlock_events,
                                 real_spread_by_coin=real_spread_by_coin)
    finally:
        store.close()

    scope = (f"liquid (median turnover >= ${min_dollar_vol:,.0f})" if min_dollar_vol > 0 else "broad/all")
    typer.echo(f"\n=== Bybit MAKER-vs-TAKER execution re-costing — READ-ONLY — {scope} — "
               f"capital=${capital:,.0f}  taker={taker_fee_bps}bp  maker={maker_fee_bps}bp ===")
    if not rep.get("available"):
        typer.echo(f"INSUFFICIENT DATA: {rep.get('reason', 'no usable book')}")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first)")
        return

    def _print_grid(grid: dict, *, label: str) -> None:
        typer.echo(f"\n  --- {label} grid ---")
        typer.echo("  fill_f   adverse   NET Sh   tot ret    CAGR    maxDD   cost drag   ΔSharpe   Δcost")
        typer.echo("  " + "-" * 96)
        for row in grid["table"]:
            tag = "  (BASELINE all-taker)" if row["is_baseline"] and row["adverse_sel"] == 0.0 else ""
            typer.echo(f"  {row['fill_fraction']:>5.1f}  {row['adverse_sel']:>6.2f}  "
                       f"{row['sharpe']:>6.2f}  {row['total_return']*100:>7.1f}%  {row['cagr']*100:>6.1f}%  "
                       f"{row['max_dd']*100:>6.1f}%  {row['cost_drag_bps']:>7.1f}bp  "
                       f"{row['delta_sharpe']:>+7.2f}  {row['delta_cost_drag_bps']:>+6.1f}bp{tag}")
        b = grid["baseline"]
        op = grid["operating_point"]
        typer.echo(f"  BASELINE (f=0 all-taker): Sharpe {b['sharpe']:.2f}  tot ret {b['total_return']*100:.1f}%"
                   f"  cost drag {b['cost_drag_bps']:.1f}bp")
        typer.echo(f"  OPERATING POINT (f={op['fill_fraction']:.1f}, adverse={op['adverse_sel']:.2f}): "
                   f"Sharpe {op['sharpe']:.2f} (ΔSharpe {op['delta_sharpe']:+.2f})  "
                   f"cost drag {op['cost_drag_bps']:.1f}bp (Δ {op['delta_cost_drag_bps']:+.1f}bp)")

    typer.echo(f"\n  n_coins={rep['n_coins']}  days={rep['days']}  median per-coin CS spread="
               f"{rep['median_spread_bps']:.1f}bp  avg daily turnover={rep['avg_daily_turnover']:.3f}")
    typer.echo("  (the maker leg saves crossing that spread; turnover is the fraction of book re-traded/day)")
    typer.echo("\n  GROSS book return is UNCHANGED across every row — only the cost model varies.")

    _print_grid(rep, label="CS spread (high/low Corwin–Schultz — worst-case UPPER BOUND on cost)")

    if "real" in rep:
        r = rep["real"]
        _print_grid(r, label="REAL spread (live L1 quote — FORWARD-realistic)")
        ratio = r["cs_over_real_ratio"]
        ratio_s = "inf" if ratio == float("inf") else f"{ratio:.1f}x"
        typer.echo(f"\n  SPREAD COMPARISON over the {r['n_coins']} book coins: median REAL quoted spread "
                   f"{r['median_real_spread_bps']:.1f}bp (mean {r['mean_real_spread_bps']:.1f}bp) vs median CS "
                   f"high/low {r['median_cs_spread_bps']:.1f}bp → CS over-states by {ratio_s}. "
                   f"{r['n_real']}/{r['n_coins']} coins had a live quote; {r['n_fallback']} fell back to CS.")
        typer.echo("\n  CAVEAT: the REAL spread is a CURRENT snapshot applied across ALL history — it is the "
                   "FORWARD-realistic cost (what we'd pay deploying NOW), NOT the 2021–22 cost (spreads were "
                   "wider then). So the REAL grid is the forward-relevant number; the CS grid is a worst-case "
                   "UPPER BOUND on cost.")
    elif real_spread:
        typer.echo("\n  (REAL spread requested but no live quotes were available — CS grid only.)")

    typer.echo("\nNOTE: writes NOTHING; the live book is untouched. The GROSS return is identical in every "
               "scenario — this is an EXECUTION re-costing, not a signal change or a fit (nothing to overfit). "
               "f=0 reproduces the deployed all-taker number EXACTLY. maker fee + adverse-selection are the "
               "only honesty knobs; impact is dropped (capital-robust per-bp cost). Does NOT deploy maker mode.")


@app.command("bybit-true-spread")
def bybit_true_spread_cmd(
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Restrict to the liquid Bybit subset (median daily $-turnover >= this). DEFAULT 10M = the "
             "deployed liquid-61 book. --min-dollar-vol 0 = the FULL universe (broad book)."),
    taker_fee_bps: float = typer.Option(
        5.5, "--taker-fee-bps",
        help="Bybit's PUBLISHED perp TAKER fee (bps). Used only to build the book (unchanged)."),
) -> None:
    """READ-ONLY: measure the TRUE Bybit L1 quoted spread (live bid/ask from api.bybit.com tickers) and compare
    it to the high/low Corwin–Schultz estimate the cost model currently uses — for the liquid-61 book coins.

    Answers "is the 144bp CS proxy real?" The CS estimate (Corwin–Schultz on DAILY HIGH/LOW) captures the
    intraday RANGE, not the quoted bid/ask, so on liquid coins it over-states the true L1 spread (BTC/ETH quote
    spreads ≈ 1bp). This prints, over the coins the live book trades: median/mean REAL quoted spread vs the
    median CS estimate, and the over-estimate ratio (median CS / median real). Needs egress to api.bybit.com
    (the live quotes) + postgres (the book). Writes NOTHING — the live book is untouched."""
    import logging

    from fxhnt.adapters.data.bybit_true_spread import BybitTrueSpread
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.bybit_maker_recost import true_spread_report

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock omitted, not a crash
            log.warning("bybit-true-spread: could not load unlock calendar: %s", exc)
            return []

    unlock_events = _load_unlock_events()

    try:
        universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if min_dollar_vol > 0 else None
        real_spread_by_coin = BybitTrueSpread().real_spread_bps(universe=universe)
        rep = true_spread_report(store, universe=universe, real_spread_by_coin=real_spread_by_coin,
                                 taker_fee_bps=taker_fee_bps, unlock_events=unlock_events)
    finally:
        store.close()

    scope = (f"liquid (median turnover >= ${min_dollar_vol:,.0f})" if min_dollar_vol > 0 else "broad/all")
    typer.echo(f"\n=== Bybit TRUE L1 quoted spread vs high/low Corwin–Schultz — READ-ONLY — {scope} ===")
    if not rep.get("available"):
        typer.echo(f"INSUFFICIENT DATA: {rep.get('reason', 'no usable book')}")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first)")
        return

    ratio = rep["cs_over_real_ratio"]
    ratio_s = "inf" if ratio == float("inf") else f"{ratio:.1f}x"
    typer.echo(f"\n  book coins={rep['n_coins']}  with live quote={rep['n_real']}  fell back to CS="
               f"{rep['n_fallback']}")
    typer.echo(f"  median REAL quoted spread : {rep['median_real_spread_bps']:.2f}bp  "
               f"(mean {rep['mean_real_spread_bps']:.2f}bp)")
    typer.echo(f"  median CS high/low spread : {rep['median_cs_spread_bps']:.1f}bp  "
               f"(mean {rep['mean_cs_spread_bps']:.1f}bp)")
    typer.echo(f"  -> CS over-states the true quoted spread by {ratio_s}")
    typer.echo("\nVERDICT: the high/low Corwin–Schultz estimate captures intraday RANGE, not the quoted bid/ask"
               ", so it is a worst-case UPPER BOUND on cost. The REAL quoted spread is the forward-realistic "
               "cost input. NOTE: it is a CURRENT snapshot — spreads were wider in 2021–22, so this is the "
               "FORWARD cost, not the historical one. Writes NOTHING.")


@app.command("spread-factor-test")
def spread_factor_test_cmd(
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Restrict the universe to the liquid Bybit subset (median daily $-turnover >= this). DEFAULT "
             "(memory-bounded). --min-dollar-vol 0 reads the FULL survivorship-free universe (OOM-risk)."),
    train_frac: float = typer.Option(
        0.6, "--train-frac",
        help="Fraction of the long-short dates used for TRAIN (rest = held-out TEST). Verdict uses TEST/OOS."),
    taker_fee_bps: float = typer.Option(
        5.5, "--taker-fee-bps",
        help="Bybit's PUBLISHED perp taker fee (bps). The NET cost = fee + half the coin's measured CS spread."),
    cs_window: int = typer.Option(
        30, "--cs-window", help="Trailing-window length (days) for the rolling Corwin–Schultz spread."),
) -> None:
    """READ-ONLY: STEP 1 of the spread-as-a-strat research — build the survivorship-free LIQUIDITY PANEL (CS
    spread + Amihud illiquidity + close-to-close return per coin per day) and FALSIFY the STATIC ILLIQUIDITY
    PREMIUM, NET of measured cost.

    FALSIFICATION-FIRST: a clean "the premium dies net of cost" (FALSIFY) is the EXPECTED, VALUABLE outcome —
    it RULES the static illiquidity premium OUT (the cost trap). Each day, rank the eligible universe by the
    trailing CS spread (and separately by Amihud) into quintiles, form Q5 (most illiquid) − Q1 (most liquid)
    with NO lookahead (sort on info up to t, hold t+1), pick the DIRECTION in-sample, and charge the MEASURED
    per-coin spread cost (Bybit taker fee + half the coin's Corwin–Schultz spread) on the rebalance turnover.
    Reports GROSS + NET Sharpe / total return, the OOS (TEST) NET Sharpe, the NON-OVERLAPPING t-stat, maxDD,
    turnover, avg cost drag, and the correlation of the NET series to the 4 live edges (tstrend/unlock/
    xsfunding/positioning). VERDICT per sort key: CONFIRM only if NET-OOS Sharpe is meaningfully positive AND
    the t-stat holds AND it is not a rebrand of a live edge (low |corr|); else FALSIFY. Writes NOTHING; the
    live paper_* tables are untouched."""
    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_book_eval import _DEFAULT_BYBIT_SLEEVES, sleeve_returns_from_store
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.spread_liquidity_factor import (
        build_liquidity_panel,
        static_illiquidity_premium,
    )

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock edge omitted, not a crash
            log.warning("spread-factor-test: could not load unlock calendar: %s", exc)
            return []

    try:
        universe = None if min_dollar_vol <= 0.0 else liquid_universe(store, min_dollar_vol=min_dollar_vol)
        panel = build_liquidity_panel(store, universe=universe, cs_window=cs_window)
        # the live-edge daily return series (GROSS) to correlate the NET premium against (the rebrand check).
        unlock_events = _load_unlock_events()
        edge_series: dict[str, dict[int, float]] = {}
        for sleeve in _DEFAULT_BYBIT_SLEEVES:
            edge = sleeve.replace("crypto_", "")
            s = sleeve_returns_from_store(store, sleeve, universe=universe, cost_bps=0.0,
                                          unlock_events=unlock_events)
            if s:
                edge_series[edge] = s
        rep = static_illiquidity_premium(panel, train_frac=train_frac, taker_fee_bps=taker_fee_bps,
                                         edge_series=edge_series or None)
    finally:
        store.close()

    bnd = "broad/all" if (universe is None) else f"${min_dollar_vol/1e6:.0f}M ({len(universe)} coins)"
    typer.echo(f"\n=== Bybit STATIC ILLIQUIDITY-PREMIUM falsification — READ-ONLY — universe={bnd} "
               f"taker_fee={taker_fee_bps}bp + 0.5·Corwin–Schultz MEASURED spread ===")
    if not rep.get("available"):
        typer.echo(f"INSUFFICIENT DATA: {rep.get('reason', 'no usable liquidity panel')}")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first — high/low + "
                   "turnover features are needed for the CS spread + Amihud panel)")
        return

    # panel summary.
    days = sorted(panel)
    n_days = len(days)
    avg_coins = (sum(len(panel[d]) for d in days) / n_days) if n_days else 0.0
    all_spreads = [v[0] for d in days for v in panel[d].values()]
    all_amihud = [v[1] for d in days for v in panel[d].values() if v[1] is not None]
    import statistics as _st
    med_spread = _st.median(all_spreads) if all_spreads else 0.0
    med_amihud = _st.median(all_amihud) if all_amihud else 0.0
    typer.echo(f"\nPANEL: {n_days} days, ~{avg_coins:.0f} coins/day, median CS spread {med_spread:.1f}bp, "
               f"median Amihud (×1e9) {med_amihud:.3f}")

    typer.echo("\n  sort by    dir   gross Sh   NET-OOS Sh    t-stat    NET maxDD   cost drag   "
               "max|corr edge|   verdict")
    typer.echo("  " + "-" * 108)
    for key in ("spread", "amihud"):
        r = rep[key]
        d_arrow = "Q5-Q1" if r.get("direction_sign", 1) == 1 else "Q1-Q5"
        typer.echo(f"  {key:<9}{d_arrow:>6}  {r['gross_full_sharpe']:>8.2f}  {r['net_test_sharpe']:>10.2f}  "
                   f"{r['t_stat']:>8.2f}  {r['max_dd']*100:>9.1f}%  {r['cost_drag_bps']:>8.1f}bp  "
                   f"{r.get('max_abs_corr', 0.0):>13.2f}   {r['verdict']}")
        corr = r.get("corr_to_edges", {})
        if corr:
            typer.echo("             corr-to-edges: "
                       + ", ".join(f"{name}={c:+.2f}" for name, c in sorted(corr.items())))

    sp, am = rep["spread"], rep["amihud"]
    confirmed = [k for k in ("spread", "amihud") if rep[k]["verdict"] == "CONFIRM"]
    if confirmed:
        typer.echo(f"\nVERDICT: CONFIRM on {confirmed} — NET-OOS Sharpe meaningfully positive, t-stat holds, "
                   "and NOT a rebrand of a live edge. Promote to a forward track.")
    else:
        typer.echo("\nVERDICT: FALSIFY (the EXPECTED, valuable outcome). The static illiquidity premium does "
                   "NOT survive net of the measured spread it must harvest — a COST TRAP. "
                   f"spread: gross Sh {sp['gross_full_sharpe']:.2f} -> NET-OOS {sp['net_test_sharpe']:.2f} "
                   f"(cost drag {sp['cost_drag_bps']:.0f}bp); "
                   f"amihud: gross Sh {am['gross_full_sharpe']:.2f} -> NET-OOS {am['net_test_sharpe']:.2f} "
                   f"(cost drag {am['cost_drag_bps']:.0f}bp). Rules it OUT as a tradeable crypto edge.")
    typer.echo("\nLARGE+FREQUENT+CONCENTRATED: a tradeable edge must move enough per trade to clear the spread "
               "(LARGE), recur often enough to compound (FREQUENT), and concentrate in a few names you can "
               "actually size (CONCENTRATED). The illiquidity premium fails LARGE — the per-name move is "
               "smaller than the spread you cross to capture it.")
    typer.echo("\nNOTE: writes NOTHING. Cost = Bybit's published 5.5bp taker + half the coin's MEASURED "
               "Corwin–Schultz spread (NO assumed impact). NET-OOS Sharpe + t-stat + corr-to-edges are the "
               "verdict. A FALSIFY rules the static premium out; STEP 2+ would test conditional / "
               "time-varying liquidity instead.")


@app.command("spread-deterioration-test")
def spread_deterioration_test_cmd(
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol",
        help="Restrict the universe to the liquid Bybit subset (median daily $-turnover >= this). DEFAULT "
             "(memory-bounded). --min-dollar-vol 0 reads the FULL survivorship-free universe (OOM-risk)."),
    holds: str = typer.Option(
        "5,10,20", "--holds", help="Comma-separated hold horizons T (days) to sweep."),
    det_window: int = typer.Option(
        20, "--det-window", help="Trailing window (days) for the per-coin spread/Amihud deterioration z-score."),
    max_entry_spread_bps: float = typer.Option(
        60.0, "--max-entry-spread-bps",
        help="Tradeable-at-entry CAP: only short coins whose entry CS spread is BELOW this (the cost-escape)."),
    min_entry_dollar_vol: float = typer.Option(
        5_000_000.0, "--min-entry-dollar-vol",
        help="Tradeable-at-entry FLOOR: only short coins whose entry dollar-volume is ABOVE this."),
    n_buckets: int = typer.Option(
        10, "--n-buckets", help="Short the top 1/n_buckets most-deteriorating coins (decile = 10)."),
    taker_fee_bps: float = typer.Option(
        5.5, "--taker-fee-bps",
        help="Bybit's PUBLISHED perp taker fee (bps). Charged per leg at BOTH entry and exit."),
    train_frac: float = typer.Option(
        0.6, "--train-frac",
        help="Fraction of the non-overlapping periods used for TRAIN (rest = held-out TEST). Verdict uses TEST."),
) -> None:
    """READ-ONLY: STEP 2 of the spread-as-a-strat research — test SPREAD-DETERIORATION SHORTS (the
    "abandonment" edge), NET of the REAL entry + exit spread cost.

    Step 1 ruled the STATIC illiquidity premium OUT (you cannot HOLD illiquid coins — the spread dwarfs the
    move). Step 2 ESCAPES that trap: SHORT coins as their liquidity STARTS deteriorating (z-score of the
    coin's own CS spread widening + Amihud rising), while they are STILL tradeable at entry (entry spread
    below a cap AND dollar-volume above a floor — the cost-escape), LONG the matched liquid-universe mean
    (market-neutral), hold T days on a NON-OVERLAPPING grid. The REAL cost is the MEASURED spread at BOTH
    ENTRY (t) AND EXIT (t+T) + the taker fee per leg — because the spread WIDENS during the hold, the exit is
    costlier, and the abandonment move must clear BOTH legs NET. Reports, per T: GROSS + NET-OOS Sharpe,
    non-overlapping t-stat, maxDD, avg entry vs exit cost drag, #names/period, corr of the NET series to the 4
    live edges (especially unlock), and the verdict. CONFIRM only if NET-OOS Sharpe >= 0.5 AND |t| >= 2 AND
    max|corr-to-edge| < 0.6 AND it passes LARGE+FREQUENT+CONCENTRATED; else FALSIFY. Writes NOTHING."""
    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_book_eval import _DEFAULT_BYBIT_SLEEVES, sleeve_returns_from_store
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.spread_liquidity_factor import (
        build_deterioration_panel,
        spread_deterioration_sweep,
    )

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    hold_horizons = tuple(int(x) for x in holds.split(",") if x.strip())

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock edge omitted, not a crash
            log.warning("spread-deterioration-test: could not load unlock calendar: %s", exc)
            return []

    try:
        universe = None if min_dollar_vol <= 0.0 else liquid_universe(store, min_dollar_vol=min_dollar_vol)
        panel = build_deterioration_panel(store, universe=universe, cs_window=det_window)
        unlock_events = _load_unlock_events()
        edge_series: dict[str, dict[int, float]] = {}
        for sleeve in _DEFAULT_BYBIT_SLEEVES:
            edge = sleeve.replace("crypto_", "")
            s = sleeve_returns_from_store(store, sleeve, universe=universe, cost_bps=0.0,
                                          unlock_events=unlock_events)
            if s:
                edge_series[edge] = s
        rep = spread_deterioration_sweep(
            panel, holds=hold_horizons, det_window=det_window,
            max_entry_spread_bps=max_entry_spread_bps, min_entry_dollar_vol=min_entry_dollar_vol,
            n_buckets=n_buckets, taker_fee_bps=taker_fee_bps, train_frac=train_frac,
            edge_series=edge_series or None)
    finally:
        store.close()

    bnd = "broad/all" if (universe is None) else f"${min_dollar_vol/1e6:.0f}M ({len(universe)} coins)"
    typer.echo(f"\n=== Bybit SPREAD-DETERIORATION SHORTS (abandonment) — READ-ONLY — universe={bnd} "
               f"taker_fee={taker_fee_bps}bp + MEASURED CS spread at ENTRY *and* EXIT ===")
    if not rep.get("available") or not panel:
        typer.echo("INSUFFICIENT DATA: no usable deterioration panel")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first — high/low + "
                   "turnover features are needed for the CS spread + Amihud + dollar-volume panel)")
        return

    # panel summary.
    import statistics as _st
    days = sorted(panel)
    n_days = len(days)
    avg_coins = (sum(len(panel[d]) for d in days) / n_days) if n_days else 0.0
    all_spreads = [v[0] for d in days for v in panel[d].values()]
    med_spread = _st.median(all_spreads) if all_spreads else 0.0
    typer.echo(f"\nPANEL: {n_days} days, ~{avg_coins:.0f} coins/day, median CS spread {med_spread:.1f}bp; "
               f"deterioration window {det_window}d; tradeable-at-entry: spread < {max_entry_spread_bps:.0f}bp "
               f"AND $vol > ${min_entry_dollar_vol/1e6:.0f}M; SHORT top-1/{n_buckets} most-deteriorating.")

    typer.echo("\n   T   gross Sh   NET-OOS Sh    t-stat    NET maxDD   entry drag   exit drag   "
               "#names/d   max|corr edge|   verdict")
    typer.echo("  " + "-" * 118)
    for T in hold_horizons:
        r = rep["by_T"].get(T, {})
        if not r.get("available"):
            typer.echo(f"  {T:>3}   (insufficient periods: {r.get('reason', 'n/a')})")
            continue
        typer.echo(f"  {T:>3}   {r['gross_full_sharpe']:>8.2f}  {r['net_test_sharpe']:>10.2f}  "
                   f"{r['t_stat']:>8.2f}  {r['max_dd']*100:>9.1f}%  {r['avg_entry_drag_bps']:>9.1f}bp  "
                   f"{r['avg_exit_drag_bps']:>8.1f}bp  {r['avg_names_per_period']:>8.1f}   "
                   f"{r.get('max_abs_corr', 0.0):>13.2f}   {r['verdict']}")
        corr = r.get("corr_to_edges", {})
        if corr:
            typer.echo("        corr-to-edges: "
                       + ", ".join(f"{name}={c:+.2f}" for name, c in sorted(corr.items()))
                       + f"  |  LARGE={r['large']} FREQUENT={r['frequent']} CONCENTRATED={r['concentrated']}"
                       + f" ({r['events_per_year']:.0f} events/yr)")

    confirmed = [T for T in hold_horizons
                 if rep["by_T"].get(T, {}).get("available") and rep["by_T"][T]["verdict"] == "CONFIRM"]
    if confirmed:
        typer.echo(f"\nVERDICT: CONFIRM on T={confirmed} — the abandonment move clears the round-trip spread "
                   "NET-OOS, the t-stat holds, and it is NOT a rebrand of a live edge. Promote to a forward "
                   "track.")
    else:
        typer.echo("\nVERDICT: FALSIFY (the EXPECTED, valuable outcome). Shorting deteriorating coins does NOT "
                   "clear the REAL round-trip spread (entry + the WIDENED exit) NET-OOS — the abandonment move "
                   "is too small, too rare, or a rebrand of a live edge. The liquidity→abandonment thesis does "
                   "not survive measured cost on this universe.")
    typer.echo("\nLARGE+FREQUENT+CONCENTRATED: a tradeable edge must move enough per trade to clear the spread "
               "paid at BOTH entry and the widened exit (LARGE), recur often enough to compound (FREQUENT), "
               "and concentrate in sizeable names you can actually short (CONCENTRATED).")
    typer.echo("\nNOTE: writes NOTHING. Cost = Bybit's published 5.5bp taker + half the coin's MEASURED "
               "Corwin–Schultz spread, charged at ENTRY (t) AND EXIT (t+T) — the spread widens during the "
               "hold, so the exit is costlier. NET-OOS Sharpe + non-overlapping t-stat + corr-to-edges are "
               "the verdict.")


@app.command("spread-overlay-test")
def spread_overlay_test_cmd(
    min_dollar_vols: str = typer.Option(
        "10000000,1000000,0", "--min-dollar-vols",
        help="Comma-separated --min-dollar-vol universes to sweep (0 = broad/all). DEFAULT = the live $10M "
             "liquid-61 bound, a broad $1M, and 0/broad — the breadth check."),
    sleeves: str = typer.Option(
        "", "--sleeves",
        help="Comma-separated book sleeves (default = the 4-edge book: tstrend,unlock,xsfunding,positioning)."),
    train_frac: float = typer.Option(
        0.6, "--train-frac",
        help="Fraction of the book dates used for TRAIN (k/λ selected on TRAIN; verdict uses held-out TEST)."),
    taker_fee_bps: float = typer.Option(
        5.5, "--taker-fee-bps",
        help="Bybit's PUBLISHED perp taker fee (bps). NET cost = fee + half the coin's measured CS spread."),
) -> None:
    """READ-ONLY: STEP 3 (final) of the spread-as-a-strat research — the SPREAD-AWARE EXECUTION OVERLAY. Use
    the MEASURED spread to AVOID cost (not to bet on it) and measure whether it lifts the EXISTING Bybit 4-edge
    book's NET-OOS Sharpe. A SHADOW eval — it NEVER changes the live book's returns.

    Steps 1-2 falsified both DIRECTIONAL spread ideas (static premium = cost trap on entry/holding;
    deterioration shorts = cost trap on the WIDENED exit). The spread is a COST signal, not alpha. Step 3 tests
    the only remaining, NON-directional value: reduce the cost of the edges we already run. Two mechanisms vs
    the baseline, NET of the MEASURED per-coin cost (Bybit 5.5bp taker + half the Corwin–Schultz spread):
      (A) BASELINE — the current naive eq-wt book.
      (B) SPREAD-AWARE SIZING — scale weight by 1/(1+k·spread_bps) then renormalise (gross preserved); sweep k.
      (C) NO-TRADE BANDS — only rebalance coin i when |Δw_i| > a cost-proportional band λ·cost_i/1e4; sweep λ.
      (B+C) combined. k/λ are SELECTED on TRAIN; the held-out TEST NET Sharpe is the verdict.

    HONESTY + BREADTH: the live book is ALREADY liquidity-bounded ($10M / liquid-61), so it is already
    cost-optimised — the lift may be MODEST. The same overlay is run on a BROADER universe ($1M, 0/broad) where
    it should help MORE — quantifying how much cost the $10M bound ALREADY saves and whether the overlay is
    worth wiring into the live book. Per universe: CONFIRM (lift >= +0.10 Sharpe — wire it in) / MARGINAL (a
    cost-drag reduction but small Sharpe lift — the bound likely already captures it) / NO-HELP. Writes
    NOTHING."""
    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.spread_overlay import spread_overlay_test

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    bounds = tuple(float(x) for x in min_dollar_vols.split(",") if x.strip())
    sleeve_list = [s.strip() for s in sleeves.split(",") if s.strip()] or None

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock edge omitted, not a crash
            log.warning("spread-overlay-test: could not load unlock calendar: %s", exc)
            return []

    try:
        unlock_events = _load_unlock_events()
        rep = spread_overlay_test(store, min_dollar_vols=bounds, train_frac=train_frac,
                                  taker_fee_bps=taker_fee_bps, sleeves=sleeve_list,
                                  unlock_events=unlock_events)
    finally:
        store.close()

    typer.echo(f"\n=== Bybit SPREAD-AWARE EXECUTION OVERLAY (Step 3) — READ-ONLY — SHADOW eval — "
               f"taker_fee={taker_fee_bps}bp + 0.5·Corwin–Schultz MEASURED spread ===")
    if not rep.get("available"):
        typer.echo(f"INSUFFICIENT DATA: {rep.get('reason', 'no usable book')}")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first — the high/low + "
                   "turnover features are needed for the measured cost model)")
        return

    typer.echo("\n  universe    variant    n_coins   NET-OOS Sh   cost drag    turnover    Δ Sh vs base   "
               "Δ cost drag   verdict")
    typer.echo("  " + "-" * 118)
    for row in rep["table"]:
        ulabel = row["universe_label"] + ("*" if row.get("is_live_bound") else "")
        base = row["baseline"]
        # baseline row, then the three overlay variants.
        typer.echo(f"  {ulabel:<10}  {'baseline':<9}  {row['n_coins']:>6d}   {base['net_test_sharpe']:>9.2f}   "
                   f"{base['cost_drag_bps']:>8.1f}bp  {base['turnover']:>9.1f}   {'—':>11}   {'—':>11}")
        for vname in ("sizing", "band", "combined"):
            v = row[vname]
            d_sh = v["net_test_sharpe"] - base["net_test_sharpe"]
            d_cd = base["cost_drag_bps"] - v["cost_drag_bps"]
            param = (f"k={v['k']:g}" if vname == "sizing"
                     else f"λ={v['lam']:g}" if vname == "band"
                     else f"k={v['k']:g},λ={v['lam']:g}")
            typer.echo(f"  {'':<10}  {vname:<9}  {param:>6}   {v['net_test_sharpe']:>9.2f}   "
                       f"{v['cost_drag_bps']:>8.1f}bp  {v['turnover']:>9.1f}   {d_sh:>+11.2f}   "
                       f"{d_cd:>+9.1f}bp")
        ver = row["verdict"]
        typer.echo(f"  {'':<10}  -> {ver['verdict']} via {ver['best_variant']} "
                   f"(best NET-OOS {ver['best_net_test_sharpe']:.2f} vs baseline "
                   f"{ver['baseline_net_test_sharpe']:.2f}; Δ {ver['d_sharpe']:+.2f} Sh, "
                   f"{ver['d_cost_drag_bps']:+.1f}bp cost-drag, {ver['d_turnover']:+.1f} turnover)")
        typer.echo("")

    confirmed = [r["universe_label"] for r in rep["table"] if r["verdict"]["verdict"] == "CONFIRM"]
    live = next((r for r in rep["table"] if r.get("is_live_bound")), None)
    if live is not None:
        lv = live["verdict"]
        typer.echo(f"LIVE BOOK ($10M / liquid-61): {lv['verdict']} — the overlay's best NET-OOS lift on the "
                   f"already-liquidity-bounded live book is {lv['d_sharpe']:+.2f} Sharpe "
                   f"({lv['d_cost_drag_bps']:+.1f}bp cost-drag). "
                   + ("WORTH WIRING IN." if lv["verdict"] == "CONFIRM"
                      else "The $10M bound already captures most of the gain." if lv["verdict"] == "MARGINAL"
                      else "No help on the bounded book."))
    if confirmed:
        typer.echo(f"\nVERDICT: CONFIRM on {confirmed} — the spread-aware overlay lifts NET-OOS Sharpe by "
                   f">= +{0.10:.2f} there. The lift grows with universe breadth (more thin coins to avoid).")
    else:
        typer.echo("\nVERDICT: no universe clears the +0.10 NET-OOS Sharpe bar — the overlay is at best a "
                   "MARGINAL cost-drag/turnover trimmer. On the liquid-61 live book the $10M bound already "
                   "captures the cost gain; the overlay's value scales with breadth, so it matters only if the "
                   "live universe is broadened.")
    typer.echo("\nNOTE: writes NOTHING — the live book's returns are untouched (SHADOW eval). The overlay uses "
               "the MEASURED spread to AVOID cost, never to bet on it (Steps 1-2 falsified the directional "
               "spread ideas). k/λ are selected on TRAIN; the verdict uses the held-out TEST. Per-coin weights "
               "REUSE the live book (bybit_paper_weights). Cost = Bybit's published 5.5bp taker + half the "
               "Corwin–Schultz MEASURED spread (NO assumed impact).")


@app.command("broaden-overlay-validate")
def broaden_overlay_validate_cmd(
    live_bound: float = typer.Option(
        10_000_000.0, "--live-bound",
        help="The LIVE book's --min-dollar-vol (the current liquid-61 deploy: $10M)."),
    min_dollar_vols: str = typer.Option(
        "5000000,1000000,500000", "--min-dollar-vols",
        help="Comma-separated BROADER --min-dollar-vol universes to consider for (B) (0 = broad/all). Each is "
             "selected-against on each fold's TRAIN; the per-fold winner is tested. DEFAULT = $5M/$1M/$500k."),
    n_folds: int = typer.Option(
        4, "--n-folds", help="Number of expanding-window walk-forward TEST blocks (knobs+breadth picked on "
                             "TRAIN, evaluated on the disjoint TEST block)."),
    participation_frac: float = typer.Option(
        0.10, "--participation-frac",
        help="A coin can absorb this fraction of its median daily $-volume (capacity assumption). The capacity "
             "RATIO B/A is invariant to it (it cancels), so the verdict does not depend on this value."),
    sleeves: str = typer.Option(
        "", "--sleeves",
        help="Comma-separated book sleeves (default = the 4-edge book: tstrend,unlock,xsfunding,positioning)."),
    taker_fee_bps: float = typer.Option(
        5.5, "--taker-fee-bps",
        help="Bybit's PUBLISHED perp taker fee (bps). NET cost = fee + half the coin's measured CS spread."),
) -> None:
    """READ-ONLY: is a BROAD universe + the spread-aware overlay a DEPLOY-BOOK UPGRADE over the current
    liquid-61 ($10M) Bybit book? WALK-FORWARD head-to-head — the rigor Step 3 lacked. A SHADOW eval that NEVER
    changes the live book's returns.

    Step 3 hinted broad+overlay BEAT the live book on ONE OOS split with TRAIN-tuned k/λ ($1M broad+overlay
    NET-OOS Sharpe 1.30 vs $10M liquid-61 1.04) — exactly where overfitting hides. This stress-tests it:
    EXPANDING-WINDOW folds; per fold the broad book's (breadth, k, λ) is SELECTED on TRAIN and BOTH books are
    evaluated NET on the disjoint TEST block (train_hi < test_lo → no leakage). Aggregates OOS test Sharpe
    across folds, estimates each book's DEPLOYABLE CAPITAL (bottleneck coin: participation_frac·ADV / |weight|;
    the overlay downweighting thin coins lifts it), and correlates the two NET series.

    VERDICT: UPGRADE (aggregate lift >= +0.20, broad wins a majority of folds, capacity >= 50% of live — wire
    it in) / CAPACITY-TRADEOFF (Sharpe win but materially lower capacity — a choice) / OVERFIT (the 1.30
    shrinks toward liquid-61 under walk-forward — keep liquid-61) / NO. Writes NOTHING."""
    import logging

    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.broaden_overlay_validate import broaden_overlay_validate

    log = logging.getLogger(__name__)
    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    bounds = tuple(float(x) for x in min_dollar_vols.split(",") if x.strip())
    sleeve_list = [s.strip() for s in sleeves.split(",") if s.strip()] or None

    def _load_unlock_events() -> list:  # type: ignore[type-arg]
        from fxhnt.adapters.orchestration.assets import _data_dir
        from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
        try:
            return load_unlock_events(operational_unlock_repo(settings),
                                      f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → unlock edge omitted, not a crash
            log.warning("broaden-overlay-validate: could not load unlock calendar: %s", exc)
            return []

    try:
        unlock_events = _load_unlock_events()
        rep = broaden_overlay_validate(store, live_bound=live_bound, broad_bounds=bounds, n_folds=n_folds,
                                       participation_frac=participation_frac, taker_fee_bps=taker_fee_bps,
                                       sleeves=sleeve_list, unlock_events=unlock_events)
    finally:
        store.close()

    typer.echo("\n=== BROADEN + OVERLAY deploy-book validation — WALK-FORWARD head-to-head vs liquid-61 — "
               "READ-ONLY — SHADOW eval ===")
    if not rep.get("available"):
        typer.echo(f"INSUFFICIENT DATA: {rep.get('reason', 'no usable book')}")
        typer.echo("  (run `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` first — the high/low + "
                   "turnover features are needed for the measured cost + capacity model)")
        return

    typer.echo(f"\nWalk-forward: {rep['n_folds']} expanding-window folds. (A) LIVE = liquid-61 "
               f"(${rep['live_bound']/1e6:.0f}M, naive eq-wt). (B) BROAD+overlay = per-fold-train-selected "
               f"breadth (from {[f'${b/1e6:g}M' for b in rep['broad_bounds']]}) + spread overlay.")
    typer.echo("\n  fold   window(test)     (A) live Sh   (B) broad Sh   Δ Sh    sel breadth/k/λ        "
               "cap live      cap broad")
    typer.echo("  " + "-" * 116)
    for f in rep["folds"]:
        sel = f"${f['selected_bound']/1e6:g}M k={f['selected_k']:g} λ={f['selected_lam']:g}"
        typer.echo(f"  {f['fold']:>4d}   [{f['test_lo']//86400},{f['test_hi']//86400}]"
                   f"{'':<6}  {f['live']['sharpe']:>10.2f}   {f['broad']['sharpe']:>11.2f}   "
                   f"{f['d_sharpe']:>+6.2f}   {sel:<22}  ${f['cap_live']:>11,.0f}  ${f['cap_broad']:>11,.0f}")

    agg = rep["aggregate"]
    typer.echo("  " + "-" * 116)
    typer.echo(f"  AGG    mean OOS NET Sh   live {agg['mean_live_sharpe']:>+.2f}±{agg['std_live_sharpe']:.2f}"
               f"   broad {agg['mean_broad_sharpe']:>+.2f}±{agg['std_broad_sharpe']:.2f}   "
               f"Δ {agg['d_sharpe']:>+.2f}   broad-wins {agg['win_frac']*100:.0f}% of folds")
    typer.echo(f"\n  CAPACITY (deployable, median across folds): live ${agg['cap_live']:,.0f}  vs  "
               f"broad ${agg['cap_broad']:,.0f}  (ratio {agg['capacity_ratio']:.2f}×)")
    typer.echo(f"  ROBUSTNESS: corr(broad NET, live NET) = {agg['corr_broad_to_live']:+.2f}  "
               f"(1 = same-book; <1 = genuinely different).  broad weight HHI = {agg['broad_hhi']:.3f} "
               f"(concentration).")

    v = rep["verdict"]
    headline = {
        "UPGRADE": "UPGRADE — broad+overlay's aggregate walk-forward NET Sharpe is meaningfully higher AND "
                   "capacity is acceptable. WIRE IT IN.",
        "CAPACITY-TRADEOFF": "CAPACITY-TRADEOFF — broad+overlay wins Sharpe but holds materially less "
                             "deployable capital. A choice, not a free win.",
        "OVERFIT": "OVERFIT — the single-split lift shrinks toward liquid-61 under walk-forward (Step 3 was a "
                   "fluke). KEEP liquid-61.",
        "NO": "NO — broad+overlay is no better on average. KEEP liquid-61.",
    }[v["verdict"]]
    typer.echo(f"\nVERDICT: {headline}")
    typer.echo(f"  (Δ aggregate OOS NET Sharpe {v['d_sharpe']:+.2f}; broad wins {v['win_frac']*100:.0f}% of "
               f"folds; capacity ratio {v['capacity_ratio']:.2f}×.)")
    typer.echo("\nNOTE: writes NOTHING — the live book is untouched (SHADOW eval). Per-fold knobs AND breadth "
               "are selected on TRAIN; the verdict uses the disjoint held-out TEST blocks (no leakage). "
               "Capacity = bottleneck participation_frac·ADV/|weight|; the ratio is invariant to "
               "participation_frac. Cost = Bybit's 5.5bp taker + half the Corwin–Schultz MEASURED spread.")


@app.command("bybit-drawdown-kill-test")
def bybit_drawdown_kill_test(
    train_frac: float = typer.Option(
        0.6, "--train-frac",
        help="Fraction of the combined-book dates used for TRAIN (rest = held-out TEST)."),
    cost_bps: float = typer.Option(
        5.5, "--cost-bps", help="Per-turnover cost (bps) applied to the naive eq-wt combined book."),
    dd_bar: float = typer.Option(
        0.25, "--dd-bar",
        help="Adoption bar: minimum OOS RELATIVE maxDD reduction a kill must deliver (default 25%)."),
    sharpe_bar: float = typer.Option(
        0.30, "--sharpe-bar",
        help="Adoption bar: maximum OOS Sharpe drop a kill may cost (default 0.30)."),
) -> None:
    """READ-ONLY: OOS test of a PURE drawdown-kill on the combined Bybit 4-edge book.

    The deployable book is the naive equal-weight of the 4 sleeves (OOS Sharpe ~2.43, maxDD ~-13%). The full
    risk OVERLAY was REJECTED (it over-levers and loses OOS). This tests a DIFFERENT, simpler thing: a PURE
    drawdown-kill (capital preservation) that flattens to cash on a tail drawdown and re-enters per the live
    `DrawdownKillSwitch` hysteresis — exposure ∈ {0, 1}, NO leverage change EVER (the rejected overlay-kill
    rode on top of vol-target leverage; this is the kill ALONE).

    Reads the PRECOMPUTED `bybit_sleeve_ret` table via PaperRepo (no edge recompute, no OOM). Scores `naive`
    + pure-kills at -10/-15/-20/-25% on a TRAIN split AND a held-out TEST split (Sharpe + maxDD + triggers).
    VERDICT judges BOTH: adopt a kill only if OUT-OF-SAMPLE it cuts maxDD >= --dd-bar relative AND costs <=
    --sharpe-bar Sharpe; else STAY NAIVE (the book's drawdown is already contained). Writes NOTHING."""
    from fxhnt.adapters.persistence.paper_repo import PaperRepo
    from fxhnt.application.bybit_drawdown_kill_test import drawdown_kill_test

    repo = PaperRepo(get_settings().operational_dsn)
    rep = drawdown_kill_test(repo, train_frac=train_frac, cost_bps=cost_bps,
                             dd_bar=dd_bar, sharpe_bar=sharpe_bar)

    typer.echo("\n=== Bybit PURE DRAWDOWN-KILL test — combined book — OOS (TRAIN/TEST) — READ-ONLY ===")
    if not rep.get("available"):
        typer.echo(f"INSUFFICIENT DATA: {rep.get('reason', 'no usable book')}")
        typer.echo("  (run `fxhnt bybit-persist-sleeve-ret` first to populate bybit_sleeve_ret)")
        return

    typer.echo(f"sleeves={rep['sleeves']}  combined-book days={rep['days']}  cost_bps={rep['cost_bps']}  "
               f"train_frac={rep['train_frac']}  (PURE kill: exposure 0|1, NO leverage)")
    typer.echo("\nConfig      TRAIN Sh  TRAIN DD  TRAIN trig   TEST Sh  TEST DD  TEST trig   (full Sh / full DD)")
    typer.echo("  " + "-" * 92)
    table = rep["table"]
    for name in table:
        tr = table[name]["train"]
        te = table[name]["test"]
        fu = table[name]["full"]
        typer.echo(f"  {name:<10}{tr['sharpe']:>7.2f}  {tr['max_dd']*100:>7.1f}%  {tr['triggers']:>7d}   "
                   f"{te['sharpe']:>7.2f}  {te['max_dd']*100:>6.1f}%  {te['triggers']:>7d}   "
                   f"({fu['sharpe']:>5.2f} / {fu['max_dd']*100:>6.1f}%)")

    v = rep["verdict"]
    typer.echo(f"\nbars: OOS relative maxDD reduction >= {v['dd_bar']*100:.0f}%  AND  OOS Sharpe drop <= "
               f"{v['sharpe_bar']:.2f}")
    typer.echo(f"naive OOS: Sharpe {v['naive_test_sharpe']:.2f}  maxDD {v['naive_test_max_dd']*100:.1f}%")
    if v["recommendation"] != "naive":
        typer.echo(f"recommended kill '{v['recommendation']}': OOS DD cut {v['rec_rel_dd_reduction']*100:.0f}% "
                   f"at {v['rec_sharpe_drop']:.2f} Sharpe cost")
    typer.echo(f"\nVERDICT / RECOMMENDATION: deploy '{v['recommendation']}'")
    typer.echo(f"  {v['note']}")
    typer.echo("\nNOTE: writes NOTHING. The kill REUSES the live DrawdownKillSwitch (peak/DD + hysteresis "
               "re-entry); exposure is binary {0,1} — NO leverage, unlike the rejected overlay-kill.")


@app.command("warehouse-catalog")
def warehouse_catalog() -> None:
    """Show the warehouse SSOT inventory: per-symbol bar count + date range (freshness)."""
    from fxhnt.adapters.warehouse import get_feature_store
    from fxhnt.application.warehouse_catalog import format_catalog

    settings = get_settings()
    store = get_feature_store(settings)
    try:
        typer.echo(format_catalog(store.catalog()))
    finally:
        store.close()


@app.command("warehouse-migrate-pg")
def warehouse_migrate_pg(
    duckdb_path: str = typer.Option("", "--duckdb-path",
                                    help="source DuckDB warehouse (default: settings.warehouse_path)"),
    chunk_rows: int = typer.Option(200_000, "--chunk-rows",
                                   help="max rows held in RAM per upsert chunk (memory-bound)"),
    verify: bool = typer.Option(True, "--verify/--no-verify", help="reconcile counts + per-feature checksums"),
) -> None:
    """Phase 2: memory-bounded, idempotent migration of the DuckDB warehouse -> TimescaleDB (operational
    DSN). Streams `features` in bounded chunks (never loads the whole table) + upserts (ON CONFLICT, safe to
    re-run), then verifies row counts + per-feature checksums. Exits non-zero if verification fails.
    NOTHING live changes — this only COPIES into the Timescale tables; FXHNT_FEATURE_STORE stays duckdb."""
    from fxhnt.application.warehouse_migrate import migrate_duckdb_to_timescale

    settings = get_settings()
    src_path = duckdb_path or settings.warehouse_path
    report = migrate_duckdb_to_timescale(
        src_path, settings.operational_dsn, chunk_rows=chunk_rows, verify=verify, progress=typer.echo,
    )
    typer.echo(
        f"\nmigrated {report['features_rows_written']} feature rows "
        f"({report['symbols']} symbols, {report['chunks']} chunks, max {report['max_chunk_rows']} rows/chunk) "
        f"+ {report['membership_rows']} membership rows -> {settings.operational_dsn.split('@')[-1]}"
    )
    if verify:
        typer.echo(f"\n{'feature':>14} {'duckdb_n':>10} {'pg_n':>10} {'duckdb_sum':>16} {'pg_sum':>16}  ok")
        for feat, b in sorted(report["per_feature"].items()):
            typer.echo(f"{feat:>14} {b['count_duckdb']:>10} {b['count_timescale']:>10} "
                       f"{b['sum_duckdb']:>16.6f} {b['sum_timescale']:>16.6f}  {'✓' if b['match'] else '✗'}")
        typer.echo(f"\ncount: duckdb={report['features_count_duckdb']} "
                   f"timescale={report['features_count_timescale']} -> "
                   f"{'VERIFIED ✅' if report['verified'] else 'MISMATCH ❌'}")
        if not report["verified"]:
            raise typer.Exit(1)


# RETIRED 2026-07-14 (Phase 0b Task 7d): the `paper-backfill` CLI + its `_paper_backfill_sleeves` /
# `_paper_backfill_warehouse_panel` helpers (the dead legacy-venue-era combined-crypto `/paper` book's backfill,
# driven by `paper_sleeves.BOOK_CONFIGS[TRUSTED_BOOK]`) are retired in full — redundant with the
# `bybit_4edge` deploy book's own `bybit-paper-backfill` (KEPT, untouched). See
# .superpowers/sdd/task-7d-report.md for the DELETE/DECOUPLE/KEEP map.


@app.command("onchain-probe")
def onchain_probe(
    liquid: bool = typer.Option(
        True, "--liquid/--all",
        help="Restrict the crypto universe to the liquid Bybit subset (median daily turnover >= "
             "--min-dollar-vol). Default --liquid (the tradeable subset)."),
    min_dollar_vol: float = typer.Option(
        10_000_000.0, "--min-dollar-vol", help="Liquid-universe floor (median daily turnover, USD)."),
    market: str = typer.Option(
        "btc", "--market", help="Timing-overlay market index for angle ①: 'btc' (BTCUSDT) or 'ew' "
                                "(liquid equal-weight mean)."),
    ratio: str = typer.Option(
        "fees_mcap", "--ratio", help="Value ratio for angle ②: fees_mcap | revenue_mcap | tvl_mcap."),
    cost_bps: float = typer.Option(5.5, "--cost-bps", help="Per-turnover Bybit-perp cost (bps)."),
    lookback: int = typer.Option(30, "--lookback", help="Trailing window (days) for the supply-growth z-score."),
    train_frac: float = typer.Option(
        0.6, "--train-frac", help="TRAIN fraction of the date axis; the rest is the held-out OOS TEST."),
) -> None:
    """READ-ONLY falsification study of crypto ON-CHAIN / FUNDAMENTAL signals — a data class never mined before
    (every prior probe was price/funding/derivatives). Two angles, each with its own verdict, both net of cost,
    both walk-forward (OOS), both checked vs the 4 live edges:

      ① STABLECOIN-SUPPLY LIQUIDITY TIMING — a long/flat macro-timing overlay on the crypto market driven by
         the trailing z-score of DefiLlama's aggregate stablecoin-supply GROWTH. Real DAILY series → a real
         verdict, but HONESTLY caveated for the ~one-cycle macro-timing overfit trap (strict OOS + non-
         overlapping t-stat + an N/A when the series is too thin).
      ② FUNDAMENTAL VALUE FACTOR — cross-sectional long-CHEAP / short-EXPENSIVE by fees/revenue/TVL ÷ market
         cap (DefiLlama + CoinGecko). The cheap endpoints return a CURRENT SNAPSHOT (not point-in-time), so a
         static ranking on past returns is lookahead-contaminated → angle ② is reported N/A with the diagnostic
         numbers (a real value factor needs historical fundamentals we do not have for free).

    NEEDS network egress to DefiLlama + CoinGecko AND a populated Bybit warehouse (postgres). Writes NOTHING —
    the live paper_* tables are untouched."""
    from fxhnt.adapters.data.onchain_fundamental_live import OnchainFundamentalLive
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.onchain_fundamental_eval import onchain_fundamental_study

    settings = get_settings()
    store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
    universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) if liquid else None

    live = OnchainFundamentalLive()
    tradeable_bases = {s[:-4] for s in store.crypto_close_panel() if s.endswith("USDT")}
    if universe is not None:
        tradeable_bases &= {s[:-4] for s in universe if s.endswith("USDT")}
    try:
        supply_series = live.stablecoin_supply_series()
        fundamentals = live.fundamentals_by_coin(tradeable_bases)
        rep = onchain_fundamental_study(
            store, supply_series=supply_series, fundamentals_by_coin=fundamentals,
            fundamentals_point_in_time=False, universe=universe, market=market,
            ratio_kind=ratio, cost_bps=cost_bps, lookback=lookback, train_frac=train_frac)
    finally:
        store.close()
    _print_onchain_probe(rep, liquid=liquid, min_dollar_vol=min_dollar_vol)


def _print_onchain_probe(rep: dict, *, liquid: bool, min_dollar_vol: float) -> None:
    """Pretty-print the on-chain/fundamental probe: data summaries + a per-angle verdict table."""
    typer.echo("\n=== Crypto ON-CHAIN / FUNDAMENTAL signals — FALSIFICATION STUDY (READ-ONLY) ===")
    scope = (f"LIQUID subset (median turnover >= ${min_dollar_vol:,.0f})" if liquid else "ALL coins")
    typer.echo(f"venue: Bybit (bybit_features)  |  scope: {scope}  |  ratio: {rep['ratio_kind']}")
    typer.echo(f"data: stablecoin-supply series = {rep['supply_series_days']} days "
               f"({rep['supply_first_day']}..{rep['supply_last_day']})  |  "
               f"fundamentals-having coins = {rep['n_fundamentals_coins']}  |  "
               f"value universe (Bybit ∩ fundamentals) = {rep['n_value_universe']}  |  "
               f"live edges for corr = {rep['n_live_edges_for_corr']}")

    def _f(x, fmt="{:+.2f}") -> str:  # noqa: ANN001
        return "N/A" if x is None else fmt.format(x)

    a1 = rep["angle1"]
    typer.echo(f"\n[ANGLE ①] STABLECOIN-SUPPLY LIQUIDITY TIMING -> {a1.get('verdict', 'N/A')}")
    if not a1.get("available"):
        typer.echo(f"  N/A — {a1.get('reason', '')}")
    else:
        typer.echo(f"  market={a1['market']}  chosen: {a1['chosen_direction']} @ thr {a1['chosen_threshold']:+.2f}  "
                   f"days={a1['n_days']}  switches={a1['n_regime_switches']}  active={a1['active_fraction']:.2f}")
        typer.echo(f"  NET full Sharpe={a1['net_full']['sharpe']:6.2f}  "
                   f"OOS Sharpe={a1['net_test_oos']['sharpe']:6.2f}  "
                   f"t-stat(full)={a1['tstat_full']:+.2f}  t-stat(OOS)={a1['tstat_oos']:+.2f}")
        typer.echo(f"  max|corr| to edges={_f(a1['max_abs_corr_edge'])} ({a1['max_corr_edge_name']})  "
                   f"LFC: large={a1['lfc']['large']} frequent={a1['lfc']['frequent']} "
                   f"concentrated={a1['lfc']['concentrated']} -> pass={a1['lfc']['pass']}")
        typer.echo(f"  -> {a1['reason']}")

    a2 = rep["angle2"]
    typer.echo(f"\n[ANGLE ②] FUNDAMENTAL VALUE FACTOR -> {a2.get('verdict', 'N/A')}")
    if not a2.get("available"):
        typer.echo(f"  N/A — {a2.get('reason', '')}")
    else:
        typer.echo(f"  value coins={a2['n_value_coins']}  point-in-time={a2['point_in_time']}")
        typer.echo(f"  NET full Sharpe={a2['net_full']['sharpe']:6.2f}  "
                   f"OOS Sharpe={a2['net_test_oos']['sharpe']:6.2f}  "
                   f"t-stat(full)={a2['tstat_full']:+.2f}  t-stat(OOS)={a2['tstat_oos']:+.2f}")
        typer.echo(f"  max|corr| to edges={_f(a2['max_abs_corr_edge'])} ({a2['max_corr_edge_name']})  "
                   f"corr tstrend={_f(a2.get('corr_tstrend'))}  "
                   f"LFC: large={a2['lfc']['large']} frequent={a2['lfc']['frequent']} "
                   f"concentrated={a2['lfc']['concentrated']} -> pass={a2['lfc']['pass']}")
        typer.echo(f"  -> {a2['reason']}")

    typer.echo("\nNOTE: writes NOTHING — paper_* tables untouched. Needs egress to DefiLlama + CoinGecko and a "
               "populated bybit_features warehouse. Macro-timing on ~one crypto cycle overfits trivially — angle "
               "① demands strict OOS + a non-overlapping t-stat; angle ② is N/A on snapshot fundamentals.")


if __name__ == "__main__":
    app()
