"""Dagster asset graph — the daily operational lifecycle (data ingest → forward tracks → cockpit).

Key assets: `futures_bars` (Databento front-month arrays), `bybit_warehouse_refresh` (nightly Bybit
`bybit_features` update), the per-edge forward-nav tracks (xsfunding/unlock/positioning/bybit_4edge/
sixtyforty/multistrat), and `cockpit_forward` (recompute the forward state from the DB anchors and
upsert into the operational cockpit DB). The DB is the single SSOT — no JSON state files.

NOTE: this module intentionally omits `from __future__ import annotations` because
Dagster 1.12 uses runtime type-hint inspection on the `context` parameter; PEP-563
lazy annotations make the hint a string, breaking the validator.
"""
import os

from dagster import AssetExecutionContext, asset

_FUTURES = ["ES", "NQ", "YM", "RTY", "EMD", "NKD"]


def _data_dir() -> str:
    return os.environ.get("FXHNT_SURFER_DATA_DIR", "/data/surfer")


def bybit_close_cross_section_staleness(store, *, universe, today_day, min_fraction=0.5):  # type: ignore[no-untyped-def]
    """(freshest_full_day, days_behind) for the Bybit `close` cross-section — the data-freshness probe.

    `freshest_full_day` = the newest epoch_day whose `close` cross-section covers >= `min_fraction` of
    `universe` symbols (None if no day meets the threshold / empty universe). Requiring a HEALTHY count (not
    merely "any row on the newest day") means a single brand-new symbol landing today cannot mask a book that
    is otherwise frozen — the exact silent-freeze the perp symbol-skip caused (`close` had 2 symbols on the
    newest day while the panel carried ~296). `days_behind` = `today_day - freshest_full_day` (None when
    `freshest_full_day` is None). READ-ONLY: one `crypto_close_panel()` read, no writes."""
    panel = store.crypto_close_panel()          # {sym: {epoch_day: close}}
    universe_set = set(universe) if universe else set(panel)
    n_universe = len(universe_set)
    if n_universe == 0:
        return None, None
    day_counts: dict = {}                        # {epoch_day: symbols present that day (within the universe)}
    for sym, series in panel.items():
        if sym not in universe_set:
            continue
        for day in series:
            day_counts[day] = day_counts.get(day, 0) + 1
    threshold = max(1, int(min_fraction * n_universe))
    healthy_days = [d for d, c in day_counts.items() if c >= threshold]
    if not healthy_days:
        return None, None
    freshest = max(healthy_days)
    return freshest, today_day - freshest


def refresh_bybit_warehouse(store, *, funding, klines, account_ratio, universe, min_dollar_vol=10_000_000.0,
                            progress=None, today_day=None, max_lag_days=2, min_fraction=0.5,
                            log_error=None):  # type: ignore[no-untyped-def]
    """Plain (Dagster-free, unit-testable) NIGHTLY Bybit warehouse refresh: incrementally update
    `bybit_features` for the liquid universe by calling the RESUMABLE ingests so only NEW days are fetched.

    Daily essentials: klines (perp close/turnover — also rebuilds the liquid set), SPOT close/high/low,
    funding, long_ratio. ALL legs resume=True → each DATE-RESUMES per symbol from its last persisted day, so
    every symbol re-fetches only its new days and ADVANCES every run (the perp klines leg included — it was
    migrated off the old symbol-skip that froze close/turnover at the last ingest date and stalled the Bybit
    forward gate; the spot leg was migrated earlier). The upsert is idempotent so it is re-run-safe regardless.
    `worst_basis` (1m, heavy) is NOT refreshed here — it is a weekly job.

    FRESHNESS GUARD: after the ingests it computes the `close` cross-section's freshest healthy day and, if
    that lags `today_day` (default: today UTC) by more than `max_lag_days`, calls `log_error(...)` with a clear
    message (does NOT crash — the run completes). `log_error` is optional (the Dagster asset passes
    `context.log.error`); the staleness numbers are also returned in the summary for monitoring.

    Returns the four ingest summaries + `close_freshest_day`/`close_days_behind`. `universe` is the active
    USDT-perp set (klines/spot/funding run survivorship-free over it); long_ratio runs over the
    `liquid_universe` subset."""
    from fxhnt.application.bybit_funding_ingest import ingest_bybit_funding
    from fxhnt.application.bybit_klines_ingest import ingest_bybit_klines, ingest_bybit_spot
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.bybit_long_short_ingest import ingest_bybit_long_short

    klines_summary = ingest_bybit_klines(store, klines, universe, resume=True, progress=progress)
    spot_summary = ingest_bybit_spot(store, klines, universe, resume=True, progress=progress)
    funding_summary = ingest_bybit_funding(store, funding, universe, resume=True, progress=progress)
    liquid = sorted(liquid_universe(store, min_dollar_vol=min_dollar_vol))
    ls_summary = ingest_bybit_long_short(store, account_ratio, liquid, resume=True, progress=progress)

    # FRESHNESS GUARD (data-layer analog of the tracker's re-inception guard): after the refresh, verify the
    # `close` cross-section actually ADVANCED. If its freshest healthy day lags `today` by more than
    # `max_lag_days`, emit a LOUD error (via `log_error`) so a future ingest stall is VISIBLE + monitorable
    # rather than silently freezing the book/gate. Never hard-crashes — the pipeline keeps running.
    if today_day is None:
        import datetime as _dt
        today_day = (_dt.datetime.now(_dt.UTC).date() - _dt.date(1970, 1, 1)).days
    freshest_day, days_behind = bybit_close_cross_section_staleness(
        store, universe=universe, today_day=today_day, min_fraction=min_fraction)
    if log_error is not None:
        if freshest_day is None:
            log_error("bybit close cross-section stale: no day meets the healthy symbol-count "
                      f"threshold ({int(min_fraction * 100)}% of {len(universe)} symbols)")
        elif days_behind is not None and days_behind > max_lag_days:
            log_error(f"bybit close cross-section stale: freshest full day = {freshest_day}, "
                      f"{days_behind} days behind (tolerance {max_lag_days}d)")

    return {"klines": klines_summary, "spot": spot_summary, "funding": funding_summary,
            "long_short": ls_summary, "liquid_symbols": len(liquid),
            "close_freshest_day": freshest_day, "close_days_behind": days_behind}


def build_bybit_4edge_forward_nav(repo, store, *, today, at, universe=None, cost_bps=5.5,
                                  unlock_events=None, stablecoin_spot_panel=None,
                                  positioning_capacity_capital=None,
                                  positioning_tail_cap_k=None,
                                  positioning_coin_gross_cap=None):  # type: ignore[no-untyped-def]
    """Plain (Dagster-free, unit-testable) snapshot driver for the Bybit 4-edge forward track: build the
    naive eq-wt 4-sleeve strategy over `store` and run it through the deterministic engine (DB anchor +
    recompute — `forward_engine.run_track`), returning its result dict. The DB is the SSOT; there is no
    JSON state to corrupt. Records the NAIVE equal-weight book (NOT the over-levering live overlay), per
    the track's design. The positioning capturability haircuts apply to the book's positioning sleeve so
    the book's number is capturable too."""
    from fxhnt.application.bybit_forward_track import BybitFourEdgeStrategy
    from fxhnt.application.forward_engine import run_track

    strategy = BybitFourEdgeStrategy(
        store, universe=universe, cost_bps=cost_bps,
        unlock_events=unlock_events, stablecoin_spot_panel=stablecoin_spot_panel,
        positioning_capacity_capital=positioning_capacity_capital,
        positioning_tail_cap_k=positioning_tail_cap_k,
        positioning_coin_gross_cap=positioning_coin_gross_cap)
    return run_track(repo, "bybit_4edge", strategy, today=today, at=at)


