"""ingest_bybit_klines writes Bybit daily perp close ('close') + spot close ('spot_close') into a
bybit_features TimescaleFeatureStore, isolated from the default features table, idempotent on re-run.
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_klines_ingest import ingest_bybit_klines


def _bar(close: float) -> dict[str, float]:
    # Synthetic OHLCV bar: high=close+1, low=close-1, turnover=close*1000 (distinct per coin).
    return {"close": close, "high": close + 1.0, "low": close - 1.0, "turnover": close * 1000.0}


class _FakeBybit:
    """Stand-in for BybitKlines: returns canned {epoch_day: ohlcv-bar} per (symbol, category). The canned
    panels are passed as {epoch_day: close}; each close is expanded into a full OHLCV bar via `_bar`.

    Honors `start_ms` like the real backward-paginating adapter: when `start_ms` is not None only days on/after
    it (`day*86_400_000 >= start_ms`) are returned — so DATE-RESUME tests can assert a re-fetched symbol pulls
    ONLY its new days. `self.starts` records `(symbol, category, start_ms)` for each call."""

    def __init__(self, perp: dict[str, dict[int, float]], spot: dict[str, dict[int, float]]) -> None:
        self._perp = perp
        self._spot = spot
        self.calls: list[tuple[str, str]] = []
        self.starts: list[tuple[str, str, int | None]] = []

    def daily_ohlcv(self, symbol: str, *, category: str, start_ms=None) -> dict[int, dict[str, float]]:
        self.calls.append((symbol, category))
        self.starts.append((symbol, category, start_ms))
        panels = self._perp if category == "linear" else self._spot
        days = panels.get(symbol, {})
        if start_ms is not None:
            days = {d: px for d, px in days.items() if d * 86_400_000 >= start_ms}
        return {day: _bar(px) for day, px in days.items()}


def test_ingest_writes_close_and_spot_close_into_bybit_features():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    bybit = _FakeBybit(
        perp={"BTCUSDT": {19000: 100.0, 19001: 101.0}, "ETHUSDT": {19000: 5.0}},
        spot={"BTCUSDT": {19000: 99.5, 19001: 100.5}, "ETHUSDT": {19000: 4.9}, "DEADUSDT": {}},
    )
    summary = ingest_bybit_klines(store, bybit, ["BTCUSDT", "ETHUSDT", "DEADUSDT"])

    assert store.crypto_close_panel() == {
        "BTCUSDT": {19000: 100.0, 19001: 101.0},
        "ETHUSDT": {19000: 5.0},
    }
    assert store.crypto_spot_panel() == {
        "BTCUSDT": {19000: 99.5, 19001: 100.5},
        "ETHUSDT": {19000: 4.9},
    }
    # read_panel(...) is the canonical reader too.
    assert store.read_panel(["BTCUSDT"], "close") == {"BTCUSDT": {19000: 100.0, 19001: 101.0}}
    assert store.read_panel(["BTCUSDT"], "spot_close") == {"BTCUSDT": {19000: 99.5, 19001: 100.5}}

    assert summary["symbols"] == 2          # symbols that produced >=1 perp- OR spot-row
    assert summary["perp_rows"] == 3        # 2 (BTC) + 1 (ETH)
    assert summary["spot_rows"] == 3        # 2 (BTC) + 1 (ETH); DEAD contributes nothing
    assert summary["oldest_day"] == 19000
    store.close()


def test_ingest_is_idempotent():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    bybit = _FakeBybit(
        perp={"BTCUSDT": {19000: 100.0, 19001: 101.0}},
        spot={"BTCUSDT": {19000: 99.5}},
    )
    ingest_bybit_klines(store, bybit, ["BTCUSDT"])
    first_close = store.crypto_close_panel()
    first_spot = store.crypto_spot_panel()
    ingest_bybit_klines(store, bybit, ["BTCUSDT"])  # re-run
    assert store.crypto_close_panel() == first_close      # upsert overwrites in place, no duplicates
    assert store.crypto_spot_panel() == first_spot
    store.close()


def test_ingest_isolated_from_default_features_table():
    import os
    import tempfile
    with tempfile.TemporaryDirectory() as td:
        fdsn = f"sqlite:///{os.path.join(td, 'shared.db')}"
        default = TimescaleFeatureStore(fdsn)
        bybit_store = TimescaleFeatureStore(fdsn, table="bybit_features")
        default.write_features("BTCUSDT", [(19000 * 86400, {"close": 42.0})])
        ingest_bybit_klines(
            bybit_store,
            _FakeBybit(perp={"BTCUSDT": {19000: 100.0}}, spot={"BTCUSDT": {19000: 99.5}}),
            ["BTCUSDT"],
        )
        # The Bybit klines ingest did NOT touch the default features table.
        assert default.crypto_close_panel() == {"BTCUSDT": {19000: 42.0}}
        assert bybit_store.crypto_close_panel() == {"BTCUSDT": {19000: 100.0}}
        assert bybit_store.crypto_spot_panel() == {"BTCUSDT": {19000: 99.5}}
        default.close()
        bybit_store.close()


def test_ingest_skips_symbol_with_no_klines():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    bybit = _FakeBybit(perp={"DEADUSDT": {}}, spot={"DEADUSDT": {}})
    summary = ingest_bybit_klines(store, bybit, ["DEADUSDT"])
    assert summary == {"symbols": 0, "perp_rows": 0, "spot_rows": 0, "oldest_day": None}
    assert store.crypto_close_panel() == {}
    store.close()


def test_ingest_progress_callback_invoked():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    bybit = _FakeBybit(
        perp={"BTCUSDT": {19000: 100.0}, "ETHUSDT": {19000: 5.0}},
        spot={"BTCUSDT": {19000: 99.5}, "ETHUSDT": {19000: 4.9}},
    )
    seen: list[str] = []
    ingest_bybit_klines(store, bybit, ["BTCUSDT", "ETHUSDT"], progress=lambda m: seen.append(m))
    assert any("BTCUSDT" in m for m in seen) and any("ETHUSDT" in m for m in seen)
    store.close()


def test_resume_datefetches_done_symbols_from_last_day_and_completes_the_rest():
    """DATE-RESUME contract (replaces the old symbol-skip): ingest {A,B}, then {A,B,C,D} with resume=True →
    EVERY symbol is fetched, but the already-done A,B RESUME FROM THEIR LAST PERSISTED PERP DAY (so the
    backward-paginating adapter pulls only new days) while the fresh C,D back-fill from the floor (start_ms
    None). The idempotent upsert keeps A,B's boundary day in place; the final store holds all four. This is the
    freeze fix — under the old symbol-skip A,B were never re-fetched so their close/turnover froze."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    perp = {"AUSDT": {19000: 1.0}, "BUSDT": {19000: 2.0}, "CUSDT": {19000: 3.0}, "DUSDT": {19000: 4.0}}
    spot = {"AUSDT": {19000: 0.9}, "BUSDT": {19000: 1.9}, "CUSDT": {19000: 2.9}, "DUSDT": {19000: 3.9}}

    first = _FakeBybit(perp=perp, spot=spot)
    ingest_bybit_klines(store, first, ["AUSDT", "BUSDT"])
    assert {c[0] for c in first.calls} == {"AUSDT", "BUSDT"}

    second = _FakeBybit(perp=perp, spot=spot)
    seen: list[str] = []
    summary = ingest_bybit_klines(
        store, second, ["AUSDT", "BUSDT", "CUSDT", "DUSDT"], resume=True,
        progress=lambda m: seen.append(m))
    fetched = {c[0] for c in second.calls}
    assert fetched == {"AUSDT", "BUSDT", "CUSDT", "DUSDT"}, "date-resume must re-fetch EVERY symbol"
    # done A,B resume from their last persisted perp day (start_ms set); fresh C,D back-fill (start_ms None).
    perp_start = {sym: sm for sym, cat, sm in second.starts if cat == "linear"}
    assert perp_start["AUSDT"] == 19000 * 86_400_000 and perp_start["BUSDT"] == 19000 * 86_400_000
    assert perp_start["CUSDT"] is None and perp_start["DUSDT"] is None
    assert any("resuming: 2 symbols from their last persisted perp day, 2 full back-fill" in m for m in seen)

    assert store.crypto_close_panel() == {
        "AUSDT": {19000: 1.0}, "BUSDT": {19000: 2.0}, "CUSDT": {19000: 3.0}, "DUSDT": {19000: 4.0}}
    assert store.crypto_spot_panel() == {
        "AUSDT": {19000: 0.9}, "BUSDT": {19000: 1.9}, "CUSDT": {19000: 2.9}, "DUSDT": {19000: 3.9}}
    store.close()


