"""Repository for the cockpit's paper-book tables (positions + trades). Mirrors `ForwardNavRepo`'s
engine/StaticPool construction so tests run against in-memory SQLite without Postgres. `replace_positions`
rewrites a run_date's open book; `latest_positions` returns the most recent run_date's holdings; trades are
append-only and read newest-first. No exchange/order side effects — this is a read-model store."""
from __future__ import annotations

import datetime as dt
from dataclasses import dataclass
from typing import Any

from sqlalchemy import create_engine, delete, select
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from fxhnt.adapters.persistence.cockpit_models import (
    BybitRealSpreadRow,
    BybitSleeveRetRow,
    CockpitBase,
    CompareMeasuredCacheRow,
    PaperBookStateRow,
    PaperNavRow,
    PaperNavSummaryRow,
    PaperPositionRow,
    PaperShadowPositionRow,
    PaperSleeveRetRow,
    PaperTradeRow,
    SimCurveRetRow,
)
from fxhnt.adapters.persistence.paper_book_migration import migrate_paper_book_schema
from fxhnt.domain.paper import Position, Trade


@dataclass(frozen=True)
class NavPoint:
    """One paper-book equity sample (one run_date). `run_date` is an ISO string for UI/serialization."""
    run_date: str
    equity: float
    realized: float
    unrealized: float


def _parse_ts(ts: str) -> dt.datetime:
    """Trade.ts is an ISO string (e.g. '2026-06-21T23:30:00Z'); store as a DateTime."""
    s = ts.replace("Z", "+00:00")
    try:
        return dt.datetime.fromisoformat(s)
    except ValueError:
        # Fallback for non-ISO sentinels used in tests (e.g. "t"); pin to epoch UTC.
        return dt.datetime(1970, 1, 1, tzinfo=dt.UTC)