def build_bybit_4edge_levered_forward_nav(repo, store, *, today, at, universe=None, cost_bps=5.5,
                                          unlock_events=None, stablecoin_spot_panel=None,
                                          positioning_capacity_capital=None,
                                          positioning_tail_cap_k=None,
                                          positioning_coin_gross_cap=None):  # type: ignore[no-untyped-def]
    """Observe-only LEVERED shadow of the 4-edge book: the SAME sleeves under the static `_LEVERED_SLEEVE_SHAPE`
    (carry-weighted, book-gross <= MAX_BOOK_GROSS), run through the deterministic engine (DB anchor +
    recompute) under its OWN strategy_id `bybit_4edge_levered`. No capital; records what the levered book
    would do so it can be validated OOS before deployment. Positioning haircuts apply to its positioning
    sleeve too (same capturable basis as the naive book)."""
    from fxhnt.application.bybit_forward_track import _LEVERED_SLEEVE_SHAPE, MAX_BOOK_GROSS, BybitFourEdgeStrategy
    from fxhnt.application.forward_engine import run_track

    strategy = BybitFourEdgeStrategy(
        store, universe=universe, cost_bps=cost_bps, unlock_events=unlock_events,
        stablecoin_spot_panel=stablecoin_spot_panel,
        leverage_shape=_LEVERED_SLEEVE_SHAPE, max_gross=MAX_BOOK_GROSS,
        positioning_capacity_capital=positioning_capacity_capital,
        positioning_tail_cap_k=positioning_tail_cap_k,
        positioning_coin_gross_cap=positioning_coin_gross_cap)
    return run_track(repo, "bybit_4edge_levered", strategy, today=today, at=at)


def build_bybit_sleeve_forward_nav(repo, store, sleeve, *, today, at, universe=None, cost_bps=5.5,
                                   unlock_events=None, positioning_capacity_capital=None,
                                   positioning_tail_cap_k=None,
                                   positioning_coin_gross_cap=None):  # type: ignore[no-untyped-def]
    """Plain (Dagster-free, unit-testable) driver for a SINGLE Bybit sleeve as its own forward track: build
    the one-sleeve `BybitFourEdgeStrategy` over `store` (bybit_features) and run it through the deterministic
    engine (DB anchor + recompute). The DB (forward_nav + anchor) is the SSOT — no JSON state. Shared by the
    per-edge Bybit tracks (xsfunding / unlock / positioning)."""
    from fxhnt.application.bybit_forward_track import BybitFourEdgeStrategy
    from fxhnt.application.forward_engine import run_track
    strategy = BybitFourEdgeStrategy(
        store, universe=universe, cost_bps=cost_bps, sleeves=[sleeve], unlock_events=unlock_events,
        positioning_capacity_capital=positioning_capacity_capital, positioning_tail_cap_k=positioning_tail_cap_k,
        positioning_coin_gross_cap=positioning_coin_gross_cap)
    return run_track(repo, sleeve, strategy, today=today, at=at)


def build_positioning_forward_nav(repo, store, *, today, at, universe=None, cost_bps=5.5,
                                  positioning_capacity_capital=None,
                                  positioning_tail_cap_k=None,
                                  positioning_coin_gross_cap=None):  # type: ignore[no-untyped-def]
    """POSITIONING single-sleeve Bybit track — thin wrapper on `build_bybit_sleeve_forward_nav` (keeps the
    capturability haircuts). See that function."""
    return build_bybit_sleeve_forward_nav(
        repo, store, "positioning", today=today, at=at, universe=universe, cost_bps=cost_bps,
        positioning_capacity_capital=positioning_capacity_capital, positioning_tail_cap_k=positioning_tail_cap_k,
        positioning_coin_gross_cap=positioning_coin_gross_cap)


def build_bybit_paper_book(repo, store, *, capital, at, universe=None, unlock_events=None,
                           enabled=True, positioning_coin_gross_cap=None):  # type: ignore[no-untyped-def]
    """Plain (Dagster-free, unit-testable) driver for the Bybit LIVE paper book: derive the naive-eq-wt
    4-edge book's per-symbol positions on the latest Bybit prices and persist them (positions/trades/nav)
    via `derive_and_persist_bybit_paper_book` into `repo` (a `PaperRepo` — bybit is its sole venue).
    Returns the number of open positions persisted. READ-ONLY against the warehouse `store` (only the paper
    store is written); memory-bounded on the liquid `universe`. `enabled=False` is a no-op.

    Charges the per-COIN MEASURED cost (real L1 spread from `bybit_real_spread`, taker_fee + 0.5·spread) so the
    forward `/paper` step ties to the cockpit Backtest NET; falls back to the flat 5.5bp when no spread refresh
    has run.

    `positioning_coin_gross_cap` (positioning sleeve ONLY) forwards into `derive_and_persist_bybit_paper_book`
    so the LIVE persisted positions honor the same per-coin cap the reconciled return uses. Default None
    preserves the byte-identical uncapped behaviour; the nightly asset passes `POSITIONING_COIN_GROSS_CAP`."""
    from fxhnt.application.bybit_paper_book import derive_and_persist_bybit_paper_book
    from fxhnt.application.bybit_spread_refresh import read_real_spread_by_coin

    real_spread = read_real_spread_by_coin(repo) or None
    n = derive_and_persist_bybit_paper_book(
        repo, store, capital=capital, at=at, universe=universe,
        unlock_events=unlock_events, enabled=enabled, real_spread_by_coin=real_spread,
        positioning_coin_gross_cap=positioning_coin_gross_cap)
    # PROACTIVE: re-materialize the headline nav-summary now the bybit nav changed, so the first / + /paper
    # request is already fast (read 1 row, not the full nav). The read path self-heals if this is skipped, so
    # a refresh failure must NEVER fail the book write — guarded.
    if enabled:
        try:
            from fxhnt.application.nav_summary import refresh_nav_summary
            refresh_nav_summary(repo)
        except Exception:  # noqa: BLE001 — the read-path self-heal is the safety net
            pass
    return n


