"""ingest_deribit_funding — writes the Deribit perp daily funding into bybit_features as feature
'deribit_funding' for the OVERLAP symbols (BTC-PERPETUAL -> BTCUSDT, ETH-PERPETUAL -> ETHUSDT, ...).

NO network — a fake DeribitFunding serves canned {epoch_day: funding}. Asserts: the instrument->symbol map,
that ONLY 'deribit_funding' is written (never touching 'funding'/'close'), the summary counts, and the
skip-on-empty behaviour.
"""
from __future__ import annotations

import math
import threading

import pytest

from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
from fxhnt.application.deribit_funding_ingest import (
    DEFAULT_DERIBIT_INSTRUMENTS,
    ingest_deribit_funding,
    instrument_to_symbol,
)

_DAY = 86_400


def test_default_instruments_are_valid_deribit_names():
    """ROOT-CAUSE REGRESSION for the prod HTTP-400: a bare 'SOL-PERPETUAL' does NOT exist on Deribit (only
    BTC/ETH are inverse '<BASE>-PERPETUAL'; alts are USDC-margined '<BASE>_USDC-PERPETUAL'). Requesting the
    bad name 400'd, which the asset caught as a (mis-classified) transient URLError → the whole deribit_funding
    panel froze and capped xvenue_carry. The default set must use only valid names, each mapping to its Bybit
    symbol."""
    assert "SOL-PERPETUAL" not in DEFAULT_DERIBIT_INSTRUMENTS    # the invalid name that 400'd
    assert "SOL_USDC-PERPETUAL" in DEFAULT_DERIBIT_INSTRUMENTS   # the correct USDC-margined SOL perp
    # BTC/ETH stay inverse perps; every ALT must be the _USDC-PERPETUAL form (never bare <ALT>-PERPETUAL).
    for inst in DEFAULT_DERIBIT_INSTRUMENTS:
        assert inst.endswith("-PERPETUAL"), inst
        base = inst.split("-")[0].split("_")[0]
        if base not in ("BTC", "ETH"):
            assert inst == f"{base}_USDC-PERPETUAL", f"alt perp must be USDC-margined, got {inst}"
    assert instrument_to_symbol("SOL_USDC-PERPETUAL") == "SOLUSDT"
    assert {instrument_to_symbol(i) for i in DEFAULT_DERIBIT_INSTRUMENTS} == {"BTCUSDT", "ETHUSDT", "SOLUSDT"}


class _FakeDeribit:
    def __init__(self, by_instrument: dict[str, dict[int, float]]) -> None:
        self._by = by_instrument
        self.calls: list[tuple[str, int | None]] = []   # (instrument, start_ms) — to assert date-resume

    def funding_history(self, instrument: str, *, start_ms: int | None = None) -> dict[int, float]:
        self.calls.append((instrument, start_ms))
        return dict(self._by.get(instrument, {}))


def test_ingest_writes_deribit_funding_for_mapped_symbols():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    der = _FakeDeribit({
        "BTC-PERPETUAL": {100: 0.0002, 101: 0.0003},
        "ETH-PERPETUAL": {100: 0.0001},
    })
    summary = ingest_deribit_funding(store, der, ["BTC-PERPETUAL", "ETH-PERPETUAL"])

    panel = store.read_panel(["BTCUSDT", "ETHUSDT"], "deribit_funding")
    assert math.isclose(panel["BTCUSDT"][100], 0.0002, abs_tol=1e-15)
    assert math.isclose(panel["BTCUSDT"][101], 0.0003, abs_tol=1e-15)
    assert math.isclose(panel["ETHUSDT"][100], 0.0001, abs_tol=1e-15)
    assert summary["symbols"] == 2
    assert summary["day_rows"] == 3
    assert summary["oldest_day"] == 100
    store.close()


