"""Ingest Bybit USDT-perp daily funding into a Bybit-namespaced feature store (table='bybit_features').

Per symbol: fetch {epoch_day: daily_funding} (the adapter already sums the day's 8h rates), write
feature='funding' at ts = epoch_day*86400 via the store's batched, idempotent upsert. Memory-bounded
(one symbol at a time) and resumable-friendly (the upsert overwrites in place, so re-running is safe and
cheap). Survivorship-free is the ADAPTER's job — a delisted symbol that returns rows is ingested like any
other; one that returns nothing simply contributes no rows and never aborts the batch.

DATE-RESUME (resume=True): each symbol is fetched FROM ITS LAST PERSISTED FUNDING DAY (not symbol-skipped),
so the nightly refresh advances funding every run instead of freezing it at the last ingest.
"""
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 _BybitFunding(Protocol):
    def funding_history(self, symbol: str, *, start_ms: int | None = None) -> dict[int, float]: ...


def ingest_bybit_funding(
    store: Any,
    bybit: _BybitFunding,
    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 funding for each symbol into `store` (feature='funding', ts=epoch_day*86400).

    Returns {"symbols": N, "day_rows": M, "oldest_day": D|None}: N = symbols that produced >=1 day-row,
    M = total day-rows written, D = the earliest epoch_day across all symbols (None if nothing ingested).

    Concurrency (`workers`, default 12): the per-symbol FETCH (which paginates the network-bound Bybit API)
    runs across a bounded thread pool; 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, so peak memory stays ~one symbol (preserving the OOM-fix property). The fetch is the
    bottleneck, so concurrent-fetch + serialized-write gives ~5–10× over the old sequential loop.

    Resumable (`resume=True`, default): each symbol resumes FROM ITS LAST PERSISTED 'funding' DAY — the
    per-symbol last day is queried ONCE up front (`store.last_day_by_symbol('funding')`) and used as the
    backward-pagination floor, so a symbol already carrying funding is re-fetched only from its last day
    forward (advancing it with new days; the idempotent upsert overwrites the boundary day in place) and a
    fresh symbol is back-filled from `start_ms` (or the adapter's backstop). NOTHING is symbol-skipped, so
    funding advances every run. `resume=False` ignores the persisted days and re-fetches everything from
    `start_ms` to the identical final state (idempotent)."""
    n_symbols = 0
    day_rows = 0
    oldest_day: int | None = None
    last_day = store.last_day_by_symbol("funding") 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 funding")

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

    def on_result(sym: str, daily: dict[int, float]) -> None:
        nonlocal n_symbols, day_rows, oldest_day
        if not daily:
            if progress is not None:
                progress(f"{sym}: 0 day-rows (no funding history in window)")
            return
        rows = [(day * 86_400, {"funding": val}) for day, val in sorted(daily.items())]
        store.write_features(sym, rows)            # batched idempotent upsert; serialized in the main thread
        n_symbols += 1
        day_rows += len(rows)
        sym_oldest = min(daily)
        oldest_day = sym_oldest if oldest_day is None else min(oldest_day, sym_oldest)
        if progress is not None:
            progress(f"{sym}: {len(rows)} day-rows (oldest day {sym_oldest})")

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