def persist_bybit_sleeve_returns(repo, store, *, at, universe=None, cost_bps=5.5,
                                 sleeves=None, unlock_events=None,
                                 stablecoin_spot_panel=None,
                                 sleeve_returns=None):  # type: ignore[no-untyped-def]
    """Plain (Dagster-free, unit-testable) writer of the PRECOMPUTED Bybit 4-edge BACKTEST data: compute each
    sleeve's full daily return series via `sleeve_returns_from_store` (the SAME live runners the forward track
    uses, memory-bounded on the liquid `universe`) and idempotently upsert them into `bybit_sleeve_ret` so the
    cockpit sim configures a precomputed backtest WITHOUT ever recomputing the heavy 4-edge book on a web
    request. Returns the number of (date, sleeve) rows written. READ-ONLY against the warehouse store.

    HEAVY compute — must run in its OWN memory, NEVER in the 4Gi dagster daemon (it OOM-killed the daemon +
    skipped the forward-nav step). Invoked by the `fxhnt bybit-persist-sleeve-ret` CLI / own-memory Job, NOT
    by any nightly asset. Lives here (not the CLI module) so it stays import-light + unit-testable."""
    import datetime as _dt

    from fxhnt.application.bybit_book_eval import _DEFAULT_BYBIT_SLEEVES, sleeve_returns_from_store

    epoch = _dt.date(1970, 1, 1)
    sel = list(sleeves) if sleeves is not None else list(_DEFAULT_BYBIT_SLEEVES)
    written = 0
    for sleeve in sel:
        # compute-once: reuse a precomputed series when the caller already has it (the persist Job computes the
        # sleeves ONCE and reuses them for the backtest references), else compute from the store as before.
        if sleeve_returns is not None and sleeve in sleeve_returns:
            series = sleeve_returns[sleeve]
        else:
            series = sleeve_returns_from_store(
                store, sleeve, universe=universe, cost_bps=cost_bps,
                unlock_events=unlock_events, stablecoin_spot_panel=stablecoin_spot_panel)
        for day, ret in series.items():
            repo.upsert_bybit_sleeve_ret((epoch + _dt.timedelta(days=int(day))).isoformat(),
                                         sleeve, float(ret), at=at)
            written += 1
    return written


@asset
def futures_bars(context: AssetExecutionContext) -> int:
    """Bronze → silver → warehouse: ingest continuous front-month futures arrays.

    Skips (with a warning) any root whose DBN file is missing — never fails the asset.
    """
    import datetime as dt
    import os

    from fxhnt.adapters.data.dbn_local import read_front_arrays
    from fxhnt.adapters.warehouse import get_feature_store
    from fxhnt.adapters.warehouse.silver_futures import normalize_futures
    from fxhnt.application.warehouse_ingest import WarehouseIngest
    from fxhnt.config import get_settings

    s = get_settings()
    s.ensure_dirs()
    # Futures fetch is Databento (paid) — refresh only on Sundays (UTC), cost-capped, like the retired cron.
    if dt.datetime.now(dt.UTC).isoweekday() == 7 and os.environ.get("DATABENTO_API_KEY"):
        from fxhnt.adapters.data.futures_fetch import fetch_futures
        start = (dt.date.today() - dt.timedelta(days=365 * s.databento.fetch_years)).isoformat()
        try:
            recs, ok, spent = fetch_futures(_data_dir(), start=start, max_cost_usd=s.databento.max_cost_usd)
            context.log.info(f"futures_bars fetched {ok} roots, {recs} records, ${spent:.2f} (since {start})")
        except Exception as e:  # noqa: BLE001
            context.log.warning(f"futures_bars fetch failed (ingesting existing data): {e}")
    ingest = WarehouseIngest(get_feature_store(s))
    total = 0
    for root in _FUTURES:
        try:
            days, inst, close, high, low = read_front_arrays(f"{_data_dir()}/{root}.dbn", root)
            total += ingest.ingest_bars(root, normalize_futures(root, days, inst, close, high, low))
        except Exception as e:  # noqa: BLE001 — missing file is expected in dev/test
            context.log.warning(f"futures_bars {root} skipped: {e}")
    context.log.info(f"futures_bars ingested {total} rows")
    return total


def _persist_track_backtest_ref(context: AssetExecutionContext, name: str, build_strategy) -> None:  # type: ignore[type-arg,no-untyped-def]
    """Persist the reconciliation-gate backtest reference for an individual forward track, routed through the
    ONE `backtest_ref` dispatcher (fxhnt.application.backtest_refs) so this inline nightly path, the
    `backtest-refs --all` CLI (report-kind), and the B3 runner all share the same dispatch. For
    `recompute-replay` strategies (multistrat/crypto_tstrend/stablecoin/unlock) this is bit-identical to
    the prior inline behaviour: `build_strategy().advance(None, {})` = the full inception series at the EXACT
    basis the forward books (e.g. crypto_tstrend's forward is 15%-book-vol-targeted, not the raw TrendRunner
    sleeve). For a `sleeve` strategy it would resolve a feature store instead, since `sleeve_returns_from_store`
    needs a store, not a replay strategy — `get_feature_store(s)` is that store. No
    registry entry is `sleeve`-kind today (xsfunding moved to `report`-kind on its own Bybit single-sleeve
    track, Phase 0b — see `xsfunding_nav`/`build_bybit_sleeve_forward_nav`); this branch stays generic
    machinery for a future standalone sleeve on some other venue. Keyed by `strategy_id` = name minus the `_nav`
    suffix.
    BEST-EFFORT: a ref-persist failure logs and returns — it must NEVER fail the forward step (the ref is a
    gate input, not the track itself). Without it the gate WITHHOLDS PASS (no ref to reconcile).

    For `recompute-replay` strategies (multistrat/crypto_tstrend/stablecoin/unlock) this ALSO persists
    the full inception series into `sim_curve_ret` (Task 1/D1) — the precomputed curve the cockpit's
    `/paper/sim` reads for its IBKR books (multistrat), mirroring the Bybit book's `bybit_sleeve_ret`
    precompute. One replay call feeds BOTH writes (the reconciliation ref and the sim curve), so the sim's
    curve is always the SAME basis the gate reconciles the forward track against."""
    import datetime as _dt

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.adapters.persistence.paper_repo import PaperRepo
    from fxhnt.adapters.warehouse import get_feature_store
    from fxhnt.application.backtest_refs import backtest_ref, registry_backtest_kind
    from fxhnt.application.bybit_forward_track import backtest_summary_from_rows
    from fxhnt.config import get_settings

    sid = name[:-4] if name.endswith("_nav") else name
    try:
        s = get_settings()
        kind = registry_backtest_kind(sid)
        replay_rows: list[tuple[str, float]] | None = None
        if kind == "recompute-replay":
            # Build the replay rows ONCE here (the same call `backtest_ref`'s recompute-replay branch would
            # make) so they can feed BOTH the reconciliation-gate ref AND the sim_curve_ret precompute,
            # instead of replaying the strategy twice. Hold the rows; they are only persisted below AFTER the
            # `sm is None` guard, so BOTH writes are gated on the SAME condition (same basis, always
            # together) — a too-short/insufficient replay writes NEITHER, never a degenerate sim curve with
            # no matching gate ref.
            replay_rows, _ = build_strategy().advance(None, {})
            sm = backtest_summary_from_rows(sid, replay_rows) if len(replay_rows) >= 2 else None
        else:
            store = get_feature_store(s) if kind == "sleeve" else None
            sm = backtest_ref(sid, store, build_strategy=build_strategy)
        if sm is None:
            return
        repo = ForwardNavRepo(s.operational_dsn)
        repo.migrate()
        repo.upsert_backtest_summary(sm, at=_dt.datetime.now(_dt.UTC).replace(tzinfo=None))
        # Persist the sim's precomputed curve from the SAME replay rows the gate ref was just written from —
        # gated together with it (only reached when sm is not None), so the two never diverge in basis.
        if replay_rows:
            curve_repo = PaperRepo(s.operational_dsn)
            curve_repo.migrate()
            at_curve = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)
            for date_iso, ret in replay_rows:
                curve_repo.upsert_sim_curve_ret(sid, date_iso, float(ret), at=at_curve)
        context.log.info(f"{name}: backtest ref cagr={sm.cagr * 100:.1f}% sharpe={sm.sharpe:.2f} "
                         f"-> backtest_summary[{sid}] (gate can reconcile)")
    except Exception as e:  # noqa: BLE001 — the ref is best-effort; never fail the forward step over it
        context.log.warning(f"{name}: backtest ref persist skipped (gate withholds PASS until present): {e}")


