"""ingest_bybit_oi writes the Bybit OPEN INTEREST (feature='open_interest') into a bybit_features
TimescaleFeatureStore — concurrent, resumable, idempotent, memory-bounded. The Bybit adapter is a fake (no
HTTP)."""
from __future__ import annotations

from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
from fxhnt.application.bybit_oi_ingest import ingest_bybit_oi


class _FakeBybit:
    """Stand-in for BybitOpenInterest: returns canned {epoch_day: open_interest} per symbol."""

    def __init__(self, panels: dict[str, dict[int, float]]) -> None:
        self._panels = panels
        self.calls: list[tuple[str, int | None]] = []

    def oi_history(self, symbol: str, *, start_ms=None) -> dict[int, float]:
        self.calls.append((symbol, start_ms))
        return self._panels.get(symbol, {})


def test_ingest_writes_open_interest_into_bybit_features():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    bybit = _FakeBybit({
        "BTCUSDT": {19000: 1_000_000.0, 19001: 950_000.0},
        "ETHUSDT": {19000: 500_000.0},
        "DEADUSDT": {},   # no OI rows in window -> contributes nothing
    })
    summary = ingest_bybit_oi(store, bybit, ["BTCUSDT", "ETHUSDT", "DEADUSDT"])

    assert store.read_panel(["BTCUSDT"], "open_interest") == {
        "BTCUSDT": {19000: 1_000_000.0, 19001: 950_000.0}}
    assert store.read_panel(["ETHUSDT"], "open_interest") == {"ETHUSDT": {19000: 500_000.0}}
    assert summary["symbols"] == 2          # only symbols that produced >=1 day-row
    assert summary["day_rows"] == 3
    assert summary["oldest_day"] == 19000
    store.close()


def test_ingest_is_idempotent():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    bybit = _FakeBybit({"BTCUSDT": {19000: 1_000_000.0, 19001: 950_000.0}})
    ingest_bybit_oi(store, bybit, ["BTCUSDT"])
    first = store.read_panel(["BTCUSDT"], "open_interest")
    ingest_bybit_oi(store, bybit, ["BTCUSDT"])   # re-run
    assert store.read_panel(["BTCUSDT"], "open_interest") == first
    store.close()


def test_ingest_only_writes_open_interest_feature_not_funding_or_close():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    ingest_bybit_oi(store, _FakeBybit({"BTCUSDT": {19000: 1_000_000.0}}), ["BTCUSDT"])
    assert store.read_panel(["BTCUSDT"], "open_interest") == {"BTCUSDT": {19000: 1_000_000.0}}
    assert store.read_panel(["BTCUSDT"], "long_ratio") == {"BTCUSDT": {}}
    assert store.crypto_funding_panel() == {}
    assert store.crypto_close_panel() == {}
    store.close()


def test_resume_advances_done_symbols_from_last_date_and_completes_the_rest():
    """DATE-RESUME (replaces symbol-skip): A,B already carry OLD open_interest -> re-fetched FROM THEIR LAST
    PERSISTED DAY so NEW days land (advance); C,D back-filled from the default floor. Nothing is skipped."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    ingest_bybit_oi(store, _FakeBybit({"AUSDT": {19000: 1.0}, "BUSDT": {19000: 1.0}}), ["AUSDT", "BUSDT"])

    panels = {
        "AUSDT": {19000: 1.0, 19001: 1.5}, "BUSDT": {19000: 1.0, 19001: 1.6},
        "CUSDT": {19000: 1.0}, "DUSDT": {19000: 1.0},
    }
    second = _FakeBybit(panels)
    seen: list[str] = []
    ingest_bybit_oi(
        store, second, ["AUSDT", "BUSDT", "CUSDT", "DUSDT"], resume=True,
        progress=lambda m: seen.append(m))
    fetched = dict(second.calls)
    assert set(fetched) == {"AUSDT", "BUSDT", "CUSDT", "DUSDT"}
    assert fetched["AUSDT"] == 19000 * 86_400_000 and fetched["BUSDT"] == 19000 * 86_400_000
    assert fetched["CUSDT"] is None and fetched["DUSDT"] is None
    assert any("resuming from last persisted date" in m for m in seen)
    assert store.read_panel(["AUSDT", "BUSDT", "CUSDT", "DUSDT"], "open_interest") == panels
    store.close()


def test_no_resume_refetches_everything():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    panels = {"AUSDT": {19000: 1.0}, "BUSDT": {19000: 2.0}}
    ingest_bybit_oi(store, _FakeBybit(panels), ["AUSDT", "BUSDT"])
    second = _FakeBybit(panels)
    ingest_bybit_oi(store, second, ["AUSDT", "BUSDT"], resume=False)
    assert {c[0] for c in second.calls} == {"AUSDT", "BUSDT"}
    store.close()


def test_concurrent_ingest_writes_all_symbols_exactly_once():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    panels = {f"S{i}USDT": {19000 + i: 1.0 + i} for i in range(40)}
    bybit = _FakeBybit(panels)
    summary = ingest_bybit_oi(store, bybit, list(panels), workers=8)
    assert summary["symbols"] == 40 and summary["day_rows"] == 40
    assert store.read_panel(list(panels), "open_interest") == {s: d for s, d in panels.items()}
    assert sorted(c[0] for c in bybit.calls) == sorted(panels)   # each symbol fetched once
    store.close()


def test_concurrent_resume_refetches_all_from_last_date():
    """Under concurrency, date-resume re-fetches ALL symbols (the done half from their last persisted day) —
    no symbol is skipped (the per-symbol last-day map is queried once up front)."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    panels = {f"S{i}USDT": {19000: 1.0} for i in range(20)}
    done_half = list(panels)[:10]
    ingest_bybit_oi(store, _FakeBybit(panels), done_half, workers=8)

    second = _FakeBybit(panels)
    ingest_bybit_oi(store, second, list(panels), resume=True, workers=8)
    fetched = dict(second.calls)
    assert set(fetched) == set(panels)
    assert all(fetched[s] == 19000 * 86_400_000 for s in done_half)
    assert all(fetched[s] is None for s in list(panels)[10:])
    store.close()