def test_ingest_writes_only_deribit_funding_feature_not_funding_or_close():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    # Pre-seed a Bybit 'funding' + 'close' for BTCUSDT — the ingest must NOT clobber them.
    store.write_features("BTCUSDT", [(100 * _DAY, {"funding": 0.0005, "close": 50000.0})])
    der = _FakeDeribit({"BTC-PERPETUAL": {100: 0.0002}})
    ingest_deribit_funding(store, der, ["BTC-PERPETUAL"])

    assert store.read_panel(["BTCUSDT"], "funding")["BTCUSDT"][100] == 0.0005   # untouched
    assert store.read_panel(["BTCUSDT"], "close")["BTCUSDT"][100] == 50000.0    # untouched
    assert store.read_panel(["BTCUSDT"], "deribit_funding")["BTCUSDT"][100] == 0.0002
    store.close()


def test_ingest_skips_instrument_with_no_history():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    der = _FakeDeribit({"BTC-PERPETUAL": {100: 0.0002}})   # SOL has none
    summary = ingest_deribit_funding(store, der, ["BTC-PERPETUAL", "SOL-PERPETUAL"])
    assert summary["symbols"] == 1
    assert store.read_panel(["SOLUSDT"], "deribit_funding") == {"SOLUSDT": {}}
    store.close()


def test_ingest_empty_universe_is_noop():
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    der = _FakeDeribit({})
    summary = ingest_deribit_funding(store, der, [])
    assert summary == {"symbols": 0, "day_rows": 0, "oldest_day": None}
    store.close()


def test_resume_advances_from_last_persisted_date():
    """DATE-RESUME: an instrument already carrying OLD deribit_funding is re-fetched FROM ITS LAST PERSISTED
    DAY — start_ms keyed on `last_day_by_symbol('deribit_funding')` (by the Bybit symbol) — so new days land
    (advance), while a fresh instrument back-fills from the default floor. This makes the nightly incremental."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    # Seed BTC (-> BTCUSDT) deribit_funding at day 100 only.
    store.write_features("BTCUSDT", [(100 * _DAY, {"deribit_funding": 0.0002})])

    der = _FakeDeribit({
        "BTC-PERPETUAL": {100: 0.0002, 101: 0.0004},   # serves an EXTRA new day 101
        "ETH-PERPETUAL": {100: 0.0001},                 # fresh (no prior deribit_funding)
    })
    ingest_deribit_funding(store, der, ["BTC-PERPETUAL", "ETH-PERPETUAL"], resume=True)

    fetched = dict(der.calls)
    assert fetched["BTC-PERPETUAL"] == 100 * 86_400_000   # resumed from its last persisted day
    assert fetched["ETH-PERPETUAL"] is None               # fresh -> default floor (None)

    panel = store.read_panel(["BTCUSDT", "ETHUSDT"], "deribit_funding")
    assert panel["BTCUSDT"] == {100: 0.0002, 101: 0.0004}  # advanced to 101
    assert panel["ETHUSDT"] == {100: 0.0001}
    store.close()


def test_no_resume_ignores_persisted_days():
    """resume=False fetches every instrument from the default floor (start_ms=None here), ignoring the
    per-symbol last persisted day — the idempotent re-fetch path."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    store.write_features("BTCUSDT", [(100 * _DAY, {"deribit_funding": 0.0002})])
    der = _FakeDeribit({"BTC-PERPETUAL": {100: 0.0002, 101: 0.0004}})
    ingest_deribit_funding(store, der, ["BTC-PERPETUAL"], resume=False)
    assert dict(der.calls)["BTC-PERPETUAL"] is None   # no date-resume floor applied
    store.close()


class _OrderRecordingDeribit:
    """Returns distinct day-sets per instrument, records concurrent in-flight fetches + completion order.
    Each fetch sleeps briefly so, under workers>1, several are genuinely in flight at once."""

    def __init__(self, by_instrument: dict[str, dict[int, float]], *, delay: float = 0.0) -> None:
        self._by = by_instrument
        self._delay = delay
        self._lock = threading.Lock()
        self.max_in_flight = 0
        self._in_flight = 0
        self.completion_order: list[str] = []

    def funding_history(self, instrument: str, *, start_ms: int | None = None) -> dict[int, float]:
        import time as _t
        with self._lock:
            self._in_flight += 1
            self.max_in_flight = max(self.max_in_flight, self._in_flight)
        if self._delay:
            _t.sleep(self._delay)
        with self._lock:
            self._in_flight -= 1
            self.completion_order.append(instrument)
        return dict(self._by.get(instrument, {}))