def _run_paper_tracker(context: AssetExecutionContext, name: str, build_strategy,
                       *, persist_backtest_ref: bool = False) -> dict:  # type: ignore[type-arg,no-untyped-def]
    """Run one forward track through the deterministic engine (spec sub-project A): the DB anchor is the
    only persistent state; the record is recomputed (recompute tracks) or appended (observed tracks). A
    transient network/API error while building the strategy logs a warning and no-ops (the anchor + prior
    record are untouched); genuine bugs propagate and fail the asset loudly.

    `persist_backtest_ref=True` also publishes this track's reconciliation-gate reference from its own
    strategy (see `_persist_track_backtest_ref`)."""
    import datetime as _dt
    import json
    import urllib.error

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.application.forward_engine import run_track
    from fxhnt.config import get_settings

    # Isolate the databento provider error base too (a 422/HTTP fault while a track recomputes off OPRA), the
    # same way transient net errors are isolated — NOT bare Exception, so genuine strategy-math bugs stay loud.
    provider_errors: tuple[type[BaseException], ...] = ()
    try:
        from databento.common.error import BentoError
        provider_errors = (BentoError,)
    except Exception:  # noqa: BLE001 — databento is an optional extra; absent → nothing extra to catch
        pass

    sid = name[:-4] if name.endswith("_nav") else name
    today = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%d")
    at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)
    try:
        repo = ForwardNavRepo(get_settings().operational_dsn)
        repo.migrate()
        out = run_track(repo, sid, build_strategy(), today=today, at=at)
        context.log.info(f"{name}: {out['days']}d through {out.get('last_date')} "
                         f"(mode={out['record_mode']}, reinception={out['reinception']})")
        if persist_backtest_ref:
            _persist_track_backtest_ref(context, name, build_strategy)
        return {"forward_days": out["days"], "last_date": out.get("last_date"),
                "reinception": out["reinception"]}
    except (urllib.error.URLError, TimeoutError, ConnectionError, OSError, json.JSONDecodeError,
            *provider_errors) as e:
        context.log.warning(f"{name} step failed (record unchanged): {e}")
        return {"forward_days": 0, "last_date": None, "reinception": False}


@asset
def sixtyforty_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """Paper-forward 60/40 baseline (Tiingo native total-return adjClose, 33yr history)."""
    from fxhnt.adapters.data.tiingo_daily import TiingoDailyClient
    from fxhnt.application.paper_strategies import SixtyFortyStrategy

    return _run_paper_tracker(
        context, "sixtyforty_nav", lambda: SixtyFortyStrategy(TiingoDailyClient())
    )


@asset
def multistrat_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """Paper-forward ADAPTIVE MULTI-STRAT book (5 pure-equity ETFs) — the TradFi diversifier, uncorrelated
    to the crypto edges; it executes on IBKR paper (see `fxhnt execute-multistrat`). Snapshots today's
    Yahoo adjusted closes PIT into the warehouse, then recomputes off the FROZEN PIT series (spec §4) so the
    record is stable against Yahoo's retroactive adjustments. `persist_backtest_ref=True` publishes its
    basis-matched backtest reference so the reconciliation gate reconciles it for real instead of
    withholding PASS."""
    import datetime as _dt

    from fxhnt.adapters.data.yahoo_daily import YahooDailyClient
    from fxhnt.adapters.data.yahoo_pit import YahooPitClient
    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.application.paper_strategies import MultiStratStrategy
    from fxhnt.config import get_settings
    from fxhnt.domain.strategies.multistrat import FUND_INSTRUMENTS

    repo = ForwardNavRepo(get_settings().operational_dsn)
    repo.migrate()
    at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)
    try:
        live = YahooDailyClient()
        for sym in FUND_INSTRUMENTS:
            repo.snapshot_yahoo_closes(sym, live.adj_closes(sym), at)   # freeze today's fetch PIT
            try:
                repo.snapshot_yahoo_volumes(sym, live.volumes(sym), at)   # freeze today's ADV input PIT
            except Exception as ve:  # noqa: BLE001 — volume is an ADV input only; never break the close snapshot
                context.log.warning(f"multistrat_nav: {sym} volume snapshot failed (skipped): {ve}")

        # Also snapshot each UCITS LSE line's volume (the ucits-regime ADV input) — guarded the same way as
        # the US volume snapshot above; a thin/absent line (e.g. DBMF.L) must not break the close snapshot.
        for listing in get_settings().ucits.map.values():
            try:
                repo.snapshot_yahoo_volumes(listing.yahoo_ticker, live.volumes(listing.yahoo_ticker), at)
            except Exception as ve:  # noqa: BLE001 — volume is an ADV input only; never break the close snapshot
                context.log.warning(
                    f"multistrat_nav: {listing.yahoo_ticker} volume snapshot failed (skipped): {ve}")
    except Exception as e:  # noqa: BLE001 — a Yahoo outage in this PRE-tracker step must not skip the gate
        # DEGRADE to a no-op (record unchanged) rather than raise out of the asset and skip cockpit_forward.
        context.log.error(f"multistrat_nav: yahoo snapshot failed (record unchanged): {e}")
        return {"forward_days": 0, "last_date": None, "reinception": False}

    # drop_incomplete_today=False: this runs at 23:30 UTC AFTER the US close (today's FINAL close already
    # frozen above), so today MUST be booked or the track stays 1 trading day stale forever. The SAME
    # build_strategy lambda backs both the forward track AND its backtest ref (persist_backtest_ref=True), so
    # both sides use the identical complete-today basis — no reconciliation-gate divergence.
    return _run_paper_tracker(
        context, "multistrat_nav",
        lambda: MultiStratStrategy(YahooPitClient(repo), instruments=FUND_INSTRUMENTS,
                                   drop_incomplete_today=False),
        persist_backtest_ref=True,
    )