class PaperRepo:
    def __init__(self, dsn: str) -> None:
        # StaticPool keeps a single in-memory SQLite connection alive across sessions; only applied for the
        # in-memory SQLite case (tests). StaticPool is NOT used for Postgres.
        kw: dict[str, Any] = {"future": True}
        if dsn == "sqlite://" or (dsn.startswith("sqlite://") and ":memory:" in dsn):
            kw |= {"connect_args": {"check_same_thread": False}, "poolclass": StaticPool}
        self._engine = create_engine(dsn, **kw)

    def migrate(self) -> None:
        CockpitBase.metadata.create_all(self._engine)
        # Reverse migration: drop the vestigial single-valued discriminator column from the paper-book tables
        # so the schema is genuinely single-book (bybit is the sole paper book). Lives in its own module so
        # this repo carries no discriminator logic. Postgres-only + idempotent + a pure NO-OP after the first
        # run (heavy DDL on every startup crashloops the cockpit); it also (re)creates the (ts, id) index the
        # newest-first trades read backward-scans.
        migrate_paper_book_schema(self._engine)

    def replace_positions(self, run_date: str, positions: list[Position],
                          weight_by_sleeve: dict[str, float], at: dt.datetime) -> None:
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            s.execute(delete(PaperPositionRow).where(PaperPositionRow.run_date == rd))
            for p in positions:
                s.add(PaperPositionRow(
                    run_date=rd, sleeve=p.sleeve, symbol=p.symbol, qty=p.qty,
                    entry_price=p.entry_price, entry_date=p.entry_date,
                    weight=float(weight_by_sleeve.get(p.sleeve, 0.0)), src_updated_at=at))
            # Record the latest run_date so a full-exit (empty book) is still the current view, and never
            # roll the marker backwards on a replay of an older date.
            cur = s.scalar(select(PaperBookStateRow.value).where(
                PaperBookStateRow.key == self._LATEST_RD_KEY))
            if cur is None or rd.toordinal() >= int(cur):
                s.merge(PaperBookStateRow(key=self._LATEST_RD_KEY,
                                          value=float(rd.toordinal())))
            s.commit()

    _LATEST_RD_KEY = "latest_run_date"

    def _latest_run_date(self, s: Session) -> dt.date | None:
        """The most recent rebalanced run_date. Read from state (set on every replace_positions, so an empty
        book — a full exit — is still recorded) and fall back to the max row date for legacy data."""
        v = s.scalar(select(PaperBookStateRow.value).where(
            PaperBookStateRow.key == self._LATEST_RD_KEY))
        if v is not None:
            return dt.date.fromordinal(int(v))
        return s.scalar(select(PaperPositionRow.run_date)
                        .order_by(PaperPositionRow.run_date.desc()).limit(1))

    def latest_positions(self) -> list[Position]:
        with Session(self._engine) as s:
            rd = self._latest_run_date(s)
            if rd is None:
                return []
            rows = s.scalars(select(PaperPositionRow)
                             .where(PaperPositionRow.run_date == rd)
                             .order_by(PaperPositionRow.sleeve, PaperPositionRow.symbol))
            return [Position(r.sleeve, r.symbol, r.qty, r.entry_price, r.entry_date) for r in rows]

    def positions_before(self, run_date: str) -> list[Position]:
        """The held book as of the most recent REBALANCED run_date STRICTLY BEFORE `run_date`. Used as the diff
        base so a re-run of the same run_date reproduces the same trades (prior is independent of the current
        run). The prior date is anchored on the nav rows (one per persisted run_date) — NOT on position rows —
        so a FLAT book (full exit / kill-switch, which persists zero position rows) is honored as the prior:
        the day after a kill correctly sees an EMPTY book, never the last non-empty book before the kill."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            prev = s.scalar(select(PaperNavRow.run_date)
                            .where(PaperNavRow.run_date < rd)
                            .order_by(PaperNavRow.run_date.desc()).limit(1))
            if prev is None:
                # No nav anchor (e.g. positions persisted without a nav row) — fall back to the latest
                # position-row date strictly before, preserving the legacy behavior for those call paths.
                prev = s.scalar(select(PaperPositionRow.run_date)
                                .where(PaperPositionRow.run_date < rd)
                                .order_by(PaperPositionRow.run_date.desc()).limit(1))
                if prev is None:
                    return []
            rows = s.scalars(select(PaperPositionRow)
                             .where(PaperPositionRow.run_date == prev)
                             .order_by(PaperPositionRow.sleeve, PaperPositionRow.symbol))
            return [Position(r.sleeve, r.symbol, r.qty, r.entry_price, r.entry_date) for r in rows]

    def replace_shadow_positions(self, run_date: str, positions: list[Position], *, at: dt.datetime) -> None:
        """Idempotent per run_date: rewrite that date's UNIT shadow book (never throttled)."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            s.execute(delete(PaperShadowPositionRow).where(
                PaperShadowPositionRow.run_date == rd))
            for p in positions:
                s.add(PaperShadowPositionRow(
                    run_date=rd, sleeve=p.sleeve, symbol=p.symbol, qty=p.qty,
                    entry_price=p.entry_price, entry_date=p.entry_date, src_updated_at=at))
            s.commit()

    def shadow_positions_before(self, run_date: str) -> list[Position]:
        """The UNIT shadow book as of the most recent run_date STRICTLY BEFORE `run_date`."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            prev = s.scalar(select(PaperShadowPositionRow.run_date)
                            .where(PaperShadowPositionRow.run_date < rd)
                            .order_by(PaperShadowPositionRow.run_date.desc()).limit(1))
            if prev is None:
                return []
            rows = s.scalars(select(PaperShadowPositionRow)
                             .where(PaperShadowPositionRow.run_date == prev)
                             .order_by(PaperShadowPositionRow.sleeve, PaperShadowPositionRow.symbol))
            return [Position(r.sleeve, r.symbol, r.qty, r.entry_price, r.entry_date) for r in rows]

    def shadow_run_dates(self) -> list[str]:
        """Every run_date with a UNIT shadow book, ascending (ISO strings) — the date axis the per-coin
        compare walks to surface within-sleeve weights. READ-ONLY."""
        with Session(self._engine) as s:
            rows = s.scalars(select(PaperShadowPositionRow.run_date)
                             .distinct().order_by(PaperShadowPositionRow.run_date.asc()))
            return [r.isoformat() for r in rows]

    def shadow_positions_on(self, run_date: str) -> list[Position]:
        """The UNIT shadow book persisted for EXACTLY `run_date` (sleeve+symbol ordered). READ-ONLY."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            rows = s.scalars(select(PaperShadowPositionRow)
                             .where(PaperShadowPositionRow.run_date == rd)
                             .order_by(PaperShadowPositionRow.sleeve, PaperShadowPositionRow.symbol))
            return [Position(r.sleeve, r.symbol, r.qty, r.entry_price, r.entry_date) for r in rows]

    def weight_for(self, sleeve: str) -> float:
        with Session(self._engine) as s:
            rd = self._latest_run_date(s)
            if rd is None:
                return 0.0
            w = s.scalar(select(PaperPositionRow.weight)
                         .where(PaperPositionRow.run_date == rd,
                                PaperPositionRow.sleeve == sleeve).limit(1))
            return float(w) if w is not None else 0.0

    def weights_for(self, sleeves: list[str]) -> dict[str, float]:
        """All requested sleeves' weights for the latest run_date in ONE query. The live view called
        weight_for per sleeve, and each call ALSO re-read the latest run_date — N sleeves = ~2N queries. This
        is two queries total (latest run_date + the weights), keyed back to the sleeves."""
        out = {sl: 0.0 for sl in sleeves}
        if not sleeves:
            return out
        with Session(self._engine) as s:
            rd = self._latest_run_date(s)
            if rd is None:
                return out
            rows = s.execute(select(PaperPositionRow.sleeve, PaperPositionRow.weight)
                             .where(PaperPositionRow.run_date == rd,
                                    PaperPositionRow.sleeve.in_(list(sleeves)))).all()
            for sleeve, w in rows:                       # all positions in a sleeve share its weight
                if w is not None:
                    out[sleeve] = float(w)
            return out

    def record_trades(self, trades: list[Trade], at: dt.datetime,
                      run_date: str | None = None) -> None:
        """Append trades. `run_date` keys them so a later `replace_trades` can rewrite that date's ledger;
        if omitted the trade's own ts date is used."""
        with Session(self._engine) as s:
            for t in trades:
                ts = _parse_ts(t.ts)
                rd = dt.date.fromisoformat(run_date) if run_date is not None else ts.date()
                s.add(PaperTradeRow(
                    run_date=rd, ts=ts, sleeve=t.sleeve, symbol=t.symbol, side=t.side,
                    qty=t.qty, price=t.price, reason=t.reason))
            s.commit()

    def replace_trades(self, run_date: str, trades: list[Trade], at: dt.datetime) -> None:
        """Idempotent per run_date: delete that run_date's trades, then insert the supplied set. A re-run of
        the same run_date thus replaces rather than appends (mirrors `replace_positions`)."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            s.execute(delete(PaperTradeRow).where(PaperTradeRow.run_date == rd))
            for t in trades:
                s.add(PaperTradeRow(
                    run_date=rd, ts=_parse_ts(t.ts), sleeve=t.sleeve, symbol=t.symbol,
                    side=t.side, qty=t.qty, price=t.price, reason=t.reason))
            s.commit()

    def recent_trades(self, n: int) -> list[Trade]:
        with Session(self._engine) as s:
            rows = s.scalars(select(PaperTradeRow)
                             .order_by(PaperTradeRow.ts.desc(), PaperTradeRow.id.desc())
                             .limit(n))
            return [Trade(ts=r.ts.isoformat(), sleeve=r.sleeve, symbol=r.symbol, side=r.side,
                          qty=r.qty, price=r.price, reason=r.reason) for r in rows]

    # Realized PnL is stored per run_date (`realized:<date>`) so a replay of a run_date OVERWRITES rather
    # than double-books; a free-floating `realized` key holds any non-dated adjustments. `realized_pnl()`
    # sums them all.
    _REALIZED_KEY = "realized"
    _REALIZED_PREFIX = "realized:"

    def realized_pnl(self) -> float:
        """Cumulative realized PnL across all closed/reduced positions (0.0 if never set)."""
        with Session(self._engine) as s:
            rows = s.scalars(select(PaperBookStateRow).where(
                (PaperBookStateRow.key == self._REALIZED_KEY)
                | PaperBookStateRow.key.like(self._REALIZED_PREFIX + "%")))
            return float(sum(r.value for r in rows))

    def realized_pnl_through(self, run_date: str) -> float:
        """Cumulative realized PnL booked ON OR BEFORE `run_date` — the non-dated `realized` key plus every
        `realized:<d>` key with `d <= run_date` (ISO date strings compare correctly).

        NOTE (A3f): equity is now compounding mark-to-market (nav_equity_before + prior-book MTM); neither
        the replay curve nor the live view uses this realized total any more — the realized:* slivers are a
        display-only ledger. Retained for that ledger / back-compat; not the equity driver."""
        with Session(self._engine) as s:
            rows = s.scalars(select(PaperBookStateRow).where(
                (PaperBookStateRow.key == self._REALIZED_KEY)
                | PaperBookStateRow.key.like(self._REALIZED_PREFIX + "%")))
            total = 0.0
            for r in rows:
                if r.key == self._REALIZED_KEY or r.key[len(self._REALIZED_PREFIX):] <= run_date:
                    total += r.value
            return float(total)

    def set_realized_for_run_date(self, run_date: str, value: float) -> None:
        """Idempotent per run_date: set (overwrite) that run_date's realized contribution."""
        with Session(self._engine) as s:
            s.merge(PaperBookStateRow(key=self._REALIZED_PREFIX + run_date,
                                      value=float(value)))
            s.commit()

    def add_realized(self, delta: float) -> None:
        """Add `delta` to the (non-dated) running realized total — read-modify-write under one session."""
        with Session(self._engine) as s:
            row = s.get(PaperBookStateRow, self._REALIZED_KEY)
            if row is None:
                s.add(PaperBookStateRow(key=self._REALIZED_KEY, value=float(delta)))
            else:
                row.value = float(row.value) + float(delta)
            s.commit()

    # ---- paper_nav equity series (replay) -------------------------------------------------------
    def upsert_nav(self, run_date: str, *, capital: float, realized: float,
                   unrealized: float, equity: float, at: dt.datetime) -> None:
        """Idempotent per run_date: write (or overwrite) that date's equity row — a re-run replaces it."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            s.merge(PaperNavRow(run_date=rd, capital=float(capital),
                                realized=float(realized), unrealized=float(unrealized),
                                equity=float(equity), src_updated_at=at))
            s.commit()

    def nav_history(self) -> list[NavPoint]:
        """The equity series ascending by run_date (one point per date)."""
        with Session(self._engine) as s:
            rows = s.scalars(select(PaperNavRow)
                             .order_by(PaperNavRow.run_date.asc()))
            return [NavPoint(run_date=r.run_date.isoformat(), equity=r.equity,
                             realized=r.realized, unrealized=r.unrealized) for r in rows]

    def nav_equity_before(self, run_date: str) -> float | None:
        """The most recent paper_nav equity STRICTLY BEFORE `run_date` (the prior equity to compound from)."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            return s.scalar(select(PaperNavRow.equity)
                            .where(PaperNavRow.run_date < rd)
                            .order_by(PaperNavRow.run_date.desc()).limit(1))

    def latest_nav_equity(self) -> float | None:
        """The most recent paper_nav equity (for the live view's curve baseline)."""
        with Session(self._engine) as s:
            return s.scalar(select(PaperNavRow.equity)
                            .order_by(PaperNavRow.run_date.desc()).limit(1))

    def latest_nav_run_date(self) -> str | None:
        """ISO date of the most recent persisted paper_nav row, or None if the series is empty. The
        incremental backfill resumes STRICTLY AFTER this date (the last fully-persisted date)."""
        with Session(self._engine) as s:
            rd = s.scalar(select(PaperNavRow.run_date)
                          .order_by(PaperNavRow.run_date.desc()).limit(1))
            return rd.isoformat() if rd is not None else None

    # ---- paper_nav_summary: the MATERIALIZED headline (stats + curve), read 1 row not 3650 ----------
    def read_nav_summary(self) -> dict[str, Any] | None:
        """The MATERIALIZED headline nav-summary for the (sole) bybit book — the stats + `since`/`has_record`
        + the pre-rendered curve SVG + the nav version `nav_run_date` — or None when none is stored yet. ONE
        1-row read — NOT the full nav series. Read-only. The caller compares `nav_run_date` to the live latest
        run_date to decide fresh-vs-stale."""
        with Session(self._engine) as s:
            row = s.get(PaperNavSummaryRow, 1)
            if row is None:
                return None
            return {"nav_run_date": row.nav_run_date, "final_equity": row.final_equity,
                    "total_return_pct": row.total_return_pct, "sharpe": row.sharpe,
                    "max_dd_pct": row.max_dd_pct, "since": row.since,
                    "has_record": row.has_record, "curve_svg": row.curve_svg}

    def upsert_nav_summary(self, payload: dict[str, Any], *, at: dt.datetime) -> None:
        """Idempotent: write/overwrite the materialized headline nav-summary for the (sole) bybit book (the
        stats + `since`/`has_record` + rendered `curve_svg`, versioned by `payload['nav_run_date']`). Replaced
        on every refresh."""
        with Session(self._engine) as s:
            s.merge(PaperNavSummaryRow(
                id=1, nav_run_date=str(payload["nav_run_date"]),
                final_equity=float(payload["final_equity"]),
                total_return_pct=float(payload["total_return_pct"]),
                sharpe=float(payload["sharpe"]), max_dd_pct=float(payload["max_dd_pct"]),
                since=str(payload["since"]), has_record=bool(payload["has_record"]),
                curve_svg=str(payload["curve_svg"]), computed_at=at))
            s.commit()

    # ---- BULK chunked writes (batched backfill) -------------------------------------------------
    # Each takes the rows for a CHUNK of dates and writes them under ONE session (one commit), instead
    # of one session+commit per date per table. The per-date methods above remain the single-date path
    # (live snapshot, tests); these are the same upsert semantics applied to many dates at once. The
    # batched backfill commits once per chunk → a crash loses at most the current chunk (resumable via
    # the incremental mode), while the per-date round-trips collapse from O(dates·tables) to O(chunks).
    def bulk_replace_positions(self, by_date: dict[str, tuple[list[Position], dict[str, float]]],
                               at: dt.datetime) -> None:
        """For each run_date, rewrite its open book (delete that date's rows, insert the supplied set) and
        advance the latest-run_date state marker — same as `replace_positions`, batched in one session."""
        if not by_date:
            return
        with Session(self._engine) as s:
            cur = s.scalar(select(PaperBookStateRow.value).where(
                PaperBookStateRow.key == self._LATEST_RD_KEY))
            cur_ord = int(cur) if cur is not None else None
            max_ord = cur_ord
            for run_date, (positions, weight_by_sleeve) in by_date.items():
                rd = dt.date.fromisoformat(run_date)
                s.execute(delete(PaperPositionRow).where(PaperPositionRow.run_date == rd))
                for p in positions:
                    s.add(PaperPositionRow(
                        run_date=rd, sleeve=p.sleeve, symbol=p.symbol, qty=p.qty,
                        entry_price=p.entry_price, entry_date=p.entry_date,
                        weight=float(weight_by_sleeve.get(p.sleeve, 0.0)), src_updated_at=at))
                if max_ord is None or rd.toordinal() >= max_ord:
                    max_ord = rd.toordinal()
            if max_ord is not None and (cur_ord is None or max_ord >= cur_ord):
                s.merge(PaperBookStateRow(key=self._LATEST_RD_KEY,
                                          value=float(max_ord)))
            s.commit()

    def bulk_replace_trades(self, by_date: dict[str, list[Trade]], at: dt.datetime) -> None:
        """For each run_date, replace its trade ledger (delete then insert) — `replace_trades`, batched."""
        if not by_date:
            return
        with Session(self._engine) as s:
            for run_date, trades in by_date.items():
                rd = dt.date.fromisoformat(run_date)
                s.execute(delete(PaperTradeRow).where(PaperTradeRow.run_date == rd))
                for t in trades:
                    s.add(PaperTradeRow(
                        run_date=rd, ts=_parse_ts(t.ts), sleeve=t.sleeve,
                        symbol=t.symbol, side=t.side, qty=t.qty, price=t.price, reason=t.reason))
            s.commit()

    def bulk_set_realized(self, by_date: dict[str, float]) -> None:
        """Overwrite each run_date's realized contribution (`set_realized_for_run_date`), batched."""
        if not by_date:
            return
        with Session(self._engine) as s:
            for run_date, value in by_date.items():
                s.merge(PaperBookStateRow(key=self._REALIZED_PREFIX + run_date,
                                          value=float(value)))
            s.commit()

    def bulk_upsert_nav(self, rows: list[dict[str, Any]]) -> None:
        """Write/overwrite each date's equity row (`upsert_nav`), batched. Each dict carries run_date,
        capital, realized, unrealized, equity, at."""
        if not rows:
            return
        with Session(self._engine) as s:
            for r in rows:
                s.merge(PaperNavRow(
                    run_date=dt.date.fromisoformat(r["run_date"]),
                    capital=float(r["capital"]), realized=float(r["realized"]),
                    unrealized=float(r["unrealized"]),
                    equity=float(r["equity"]), src_updated_at=r["at"]))
            s.commit()

    def bulk_replace_shadow_positions(self, by_date: dict[str, list[Position]], at: dt.datetime) -> None:
        """For each run_date, rewrite its UNIT shadow book (`replace_shadow_positions`), batched."""
        if not by_date:
            return
        with Session(self._engine) as s:
            for run_date, positions in by_date.items():
                rd = dt.date.fromisoformat(run_date)
                s.execute(delete(PaperShadowPositionRow).where(
                    PaperShadowPositionRow.run_date == rd))
                for p in positions:
                    s.add(PaperShadowPositionRow(
                        run_date=rd, sleeve=p.sleeve, symbol=p.symbol, qty=p.qty,
                        entry_price=p.entry_price, entry_date=p.entry_date, src_updated_at=at))
            s.commit()

    def bulk_upsert_sleeve_ret(self, rows: list[tuple[str, str, float]], at: dt.datetime) -> None:
        """Write/overwrite each (run_date, sleeve) return (`upsert_sleeve_ret`), batched. Rows are
        (run_date_iso, sleeve, ret)."""
        if not rows:
            return
        with Session(self._engine) as s:
            for run_date, sleeve, ret in rows:
                s.merge(PaperSleeveRetRow(run_date=dt.date.fromisoformat(run_date),
                                          sleeve=sleeve, ret=float(ret), src_updated_at=at))
            s.commit()

    def positions_at(self, run_date: str) -> list[Position]:
        """The book persisted for exactly `run_date`."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            rows = s.scalars(select(PaperPositionRow)
                             .where(PaperPositionRow.run_date == rd)
                             .order_by(PaperPositionRow.sleeve, PaperPositionRow.symbol))
            return [Position(r.sleeve, r.symbol, r.qty, r.entry_price, r.entry_date) for r in rows]

    def trades_on(self, run_date: str) -> list[Trade]:
        """The trades keyed to `run_date` (paper_trades.run_date)."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            rows = s.scalars(select(PaperTradeRow)
                             .where(PaperTradeRow.run_date == rd)
                             .order_by(PaperTradeRow.ts.asc(), PaperTradeRow.id.asc()))
            return [Trade(ts=r.ts.isoformat(), sleeve=r.sleeve, symbol=r.symbol, side=r.side,
                          qty=r.qty, price=r.price, reason=r.reason) for r in rows]

    # ---- paper_sleeve_ret per-sleeve return series --------------------------------------------------
    def upsert_sleeve_ret(self, run_date: str, sleeve: str, ret: float, *, at: dt.datetime) -> None:
        """Idempotent per (run_date, sleeve): write/overwrite that date's per-sleeve simulated return."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            s.merge(PaperSleeveRetRow(run_date=rd, sleeve=sleeve, ret=float(ret),
                                      src_updated_at=at))
            s.commit()

    def sleeve_returns(self, *, before: str) -> dict[str, dict[int, float]]:
        """Trailing per-sleeve return series STRICTLY before `before`, as {sleeve: {epoch_day: ret}}
        (epoch_day = date.toordinal()) — the shape isv_adaptive_weights expects."""
        cutoff = dt.date.fromisoformat(before)
        out: dict[str, dict[int, float]] = {}
        with Session(self._engine) as s:
            rows = s.scalars(select(PaperSleeveRetRow)
                             .where(PaperSleeveRetRow.run_date < cutoff)
                             .order_by(PaperSleeveRetRow.run_date.asc()))
            for r in rows:
                out.setdefault(r.sleeve, {})[r.run_date.toordinal()] = r.ret
        return out

    # ---- bybit_sleeve_ret: the PRECOMPUTED Bybit 4-edge backtest sleeves the sim reads ---------------
    def upsert_bybit_sleeve_ret(self, run_date: str, sleeve: str, ret: float, *, at: dt.datetime) -> None:
        """Idempotent per (run_date, sleeve): write/overwrite that Bybit sleeve's per-day return. The
        own-memory `fxhnt bybit-persist-sleeve-ret` Job persists the full 4-sleeve series here so the cockpit
        sim configures a precomputed backtest WITHOUT ever running the heavy 4-edge compute on a web request.
        (Writer moved OUT of the nightly daemon asset — that heavy compute OOM-killed the 4Gi daemon.)"""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            s.merge(BybitSleeveRetRow(run_date=rd, sleeve=sleeve, ret=float(ret), src_updated_at=at))
            s.commit()

    def bybit_sleeve_returns(self) -> dict[str, dict[int, float]]:
        """The FULL precomputed Bybit per-sleeve return series as {sleeve: {epoch_day: ret}}
        (epoch_day = date.toordinal()) — the shape the sim's naive eq-wt / cheap knobs consume. Read-only."""
        out: dict[str, dict[int, float]] = {}
        with Session(self._engine) as s:
            rows = s.scalars(select(BybitSleeveRetRow).order_by(BybitSleeveRetRow.run_date.asc()))
            for r in rows:
                out.setdefault(r.sleeve, {})[r.run_date.toordinal()] = r.ret
        return out

    # ---- sim_curve_ret: the PRECOMPUTED recompute-replay (IBKR) book curve /paper/sim reads ------------
    def upsert_sim_curve_ret(self, strategy_id: str, run_date: str, ret: float, *, at: dt.datetime) -> None:
        """Idempotent per (strategy_id, run_date): write/overwrite that recompute-replay book's (multistrat)
        per-day honest-cost return. `_persist_track_backtest_ref` writes the full inception series here
        from the SAME rows it derives the reconciliation-gate ref from, so /paper/sim NEVER re-runs an IBKR
        replay on a web request — mirrors `upsert_bybit_sleeve_ret`'s pattern for the Bybit-book sim."""
        rd = dt.date.fromisoformat(run_date)
        with Session(self._engine) as s:
            s.merge(SimCurveRetRow(strategy_id=strategy_id, run_date=rd, ret=float(ret), src_updated_at=at))
            s.commit()

    def sim_curve_returns(self, strategy_id: str) -> dict[int, float]:
        """The FULL precomputed per-day return series for one recompute-replay book (`strategy_id`), as
        {epoch_day: ret} (epoch_day = date.toordinal()) — the shape /paper/sim's IBKR curve reads. Read-only,
        scoped to this one strategy (unlike `bybit_sleeve_returns`, this table holds several IBKR books)."""
        out: dict[int, float] = {}
        with Session(self._engine) as s:
            rows = s.scalars(select(SimCurveRetRow)
                             .where(SimCurveRetRow.strategy_id == strategy_id)
                             .order_by(SimCurveRetRow.run_date.asc()))
            for r in rows:
                out[r.run_date.toordinal()] = r.ret
        return out

    # ---- compare_measured_cache: the PRECOMPUTED measured-cost COMPARE artifact ----------------------
    def upsert_compare_measured_cache(self, key: str, payload: dict[str, Any], *,
                                      at: dt.datetime) -> None:
        """Idempotent per key: write/overwrite the precomputed measured-cost COMPARE payload. The own-memory
        `fxhnt compare-measured-precompute` Job writes it; the cockpit ONLY reads it (the heavy per-coin
        measured-cost path OOM-killed the 4Gi dashboard when run live). Not book-scoped (one global artifact)."""
        with Session(self._engine) as s:
            s.merge(CompareMeasuredCacheRow(key=key, payload=payload, computed_at=at))
            s.commit()

    def read_compare_measured_cache(self, key: str) -> tuple[dict[str, Any], dt.datetime] | None:
        """`(payload, computed_at)` for the precomputed measured-cost COMPARE artifact, or None if absent.
        Read-only — the single read the cockpit's measured compare path does instead of recomputing."""
        with Session(self._engine) as s:
            row = s.get(CompareMeasuredCacheRow, key)
            if row is None:
                return None
            return dict(row.payload), row.computed_at

    # ---- bybit_real_spread: the REAL live L1 quoted spread per coin (forward-realistic cost input) ----
    def upsert_real_spreads(self, spreads: dict[str, float], *, at: dt.datetime) -> None:
        """Idempotent per coin: write/overwrite the REAL median L1 quoted spread (bps) for each coin. The
        own-memory `fxhnt bybit-spread-refresh` Job writes it (median of several samples a few seconds apart);
        the measured-cost precompute READS it to charge the Bybit liquid book the REAL spread instead of the
        high/low Corwin–Schultz proxy. Defensive: non-finite / negative spreads are skipped (never stored as a
        garbage cost). Not book-scoped (one global per-coin table)."""
        import math as _math

        with Session(self._engine) as s:
            for coin, sp in spreads.items():
                if coin is None or sp is None:
                    continue
                try:
                    v = float(sp)
                except (TypeError, ValueError):
                    continue
                if not _math.isfinite(v) or v < 0.0:
                    continue
                s.merge(BybitRealSpreadRow(coin=str(coin), spread_bps=v, fetched_at=at))
            s.commit()

    def read_real_spread_by_coin(self) -> dict[str, float]:
        """`{coin: real L1 quoted spread bps}` — the stored REAL spreads the measured cost model overrides the
        CS estimate with. Empty when none have been refreshed yet (→ the cost model uses CS for every coin).
        Read-only."""
        out: dict[str, float] = {}
        with Session(self._engine) as s:
            for r in s.scalars(select(BybitRealSpreadRow)):
                out[r.coin] = r.spread_bps
        return out
