"""Ingest the REAL intraday 1m `worst_basis` series into the Bybit-namespaced feature store
(table='bybit_features') — the squeeze-tail input the TRUSTWORTHY carry accounting OVERRIDES the daily-close
basis with.

Per symbol: `BybitKlines.intraday_worst_basis(symbol)` computes `{epoch_day: worst_basis}` (the deepest the
delta-neutral leg marked underwater intraday that day — see the adapter; same definition as the legacy
venue's `worst_basis_by_day`, capped at −1.0). The series is written as the `worst_basis` feature at
ts=epoch_day*86400 via the store's batched idempotent upsert. ONLY days with both perp+spot 1m bars produce
a value, so a perp-only coin lands no worst_basis and the carry falls back to its daily-close basis for
those days — exactly the legacy venue's fallback.

Concurrency (`workers`, default 12): the per-symbol fetch (network-bound 1m pagination of BOTH legs) runs
across the shared bounded pool (`fetch_symbols_concurrently`); each symbol's WRITE happens 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 (~one symbol's day-series resident); the adapter itself is per-day
bounded. DATE-RESUME (`resume=True`): each symbol resumes FROM ITS LAST PERSISTED `worst_basis` DAY (not
symbol-skipped), so the heavy weekly job advances every run instead of freezing at the last ingest. Idempotent
upsert ⇒ `resume=False` re-fetches to the identical state.
"""
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 intraday_worst_basis(self, symbol: str, *, start_ms: int | None = None,
                             interval: str = "1") -> dict[int, float]: ...


def ingest_bybit_worst_basis(
    store: Any,
    bybit: _BybitKlines,
    symbols: list[str],
    *,
    start_ms: int | None = None,
    interval: str = "1",
    resume: bool = True,
    workers: int = DEFAULT_WORKERS,
    progress: Callable[[str], None] | None = None,
) -> dict[str, Any]:
    """Compute + write the daily 1m `worst_basis` series for each symbol into `store` at ts=epoch_day*86400.

    Returns {"symbols": N, "days": D, "worst": W}: N = symbols that produced >=1 worst_basis day, D = total
    day-entries written, W = the single deepest (most negative) worst_basis observed (0.0 if none). A symbol
    with no overlap days contributes nothing and never aborts.

    Resumable: with `resume=True`, each symbol resumes FROM ITS LAST PERSISTED `worst_basis` DAY — the
    per-symbol last day is queried ONCE up front (`store.last_day_by_symbol('worst_basis')`) and used as the
    forward-window floor, so a symbol already carrying worst_basis is re-computed only from its last day
    forward (advancing it; idempotent upsert) and a fresh symbol is computed from `start_ms`. NOTHING is
    symbol-skipped, so worst_basis advances every run. The summary counts only this run's writes.
    """
    n_symbols = 0
    total_days = 0
    worst = 0.0
    last_day = store.last_day_by_symbol("worst_basis") if resume else {}
    if resume and last_day and progress is not None:
        progress(f"resuming from last persisted date: {len(last_day)} symbols already carry worst_basis")

    def fetch_one(sym: str) -> dict[int, float]:
        return bybit.intraday_worst_basis(
            sym, start_ms=resume_floor_ms(last_day, sym, start_ms), interval=interval)

    def on_result(sym: str, wb: dict[int, float]) -> None:
        nonlocal n_symbols, total_days, worst
        if not wb:
            if progress is not None:
                progress(f"{sym}: 0 worst_basis days (no overlapping 1m perp+spot)")
            return
        store.write_features(sym, [(day * 86_400, {"worst_basis": v}) for day, v in sorted(wb.items())])
        n_symbols += 1
        total_days += len(wb)
        worst = min(worst, min(wb.values()))
        if progress is not None:
            progress(f"{sym}: {len(wb)} worst_basis days (worst {min(wb.values()):.4f})")

    fetch_symbols_concurrently(symbols, fetch_one, workers=workers, on_result=on_result)
    return {"symbols": n_symbols, "days": total_days, "worst": worst}