@asset(deps=[multistrat_nav])
def multistrat_levered_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """OBSERVE-ONLY levered shadow of the multistrat ETF book — the SAME risk-parity book LEVERED to its 10%
    vol target minus the honest financing drag on the borrowed leg (return-defined, never executed). Depends
    on `multistrat_nav` so today's Yahoo PIT closes are ALREADY frozen; it recomputes off that SAME frozen PIT
    series wrapped in `LeveredShadowStrategy`, and publishes its own forward track + basis-matched backtest
    ref (persist_backtest_ref=True) so the reconciliation gate + the cockpit sim read it like the base book."""
    from fxhnt.adapters.data.yahoo_pit import YahooPitClient
    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.application.paper_strategies import LeveredShadowStrategy, MultiStratStrategy
    from fxhnt.config import get_settings
    from fxhnt.domain.strategies.multistrat import FUND_INSTRUMENTS
    from fxhnt.registry import STRATEGY_REGISTRY

    repo = ForwardNavRepo(get_settings().operational_dsn)
    repo.migrate()
    p = STRATEGY_REGISTRY["multistrat_levered"]["definition"]["params"]
    return _run_paper_tracker(
        context, "multistrat_levered_nav",
        lambda: LeveredShadowStrategy(
            MultiStratStrategy(YahooPitClient(repo), instruments=FUND_INSTRUMENTS,
                               drop_incomplete_today=False),
            target_vol=p["target_vol"], financing_bps=p["financing_bps"], max_leverage=p["max_leverage"]),
        persist_backtest_ref=True,
    )


# RETIRED 2026-06-20 (cockpit cleanup): the live tracks funding_nav (absolute-hurdle, superseded by
# cross-sectional xsfunding), gd_nav / multistrat_nav (futures/TradFi premia — falsified this session,
# scale-gated, no deployable edge), and poc_nav (old crypto residual-mom+VRP PoC, superseded by the two
# validated edges). Their strategy classes stay in paper_strategies.py (tested, on the shelf). The live
# fleet now carries only the 3 crypto edges (combined, xsfunding, unlock), the 2 equity-factor sleeves,
# and the 60/40 benchmark — signal, not noise.


# RETIRED 2026-06-20: eqfactor_scores + eqfactor_long_nav + eqfactor_tilt_nav (US-equity price-factor
# sleeves — weakly-held, edge never established; kept briefly as a non-crypto experiment, now retired as
# unproven). Strategy classes (compute_factor_scores / EquityFactorLong / EquityFactorTilt) stay in
# equity_factor_strategy.py for the record. Cockpit fleet now = the 2 validated crypto edges + the
# crypto-momentum book + the 60/40 benchmark.

# NOTE: xsfunding_nav moved (Phase 0b venue consolidation) — it is now a Bybit single-sleeve track defined
# alongside positioning_nav (below, after `bybit_warehouse_refresh`), since it depends on that asset.

# NOTE: unlock_nav ALSO moved (Phase 0b venue consolidation, Task 2) — it is now a Bybit single-sleeve track
# defined alongside xsfunding_nav/positioning_nav (below, after `bybit_warehouse_refresh`). CONCERN carried
# into the task report: the retired construction here (`UnlockCalendarLive(...).snapshot(...)`, reached via
# `UnlockShortForward.advance()`) was the ONLY nightly caller of `UnlockCalendarLive`'s weekly DefiLlama-
# emissions refresh, which is what kept the operational `unlock_events` table (read everywhere via
# `load_unlock_events`) current. Every read site degrades gracefully to `[]`/stale data on failure — nothing
# crashes — but with this asset gone, NOTHING refreshes new cliff events into that table anymore. Needs a
# follow-up (its own small nightly asset/CLI wrapping the DefiLlama refresh), out of scope for this migration.


# RETIRED 2026-07-13 (Phase 0b Task 4): crypto_tstrend_nav (standalone legacy-venue-perp TS-momentum forward
# track) + stablecoin_rotation_nav (legacy-venue-spot peg-reversion). RETIRED 2026-07-14 (Phase 0b Task 7d): the
# dead legacy-venue-era combined-crypto `/paper` book (`_live_sleeves`, `_sleeve_return_series`,
# `_assemble_paper_snapshot_inputs`, `build_paper_book_snapshot`, the `paper_book_snapshot` asset) — it read
# the `features` (legacy venue) table Task 7b stopped ingesting, the cockpit has not displayed it since Task 7a
# (Bybit-only), and it was redundant with the `bybit_4edge` deploy book (the real crypto fund, on
# `bybit_features`). See .superpowers/sdd/task-7d-report.md for the DELETE/DECOUPLE/KEEP map. The
# `crypto_tstrend`/`unlock` SLEEVES inside `bybit_4edge` (Bybit data) are UNCHANGED — separate live edges,
# not either retired display.