def test_concurrent_fetch_merges_all_instruments_regardless_of_order():
    """Distinct day-sets per instrument; under workers>1 fetches run concurrently and out of submission
    order, yet every instrument's rows land in the store and the summary is correct."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    by = {f"C{i}-PERPETUAL": {19000 + i: float(i) / 1000} for i in range(30)}
    der = _OrderRecordingDeribit(by, delay=0.005)
    instruments = list(by)
    summary = ingest_deribit_funding(store, der, instruments, workers=8)

    assert der.max_in_flight > 1, "fetches were not actually concurrent"
    assert summary["symbols"] == 30
    assert summary["day_rows"] == 30
    assert summary["oldest_day"] == 19000
    syms = [f"C{i}USDT" for i in range(30)]
    panel = store.read_panel(syms, "deribit_funding")
    for i in range(30):
        assert math.isclose(panel[f"C{i}USDT"][19000 + i], float(i) / 1000, abs_tol=1e-15)
    store.close()


def test_deterministic_regardless_of_thread_scheduling():
    """Same inputs -> identical summary AND identical store writes across two runs (different completion
    orders), independent of thread scheduling."""
    by = {f"C{i}-PERPETUAL": {19000 + (i % 7): float(i)} for i in range(25)}
    instruments = list(by)

    def run_once() -> tuple[dict, dict]:
        store = TimescaleFeatureStore("sqlite://", table="bybit_features")
        der = _OrderRecordingDeribit(by, delay=0.003)
        summary = ingest_deribit_funding(store, der, instruments, workers=8)
        syms = [f"C{i}USDT" for i in range(25)]
        panel = store.read_panel(syms, "deribit_funding")
        store.close()
        return summary, panel

    s1, p1 = run_once()
    s2, p2 = run_once()
    assert s1 == s2
    assert p1 == p2


def test_concurrent_skip_on_empty_preserved():
    """An instrument returning {} contributes nothing under concurrency; the rest still ingest."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    by = {"BTC-PERPETUAL": {100: 0.0002}, "ETH-PERPETUAL": {100: 0.0001}}  # SOL absent -> {}
    der = _FakeDeribit(by)
    summary = ingest_deribit_funding(
        store, der, ["BTC-PERPETUAL", "SOL-PERPETUAL", "ETH-PERPETUAL"], workers=4)
    assert summary["symbols"] == 2
    assert summary["day_rows"] == 2
    assert store.read_panel(["SOLUSDT"], "deribit_funding") == {"SOLUSDT": {}}
    assert store.read_panel(["BTCUSDT"], "deribit_funding")["BTCUSDT"][100] == 0.0002
    assert store.read_panel(["ETHUSDT"], "deribit_funding")["ETHUSDT"][100] == 0.0001
    store.close()


class _ErroringDeribit:
    def __init__(self, by_instrument: dict[str, dict[int, float]], *, bad: str) -> None:
        self._by = by_instrument
        self._bad = bad
        self.calls: list[str] = []

    def funding_history(self, instrument: str, *, start_ms: int | None = None) -> dict[int, float]:
        self.calls.append(instrument)
        if instrument == self._bad:
            raise ValueError(f"400 Bad Request: {instrument} does not exist")
        return dict(self._by.get(instrument, {}))


def test_one_failing_instrument_propagates_like_bybit():
    """Matches the Bybit concurrency contract: the first fetch exception surfaces to the caller (the bad
    SOL-PERPETUAL inverse that 400s is NOT silently swallowed). fetch_symbols_concurrently re-raises via
    future.result(), so the deribit ingest behaves identically to the bybit ingests."""
    store = TimescaleFeatureStore("sqlite://", table="bybit_features")
    by = {"BTC-PERPETUAL": {100: 0.0002}, "ETH-PERPETUAL": {100: 0.0001}}
    der = _ErroringDeribit(by, bad="SOL-PERPETUAL")
    with pytest.raises(ValueError, match="does not exist"):
        ingest_deribit_funding(
            store, der, ["BTC-PERPETUAL", "SOL-PERPETUAL", "ETH-PERPETUAL"], workers=4)
    store.close()