def test_resume_datefetches_symbol_with_perp_leg_having_turnover_and_advances():
    """A symbol that already carries `turnover` is DATE-RESUMED (not skipped): it is re-fetched from its last
    persisted perp day and ADVANCES to any newer days. This is the exact freeze the perp date-resume migration
    fixes — the old symbol-skip left close/turnover frozen at the last ingest date."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    # Landed through day 19000 (carries turnover).
    ingest_bybit_klines(store, _FakeBybit(perp={"AUSDT": {19000: 1.0}}, spot={"AUSDT": {19000: 0.9}}),
                        ["AUSDT"])

    # A newer day 19001 is now available; resume must fetch it (re-fetched from last day 19000, advancing).
    second = _FakeBybit(perp={"AUSDT": {19000: 1.0, 19001: 1.5}}, spot={"AUSDT": {19000: 0.9, 19001: 0.95}})
    ingest_bybit_klines(store, second, ["AUSDT"], resume=True)
    assert {c[0] for c in second.calls} == {"AUSDT"}, "symbol with turnover must be RE-FETCHED (date-resume)"
    perp_start = next(sm for sym, cat, sm in second.starts if cat == "linear")
    assert perp_start == 19000 * 86_400_000, "must resume from the last persisted perp day, not from scratch"
    assert store.crypto_close_panel() == {"AUSDT": {19000: 1.0, 19001: 1.5}}, "perp close must ADVANCE to 19001"
    store.close()


def test_ingest_writes_high_low_turnover_features():
    """The 7 features land: perp close/high/low/turnover + spot spot_close/spot_high/spot_low."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    bybit = _FakeBybit(perp={"BTCUSDT": {19000: 100.0}}, spot={"BTCUSDT": {19000: 99.5}})
    ingest_bybit_klines(store, bybit, ["BTCUSDT"])
    # perp leg: close=100, high=101, low=99, turnover=100*1000
    assert store.read_panel(["BTCUSDT"], "high") == {"BTCUSDT": {19000: 101.0}}
    assert store.read_panel(["BTCUSDT"], "low") == {"BTCUSDT": {19000: 99.0}}
    assert store.read_panel(["BTCUSDT"], "turnover") == {"BTCUSDT": {19000: 100_000.0}}
    # spot leg: spot_close=99.5, spot_high=100.5, spot_low=98.5
    assert store.read_panel(["BTCUSDT"], "spot_close") == {"BTCUSDT": {19000: 99.5}}
    assert store.read_panel(["BTCUSDT"], "spot_high") == {"BTCUSDT": {19000: 100.5}}
    assert store.read_panel(["BTCUSDT"], "spot_low") == {"BTCUSDT": {19000: 98.5}}
    store.close()