@asset
def bybit_warehouse_refresh(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """NIGHTLY incremental refresh of the Bybit `bybit_features` warehouse for the active USDT-perp universe:
    klines (perp close/turnover), SPOT close/high/low (the carry's long leg), funding, long_ratio — the daily
    essentials feeding the 4-edge book. All ingests are RESUMABLE → only new days are fetched (the spot leg
    resumes from its last persisted date so it advances every run). worst_basis (1m, heavy) is a weekly job,
    not refreshed here.

    Best-effort for TRANSIENT failures: a network/API hiccup logs a warning and returns partial state rather
    than killing the nightly run (the snapshot asset then steps the tracker on whatever data IS fresh)."""
    import urllib.error

    from fxhnt.adapters.data.bybit_account_ratio import BybitAccountRatio
    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.config import get_settings

    s = get_settings()
    funding = BybitFunding()
    store = TimescaleFeatureStore(s.operational_dsn, table="bybit_features")
    try:
        universe = funding.active_universe()
        summary = refresh_bybit_warehouse(
            store, funding=funding, klines=BybitKlines(), account_ratio=BybitAccountRatio(),
            universe=universe, progress=context.log.info, log_error=context.log.error)
        context.log.info(
            f"bybit_warehouse_refresh: {summary['klines']['symbols']} klines, "
            f"{summary['spot']['symbols']} spot, "
            f"{summary['funding']['symbols']} funding, {summary['long_short']['symbols']} long_ratio "
            f"({summary['liquid_symbols']} liquid)")
        return summary
    except (urllib.error.URLError, TimeoutError, ConnectionError, OSError) as e:  # transient: don't kill the run
        context.log.warning(f"bybit_warehouse_refresh failed (warehouse unchanged): {e}")
        return {"klines": {}, "spot": {}, "funding": {}, "long_short": {}, "liquid_symbols": 0}
    finally:
        store.close()


@asset(
    deps=[bybit_warehouse_refresh],
    op_tags={"dagster-k8s/config": {"container_config": {"resources": {
        "requests": {"memory": "6Gi"}, "limits": {"memory": "8Gi"}}}}},
)
def bybit_book_precompute(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """NIGHTLY heavy precompute of the Bybit 4-edge book's scheduled data — the SINGLE asset that folds in the
    retired `compare-measured-precompute` + `backtest-refs` CronJobs. Runs AFTER `bybit_warehouse_refresh` on
    the LIQUID `bybit_features` universe and, in ONE compute-once pass over the 4 deploy sleeves, writes BOTH
    the `bybit_sleeve_ret` cockpit-sim precompute AND the `_BYBIT_BOOK_SIDS` report-kind reconciliation-gate
    refs (bybit_4edge / bybit_4edge_levered / positioning / xsfunding / unlock). Thin wrapper over the shared
    `_persist_bybit_book` helper — the SAME logic `backtest-refs --all` and `bybit-persist-sleeve-ret` run —
    so there is exactly one source of the persist logic.

    HEAVY compute — must NOT run in the 4Gi dagster daemon (inline in `bybit_4edge_nav` it OOM-killed the
    daemon, exit 137, and — OOM being uncatchable — skipped the forward-nav step). It is tagged for a
    high-memory per-run K8s Job (K8sRunLauncher isolates it in its own pod, `op_tags` requests 6Gi/limits
    8Gi), so the daemon never carries the compute. `_persist_bybit_book` is imported lazily inside the body
    (assets.py stays import-light, matching the other assets)."""
    from fxhnt.application.bybit_book_persist import _persist_bybit_book
    from fxhnt.config import get_settings

    n, scope = _persist_bybit_book(get_settings())
    context.log.info(f"bybit_book_precompute: persisted {n} bybit_sleeve_ret row(s) + the report-kind "
                     f"reconciliation-gate refs from {scope}")
    return {"bybit_sleeve_ret_rows": n, "scope": scope}


@asset
def deribit_funding_bars(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """NIGHTLY incremental ingest of Deribit perp DAILY funding into `bybit_features` (feature
    'deribit_funding', keyed by the Bybit perp symbol) — the Deribit leg of the cross-venue carry edge.
    Its forward consumer `xvenue_carry_nav` was RETIRED 2026-07-07 (validated-but-not-live); this ingest is
    KEPT running standalone to hold the deribit panel warm for a possible cross-venue-carry revival. Mirrors
    the manual `fxhnt ingest-deribit-funding` CLI (SAME default instruments, DEFAULT_DERIBIT_INSTRUMENTS) and
    the retired legacy-venue `crypto_spot` asset. DATE-RESUMABLE: each instrument resumes from its last
    persisted day, so only new days are fetched each run.

    Best-effort for TRANSIENT failures: a Deribit network/API hiccup logs a warning and leaves the panel
    unchanged rather than killing the nightly run."""
    import urllib.error

    from fxhnt.adapters.data.deribit_funding import DeribitFunding
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.deribit_funding_ingest import DEFAULT_DERIBIT_INSTRUMENTS, ingest_deribit_funding
    from fxhnt.config import get_settings

    s = get_settings()
    store = TimescaleFeatureStore(s.operational_dsn, table="bybit_features")
    try:
        summary = ingest_deribit_funding(store, DeribitFunding(), list(DEFAULT_DERIBIT_INSTRUMENTS))
        context.log.info(
            f"deribit_funding_bars: {summary['symbols']} instruments, {summary['day_rows']} day-rows "
            f"(oldest day {summary['oldest_day']})")
        return summary
    except (urllib.error.URLError, TimeoutError, ConnectionError, OSError) as e:  # transient: don't kill the run
        context.log.warning(f"deribit_funding_bars failed (panel unchanged): {e}")
        return {"symbols": 0, "day_rows": 0, "oldest_day": None}
    finally:
        store.close()


@asset(deps=[bybit_warehouse_refresh])
def bybit_4edge_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """Paper-forward 4-edge Bybit book (NAIVE equal-weight — the honest deployable number; NOT the
    over-levering live overlay). Runs AFTER the warehouse refresh, on the LIQUID universe over `bybit_features`,
    and runs the strategy through the deterministic engine (DB anchor + recompute — T0 frozen on first run,
    idempotent). Records directly into the `forward_nav`/`strategy_anchor` cockpit tables (the DB is the SSOT;
    no JSON state to corrupt)."""
    import datetime as _dt
    import urllib.error

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
    from fxhnt.config import get_settings

    s = get_settings()
    store = TimescaleFeatureStore(s.operational_dsn, table="bybit_features")
    try:
        universe = liquid_universe(store) or None   # None → full universe (empty warehouse → empty book)
        # unlock calendar DB-first (operational unlock_events), JSON fallback — same source the live sleeve uses.
        unlock_events = load_unlock_events(operational_unlock_repo(s), f"{_data_dir()}/unlock_calendar.json")
        from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP, POSITIONING_TAIL_CAP_K
        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)
        out = build_bybit_4edge_forward_nav(
            repo, store, today=today, at=at,
            universe=universe, cost_bps=s.bybit_cost_bps, unlock_events=unlock_events,
            positioning_capacity_capital=s.paper_capital, positioning_tail_cap_k=POSITIONING_TAIL_CAP_K,
            positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP)
        context.log.info(
            f"bybit_4edge_nav: T0={out['t0']}, {out['days']}d through {out['last_date']} "
            f"(+{out['total_return'] * 100:.2f}%, Sharpe {out['sharpe']:.2f}, "
            f"reinception={out['reinception']})")
        # NOTE: the heavy PRECOMPUTED per-sleeve backtest series (bybit_sleeve_ret, the cockpit sim's input)
        # is DELIBERATELY NOT persisted here. Running the full 4-sleeve compute in the 4Gi dagster daemon
        # OOM-killed the pod (exit 137); because OOM is uncatchable, that also skipped the forward-nav step
        # above and broke the nightly forward track. The writer now lives in its OWN-MEMORY scheduled
        # CronJob: `fxhnt backtest-refs --all` (infra/k8s/jobs/fxhnt-backtest-refs.yaml, via the shared
        # `_persist_bybit_book` helper in cli.py), run weekly / on-demand. This asset is forward-nav-ONLY —
        # light (~2m) and OOM-proof.
        return {"inception": out["t0"], "last_date": out["last_date"], "forward_days": out["days"],
                "forward_return_pct": out["total_return"] * 100, "forward_sharpe": out["sharpe"]}
    except (urllib.error.URLError, TimeoutError, ConnectionError, OSError) as e:  # transient: don't kill the run
        context.log.warning(f"bybit_4edge_nav step failed (record unchanged): {e}")
        return {"inception": None, "last_date": None, "forward_days": 0}
    finally:
        store.close()


@asset(deps=[bybit_warehouse_refresh])
def bybit_4edge_levered_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """OBSERVE-ONLY levered shadow of bybit_4edge (static leverage, book-gross <= 1.5x). Runs through the
    deterministic engine (DB anchor + recompute) under its own strategy_id `bybit_4edge_levered`; forward-nav-
    ONLY (light, OOM-proof), same as bybit_4edge_nav."""
    import datetime as _dt
    import urllib.error

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
    from fxhnt.config import get_settings

    s = get_settings()
    store = TimescaleFeatureStore(s.operational_dsn, table="bybit_features")
    try:
        universe = liquid_universe(store) or None
        unlock_events = load_unlock_events(operational_unlock_repo(s), f"{_data_dir()}/unlock_calendar.json")
        from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP, POSITIONING_TAIL_CAP_K
        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)
        out = build_bybit_4edge_levered_forward_nav(
            repo, store, today=today, at=at,
            universe=universe, cost_bps=s.bybit_cost_bps, unlock_events=unlock_events,
            positioning_capacity_capital=s.paper_capital, positioning_tail_cap_k=POSITIONING_TAIL_CAP_K,
            positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP)
        context.log.info(
            f"bybit_4edge_levered_nav: T0={out['t0']}, {out['days']}d through {out['last_date']} "
            f"(+{out['total_return'] * 100:.2f}%, Sharpe {out['sharpe']:.2f}, "
            f"reinception={out['reinception']})")
        return {"inception": out["t0"], "last_date": out["last_date"], "forward_days": out["days"],
                "forward_return_pct": out["total_return"] * 100, "forward_sharpe": out["sharpe"]}
    except (urllib.error.URLError, TimeoutError, ConnectionError, OSError) as e:  # transient: don't kill the run
        context.log.warning(f"bybit_4edge_levered_nav step failed (record unchanged): {e}")
        return {"inception": None, "last_date": None, "forward_days": 0}
    finally:
        store.close()


@asset(deps=[bybit_warehouse_refresh])
def positioning_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """Paper-forward POSITIONING sleeve (contrarian long/short account-ratio) as its OWN edge — the standout
    constituent of the Bybit 4-edge book. Runs AFTER the warehouse refresh on the LIQUID universe over
    `bybit_features`, runs the strategy through the deterministic engine (DB anchor + recompute — T0 frozen
    on first run, idempotent). Records directly into the `forward_nav`/`strategy_anchor` cockpit tables (the
    DB is the SSOT; no JSON state to corrupt)."""
    import datetime as _dt
    import urllib.error

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.config import get_settings

    s = get_settings()
    store = TimescaleFeatureStore(s.operational_dsn, table="bybit_features")
    try:
        universe = liquid_universe(store) or None   # None → full universe (empty warehouse → empty series)
        from fxhnt.application.bybit_forward_track import POSITIONING_COIN_GROSS_CAP, POSITIONING_TAIL_CAP_K
        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)
        out = build_positioning_forward_nav(
            repo, store, today=today, at=at,
            universe=universe, cost_bps=s.bybit_cost_bps,
            positioning_capacity_capital=s.paper_capital, positioning_tail_cap_k=POSITIONING_TAIL_CAP_K,
            positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP)
        context.log.info(
            f"positioning_nav: T0={out['t0']}, {out['days']}d through {out['last_date']} "
            f"(+{out['total_return'] * 100:.2f}%, Sharpe {out['sharpe']:.2f}, "
            f"reinception={out['reinception']})")
        return {"inception": out["t0"], "last_date": out["last_date"], "forward_days": out["days"],
                "forward_return_pct": out["total_return"] * 100, "forward_sharpe": out["sharpe"]}
    except (urllib.error.URLError, TimeoutError, ConnectionError, OSError) as e:  # transient: don't kill the run
        context.log.warning(f"positioning_nav step failed (record unchanged): {e}")
        return {"inception": None, "last_date": None, "forward_days": 0}
    finally:
        store.close()


