"""Ingest Bybit daily PERP + SPOT OHLCV into a Bybit-namespaced feature store (table='bybit_features').

Per symbol: fetch {epoch_day: {close,high,low,turnover}} for the perp leg (category=linear) → write
features close/high/low/turnover, and for the spot leg (category=spot) → write spot_close/spot_high/spot_low,
at ts = epoch_day*86400 via the store's batched, idempotent upsert. (7 features total.) Feature names match
crypto_close_panel / crypto_spot_panel; the added high/low/turnover feed the liquid universe (turnover) and
the intraday squeeze-tail proxy (high/low), so Phase C's TRUSTWORTHY carry accounting reads them off the
same store Phase A's funding lands in.

Memory-bounded (one symbol, one leg at a time) and resumable-friendly (upsert overwrites in place, so
re-running is safe and cheap). Survivorship-free is the ADAPTER's / caller's job — a delisted symbol that
returns rows is ingested like any other; one that returns nothing contributes no rows and never aborts.

RESUME = DATE-RESUME per symbol (NOT symbol-skip). Each symbol resumes the perp leg from its last persisted
`turnover` day (via `store.last_day_by_symbol("turnover")` + `resume_floor_ms`), so the backward-paginating
adapter pulls only NEW days and the idempotent upsert overwrites the boundary day in place — every symbol
therefore ADVANCES its close/turnover/high/low every run. A symbol with no `turnover` yet (a clean store, or
one populated by the OLD ingest that has close/spot_close but no turnover) has last-day = None → floor falls
back to the caller's `start_ms` (None → the adapter's 2020 backstop) → a FULL back-fill of all 7 features, so
the missing-turnover backfill property is preserved. `resume=False` re-fetches every symbol's full history
(same final state — idempotent). (Before this change resume was SYMBOL-SKIP on `turnover`: once a symbol
carried turnover it was never fetched again, which FROZE close/turnover at the last ingest date for all
existing symbols — only brand-new symbols advanced — and stalled the whole Bybit forward gate, exactly as the
spot leg had before its own date-resume migration.)
"""
from __future__ import annotations

from collections.abc import Callable
from typing import Any, Protocol

from fxhnt.application.bybit_concurrency import (
    DEFAULT_WORKERS,
    fetch_symbols_concurrently,
    resume_floor_ms,
)


class _BybitKlines(Protocol):
    def daily_ohlcv(self, symbol: str, *, category: str,
                    start_ms: int | None = None) -> dict[int, dict[str, float]]: ...


def ingest_bybit_klines(
    store: Any,
    bybit: _BybitKlines,
    symbols: list[str],
    *,
    start_ms: int | None = None,
    resume: bool = True,
    workers: int = DEFAULT_WORKERS,
    progress: Callable[[str], None] | None = None,
) -> dict[str, Any]:
    """Fetch + write daily perp OHLCV (close/high/low/turnover) + spot OHLC (spot_close/spot_high/spot_low)
    for each symbol into `store` at ts=epoch_day*86400.

    Returns {"symbols": N, "perp_rows": P, "spot_rows": S, "oldest_day": D|None}: N = symbols that produced
    >=1 perp- OR spot-row, P/S = total perp/spot day-rows written, D = earliest epoch_day across both legs
    (None if nothing ingested). Skip-on-empty per leg.

    Concurrency (`workers`, default 12): the per-symbol FETCH of BOTH legs (perp + spot — each paginates the
    network-bound Bybit API) runs across a bounded thread pool; each symbol's two WRITES happen in the main
    thread as the fetch completes (serialized — no store thread-safety needed). Memory-bounded: results are
    written as completed, never collected up front, so peak memory stays ~one symbol's two legs (preserving
    the OOM-fix property). Klines are 2× funding's fetch volume, so this is where most of the wall-clock win
    lands — ~5–10× over the old sequential loop.

    Resumable (`resume=True`, default): DATE-RESUME per symbol — each symbol is fetched from its last persisted
    perp `turnover` day (`store.last_day_by_symbol("turnover")` → `resume_floor_ms`), so the backward-paginating
    adapter pulls only NEW days and the idempotent upsert overwrites the boundary day in place; EVERY symbol
    therefore ADVANCES every run (mirroring `ingest_bybit_spot`). A symbol with no `turnover` yet (clean store,
    or an OLD-ingest store that has close/spot_close but no turnover) has last-day = None → floor falls back to
    the caller's `start_ms` (None → the adapter's 2020 backstop) → a FULL back-fill of all 7 features, so the
    missing-turnover backfill property is preserved. The last-day map is queried ONCE up front (before the pool).
    The summary counts only this run's writes. Idempotent upsert ⇒ `resume=False` re-fetches everything to the
    identical state.

    (Before this change resume was SYMBOL-SKIP on `turnover`: once a symbol carried turnover it was never
    re-fetched, which FROZE its close/turnover/high/low at the last ingest date — only brand-new symbols
    advanced — stalling the whole Bybit forward gate, exactly the freeze the spot leg's own date-resume fixed.)"""
    n_symbols = 0
    perp_rows = 0
    spot_rows = 0
    oldest_day: int | None = None
    last_day = store.last_day_by_symbol("turnover") if resume else {}
    if resume and last_day and progress is not None:
        n_resuming = sum(1 for s in symbols if s in last_day)
        progress(f"resuming: {n_resuming} symbols from their last persisted perp day, "
                 f"{len(symbols) - n_resuming} full back-fill")

    def fetch_one(sym: str) -> tuple[dict[int, dict[str, float]], dict[int, dict[str, float]]]:
        # Both legs fetched together in the pool worker so each symbol's network work is one unit of concurrency.
        # DATE-RESUME from the symbol's last persisted perp day (shared primitive); fresh symbols back-fill from
        # the caller's floor (None → the adapter's backstop).
        floor = resume_floor_ms(last_day, sym, start_ms)
        perp = bybit.daily_ohlcv(sym, category="linear", start_ms=floor)
        spot = bybit.daily_ohlcv(sym, category="spot", start_ms=floor)
        return perp, spot

    def on_result(sym: str,
                  legs: tuple[dict[int, dict[str, float]], dict[int, dict[str, float]]]) -> None:
        nonlocal n_symbols, perp_rows, spot_rows, oldest_day
        perp, spot = legs
        if not perp and not spot:
            if progress is not None:
                progress(f"{sym}: 0 day-rows (no klines in window)")
            return
        if perp:
            store.write_features(sym, [
                (day * 86_400, {"close": bar["close"], "high": bar["high"],
                                "low": bar["low"], "turnover": bar["turnover"]})
                for day, bar in sorted(perp.items())])
            perp_rows += len(perp)
        if spot:
            store.write_features(sym, [
                (day * 86_400, {"spot_close": bar["close"], "spot_high": bar["high"],
                                "spot_low": bar["low"]})
                for day, bar in sorted(spot.items())])
            spot_rows += len(spot)
        n_symbols += 1
        days = list(perp) + list(spot)
        sym_oldest = min(days)
        oldest_day = sym_oldest if oldest_day is None else min(oldest_day, sym_oldest)
        if progress is not None:
            progress(f"{sym}: {len(perp)} perp + {len(spot)} spot day-rows (oldest day {sym_oldest})")

    fetch_symbols_concurrently(symbols, fetch_one, workers=workers, on_result=on_result)
    return {"symbols": n_symbols, "perp_rows": perp_rows, "spot_rows": spot_rows,
            "oldest_day": oldest_day}