def test_resume_backfills_legacy_store_missing_turnover():
    """A store written by the OLD ingest (close/spot_close only, NO turnover) is correctly BACKFILLED by a
    resumed re-run: the legacy symbol lacks 'turnover', so it is NOT in the done-set → re-fetched, writing
    all 7 features. No --no-resume required."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    # Simulate the legacy state: close + spot_close, but no high/low/turnover.
    store.write_features("AUSDT", [(19000 * 86_400, {"close": 1.0, "spot_close": 0.9})])
    assert store.symbols_with_feature("turnover") == set()   # legacy → no turnover

    second = _FakeBybit(perp={"AUSDT": {19000: 1.0}}, spot={"AUSDT": {19000: 0.9}})
    ingest_bybit_klines(store, second, ["AUSDT"], resume=True)
    assert {c[0] for c in second.calls} == {"AUSDT"}, "legacy symbol must be re-fetched to backfill"
    assert store.read_panel(["AUSDT"], "turnover") == {"AUSDT": {19000: 1000.0}}
    store.close()


def test_no_resume_refetches_everything():
    """resume=False re-fetches every symbol, landing the identical final state (idempotent)."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    perp = {"AUSDT": {19000: 1.0}}
    spot = {"AUSDT": {19000: 0.9}}
    ingest_bybit_klines(store, _FakeBybit(perp=perp, spot=spot), ["AUSDT"])

    second = _FakeBybit(perp=perp, spot=spot)
    ingest_bybit_klines(store, second, ["AUSDT"], resume=False)
    assert {c[0] for c in second.calls} == {"AUSDT"}
    assert store.crypto_close_panel() == {"AUSDT": {19000: 1.0}}
    store.close()