@asset(deps=[bybit_warehouse_refresh])
def xsfunding_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """Paper-forward cross-sectional crypto funding-dispersion — the SINGLE xsfunding sleeve as its own edge,
    on BYBIT (`bybit_features`), the venue it would actually trade (Phase 0b venue consolidation — this track
    moved off the standalone legacy-venue `XsFundingForward` construction; xsfunding still
    runs as a sleeve inside the `bybit_4edge` deploy book, unchanged). Runs AFTER the warehouse refresh on the
    LIQUID universe over `bybit_features`, through the deterministic engine (DB anchor + recompute — T0
    frozen on first run, idempotent). Its reconciliation ref is written by the weekly Bybit book persist
    (`_persist_bybit_book`), like positioning — NOT inline here."""
    import datetime as _dt
    import urllib.error

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.config import get_settings

    s = get_settings()
    store = TimescaleFeatureStore(s.operational_dsn, table="bybit_features")
    try:
        universe = liquid_universe(store) or None
        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)
        out = build_bybit_sleeve_forward_nav(repo, store, "xsfunding", today=today, at=at,
                                             universe=universe, cost_bps=s.bybit_cost_bps)
        context.log.info(f"xsfunding_nav: T0={out['t0']}, {out['days']}d through {out['last_date']} "
                         f"(reinception={out['reinception']})")
        return {"inception": out["t0"], "last_date": out["last_date"], "forward_days": out["days"]}
    except (urllib.error.URLError, TimeoutError, ConnectionError, OSError) as e:
        context.log.warning(f"xsfunding_nav step failed (record unchanged): {e}")
        return {"inception": None, "last_date": None, "forward_days": 0}
    finally:
        store.close()


@asset(deps=[bybit_warehouse_refresh])
def unlock_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """Paper-forward token-unlock dilution shorts — the SINGLE unlock sleeve as its own edge, on BYBIT
    (`bybit_features`), the venue it would actually trade (Phase 0b venue consolidation — this track moved
    off the standalone legacy-venue `UnlockShortForward` construction; unlock still runs
    as a sleeve inside the `bybit_4edge` deploy book, unchanged). Runs AFTER the warehouse refresh on the
    LIQUID universe over `bybit_features`, through the deterministic engine (DB anchor + recompute — T0
    frozen on first run, idempotent). Calendar = DefiLlama cliff events, read (never fetched live) from the
    operational `unlock_events` table via `load_unlock_events` — the SAME loader `_persist_bybit_book`/
    `bybit_paper_book` use — with a missing/errored calendar degrading to `[]` (never fabricated). Its
    reconciliation ref is written by the weekly Bybit book persist (`_persist_bybit_book`), like
    positioning/xsfunding — NOT inline here."""
    import datetime as _dt
    import urllib.error

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
    from fxhnt.application.bybit_liquidity import liquid_universe
    from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
    from fxhnt.config import get_settings

    s = get_settings()
    store = TimescaleFeatureStore(s.operational_dsn, table="bybit_features")
    try:
        universe = liquid_universe(store) or None
        try:
            unlock_events = load_unlock_events(operational_unlock_repo(s), f"{_data_dir()}/unlock_calendar.json")
        except Exception as exc:  # noqa: BLE001 — absent calendar → sleeve simply has no events, never fabricated
            context.log.warning(f"unlock_nav: could not load unlock calendar: {exc}")
            unlock_events = []
        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)
        out = build_bybit_sleeve_forward_nav(repo, store, "unlock", today=today, at=at,
                                             universe=universe, cost_bps=s.bybit_cost_bps,
                                             unlock_events=unlock_events)
        context.log.info(f"unlock_nav: T0={out['t0']}, {out['days']}d through {out['last_date']} "
                         f"(reinception={out['reinception']})")
        return {"inception": out["t0"], "last_date": out["last_date"], "forward_days": out["days"]}
    except (urllib.error.URLError, TimeoutError, ConnectionError, OSError) as e:
        context.log.warning(f"unlock_nav step failed (record unchanged): {e}")
        return {"inception": None, "last_date": None, "forward_days": 0}
    finally:
        store.close()