def ingest_bybit_spot(
    store: Any,
    bybit: _BybitKlines,
    symbols: list[str],
    *,
    resume: bool = True,
    workers: int = DEFAULT_WORKERS,
    progress: Callable[[str], None] | None = None,
) -> dict[str, Any]:
    """Fetch + write ONLY the Bybit SPOT leg (spot_close/spot_high/spot_low, ts=epoch_day*86400) — the long
    leg of the delta-neutral carry — REUSING the same `BybitKlines` adapter (category="spot"); no new fetcher.

    Why a dedicated spot ingest (mirroring the retired legacy-venue `crypto_spot` asset's separate `ingest_spot`): the
    combined `ingest_bybit_klines` SYMBOL-SKIPS any symbol already carrying the perp `turnover` feature, which
    also freezes that symbol's spot leg at the last manual ingest — so the carry sleeve + the whole Bybit
    forward gate stall. This ingest instead RESUMES FROM THE LAST PERSISTED DATE: with `resume=True`, each
    symbol is fetched with `start_ms = its last persisted spot_close day` (via `store.last_day_by_symbol`),
    so the backward-paginating adapter pulls only NEW days and the idempotent upsert overwrites the boundary
    day in place — the spot leg therefore ADVANCES every nightly run. A symbol with no spot_close yet is
    back-filled from the adapter's 2020 floor (`start_ms=None`); `resume=False` re-fetches every symbol's full
    history (same final state — idempotent).

    Returns {"symbols": N, "spot_rows": S, "newest_day": D|None}: N = symbols that produced >=1 spot-row,
    S = total spot day-rows written, D = the newest epoch_day across all symbols (None if nothing ingested).

    Concurrency/memory: identical contract to `ingest_bybit_klines` — the per-symbol network FETCH runs across
    a bounded thread pool, each WRITE happens in the main thread as the fetch completes (serialized, no store
    thread-safety needed), results written as completed (peak memory ~one symbol). Skip-on-empty per symbol
    (survivorship-free: a name with no spot market returns nothing and never aborts the batch)."""
    last_day = store.last_day_by_symbol("spot_close") if resume else {}
    n_symbols = 0
    spot_rows = 0
    newest_day: int | None = None

    def fetch_one(sym: str) -> dict[int, dict[str, float]]:
        # Resume from the last persisted spot day (shared primitive); fresh symbols back-fill from the floor.
        return bybit.daily_ohlcv(sym, category="spot", start_ms=resume_floor_ms(last_day, sym, None))

    def on_result(sym: str, spot: dict[int, dict[str, float]]) -> None:
        nonlocal n_symbols, spot_rows, newest_day
        if not spot:
            if progress is not None:
                progress(f"{sym}: 0 spot day-rows (no klines in window)")
            return
        store.write_features(sym, [
            (day * 86_400, {"spot_close": bar["close"], "spot_high": bar["high"], "spot_low": bar["low"]})
            for day, bar in sorted(spot.items())])
        spot_rows += len(spot)
        n_symbols += 1
        sym_newest = max(spot)
        newest_day = sym_newest if newest_day is None else max(newest_day, sym_newest)
        if progress is not None:
            progress(f"{sym}: {len(spot)} spot day-rows (newest day {sym_newest})")

    fetch_symbols_concurrently(symbols, fetch_one, workers=workers, on_result=on_result)
    return {"symbols": n_symbols, "spot_rows": spot_rows, "newest_day": newest_day}