def test_resume_equals_single_full_run():
    """Partial run + resume run == one single full run (complete on the union, idempotent on the overlap)."""
    perp = {f"{c}USDT": {19000: float(i)} for i, c in enumerate("ABCDE", start=1)}
    spot = {f"{c}USDT": {19000: float(i) - 0.1} for i, c in enumerate("ABCDE", start=1)}
    syms = list(perp)

    partial = TimescaleFeatureStore("sqlite://", table="bybit_features")
    ingest_bybit_klines(partial, _FakeBybit(perp=perp, spot=spot), syms[:2])
    ingest_bybit_klines(partial, _FakeBybit(perp=perp, spot=spot), syms, resume=True)
    resumed = (partial.crypto_close_panel(), partial.crypto_spot_panel())
    partial.close()

    full = TimescaleFeatureStore("sqlite://", table="bybit_features")
    ingest_bybit_klines(full, _FakeBybit(perp=perp, spot=spot), syms)
    full_state = (full.crypto_close_panel(), full.crypto_spot_panel())
    full.close()

    assert resumed == full_state


def test_concurrent_ingest_writes_all_symbols_exactly_once():
    """With workers>1 every symbol's BOTH legs are still fetched + written exactly once, matching sequential."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    perp = {f"S{i}USDT": {19000: float(i)} for i in range(30)}
    spot = {f"S{i}USDT": {19000: float(i) - 0.1} for i in range(30)}
    bybit = _FakeBybit(perp=perp, spot=spot)
    summary = ingest_bybit_klines(store, bybit, list(perp), workers=8)
    assert summary["symbols"] == 30 and summary["perp_rows"] == 30 and summary["spot_rows"] == 30
    assert store.crypto_close_panel() == perp
    assert store.crypto_spot_panel() == spot
    # both legs fetched for every symbol, once each
    assert sorted(bybit.calls) == sorted([(s, "linear") for s in perp] + [(s, "spot") for s in perp])
    store.close()


def test_concurrent_resume_datefetches_all_last_day_map_computed_before_pool():
    """Resume under concurrency: the last-day map is queried once up front, then workers>1 DATE-RESUME every
    symbol — the already-done first half from their last persisted day (start_ms set), the not-yet-done half
    from the floor (start_ms None). Idempotent upsert → final state complete."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    perp = {f"S{i}USDT": {19000: float(i)} for i in range(20)}
    spot = {f"S{i}USDT": {19000: float(i) - 0.1} for i in range(20)}
    ingest_bybit_klines(store, _FakeBybit(perp=perp, spot=spot), list(perp)[:10], workers=8)

    second = _FakeBybit(perp=perp, spot=spot)
    ingest_bybit_klines(store, second, list(perp), resume=True, workers=8)
    assert {c[0] for c in second.calls} == set(perp)  # date-resume re-fetches EVERY symbol
    perp_start = {sym: sm for sym, cat, sm in second.starts if cat == "linear"}
    done_half, fresh_half = set(list(perp)[:10]), set(list(perp)[10:])
    assert all(perp_start[s] == 19000 * 86_400_000 for s in done_half)  # done → resume from last day
    assert all(perp_start[s] is None for s in fresh_half)               # fresh → full back-fill
    assert store.crypto_close_panel() == perp
    store.close()