@asset(deps=[bybit_warehouse_refresh])
def bybit_paper_book(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """The Bybit LIVE paper book — per-symbol positions, marked to live Bybit prices (the deploy venue's real
    live book, positions/trades/MTM, not only the returns/forward-NAV `bybit_4edge_nav` track). Runs AFTER
    the warehouse refresh on the LIQUID universe over `bybit_features`, derives the naive-eq-wt 4-edge book's
    per-symbol target positions on the latest close, diffs vs the prior bybit book → trades, charges the
    5.5bp turnover, and persists into `PaperRepo` (bybit is its sole venue; the orphaned rows under the
    retired combined book's venue values, Phase 0b Task 7d, are untouched by this asset and vice versa).

    READ-ONLY against the warehouse (only the paper store is written). Gated on `paper_enabled`. A genuinely
    TRANSIENT (network/API) hiccup is best-effort (warn + skip so the nightly run survives); any OTHER error
    is SURFACED (logged at error + re-raised) rather than silently swallowed — a silent swallow froze
    `paper_nav`/`paper_positions` at the last success with no signal, which is exactly the invisible-freeze
    failure we must never repeat."""
    import datetime as dt
    import urllib.error

    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
    from fxhnt.config import get_settings

    s = get_settings()
    if not s.paper_enabled:
        context.log.info("bybit_paper_book: paper_enabled=False, skipping")
        return {"positions": 0, "enabled": False}
    store = TimescaleFeatureStore(s.operational_dsn, table="bybit_features")
    try:
        repo = PaperRepo(s.operational_dsn)
        repo.migrate()
        universe = liquid_universe(store) or None   # None → full universe (empty warehouse → empty book)
        unlock_events = load_unlock_events(operational_unlock_repo(s), f"{_data_dir()}/unlock_calendar.json")
        n = build_bybit_paper_book(
            repo, store, capital=s.paper_capital, at=dt.datetime.now(dt.UTC),
            universe=universe, unlock_events=unlock_events, enabled=True,
            positioning_coin_gross_cap=POSITIONING_COIN_GROSS_CAP)
        context.log.info(f"bybit_paper_book: persisted {n} open positions (venue=bybit)")
        return {"positions": n, "enabled": True}
    except (urllib.error.URLError, TimeoutError, ConnectionError, OSError) as e:  # transient: don't kill the run
        context.log.warning(f"bybit_paper_book skipped (transient, book unchanged): {e}")
        return {"positions": 0, "enabled": True, "transient": True}
    except Exception as e:
        # NOT a silent swallow: a logic/data/state error here previously froze the venue=bybit paper book at
        # the last success with no signal. Surface it (fail this asset loudly) so the freeze is visible; the
        # asset has no downstream deps, so this does not block the forward-nav track.
        context.log.error(f"bybit_paper_book FAILED (surfacing, not silently freezing the book): {e}")
        raise
    finally:
        store.close()


@asset(deps=[sixtyforty_nav, xsfunding_nav, unlock_nav, multistrat_nav, bybit_4edge_nav,
             bybit_4edge_levered_nav, positioning_nav])
def cockpit_forward(context: AssetExecutionContext) -> None:
    """Evaluate + persist the forward gate from the DB record the engine already wrote (all_summaries +
    nav_history — see `forward_ingest.ingest_forward_state`), then auto-promote any `promote_on_pass`
    track that just cleared its gate (research→deploy). After promotion, also records the B2a
    strategy-level allocation snapshot for the (now-fresh) deploy set."""
    from fxhnt.application.allocation_ingest import record_allocation
    from fxhnt.application.forward_health import evaluate_health
    from fxhnt.application.forward_ingest import evaluate_promotions, ingest_forward_state
    from fxhnt.config import get_settings

    dsn = get_settings().operational_dsn
    ingest_forward_state(dsn)
    # Runtime health axis (SEPARATE from the gate): surface ANY track that stopped updating LOUDLY — a red
    # cockpit badge + an ERROR log the FIRST night, so a stale recompute (a dead strategy's silent rot) can never sit as
    # a quiet WAIT again. evaluate_health logs at ERROR with the unhealthy list; also surface the count here.
    unhealthy = [v for v in evaluate_health(dsn) if not v.ok]
    if unhealthy:
        context.log.error(
            f"cockpit_forward HEALTH: {len(unhealthy)} unhealthy track(s) — "
            + "; ".join(f"{v.strategy_id}={v.health}({v.reason})" for v in unhealthy))
    promoted = evaluate_promotions(dsn)   # one-way research→deploy on gate PASS (fresh gate from the ingest)
    context.log.info(f"cockpit_forward ingested (promoted this run: {promoted or 'none'})")

    # Honest-reference sizing: build the store/spread/unlock-events the SAME way `honest-reference-report`
    # does and pass them to record_allocation (capacity-aware, stable backtest series). GRACEFUL: any failure
    # here (warehouse unreachable, missing spread SSOT, etc.) must NEVER crash the nightly — fall back to the
    # nav_history path (store=None, today's behavior). Store closed in a finally.
    import dataclasses

    from fxhnt.application.allocation_policy import resolve_allocation_policy
    from fxhnt.application.fund_config import read_fund_config

    _cfg = read_fund_config(dsn)
    equity = _cfg.capital
    _policy = dataclasses.replace(resolve_allocation_policy(get_settings()), target_vol=_cfg.target_vol,
                                  kelly_fraction=_cfg.kelly_fraction, max_venue_weight=_cfg.max_venue_weight)
    try:
        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.unlock_calendar_loader import load_unlock_events, operational_unlock_repo

        s = get_settings()
        raw = TimescaleFeatureStore(s.operational_dsn, table="bybit_features")
        try:
            universe = liquid_universe(raw) or None   # None → raw store (documented OOM-risk on empty universe)
            bounded = _bounded_store(raw, universe)
            try:
                unlock_events = load_unlock_events(operational_unlock_repo(s),
                                                   f"{_data_dir()}/unlock_calendar.json")
            except Exception:   # noqa: BLE001 — absent calendar → unlock sleeve omitted, not a crash
                unlock_events = []
            spread_bps = PaperRepo(s.operational_dsn).read_real_spread_by_coin()
            allocated = record_allocation(dsn, equity=equity, store=bounded,
                                          spread_bps=spread_bps, unlock_events=unlock_events, policy=_policy)
        finally:
            raw.close()
    except Exception as e:   # noqa: BLE001 — honest sizing is best-effort; nightly must never crash on it
        context.log.warning(
            f"cockpit_forward: honest-reference store unavailable, falling back to nav_history sizing: {e}")
        # deploy-set is fresh (B2a) — nav_history fallback
        allocated = record_allocation(dsn, equity=equity, policy=_policy)
    context.log.info(f"cockpit_forward: allocated {len(allocated)} strategies (promoted: {promoted or 'none'})")
