fxhnt/tests/integration/test_bybit_klines.py

"""BybitKlines adapter — perp + spot daily close from Bybit v5 market data, paginated BACKWARD via `end`.

ALL HTTP is injected/mocked — no network in these tests. The adapter takes a `fetch(url)->dict` callable and
drives daily-kline pagination off the `end` cursor exactly like Bybit (newest-first rows, walk `end` back).
Tests assert: backward pagination + dedupe, empty/floor stop, perp vs spot category in the URL, and the
startMs -> epoch_day keying (= close at row index 4).
"""
from __future__ import annotations

from fxhnt.adapters.data.bybit_klines import BYBIT_KLINE_FLOOR_MS, BybitKlines

_DAY_MS = 86_400_000


def _kline_row(start_ms: int, close: float) -> list[str]:
    # [startMs, open, high, low, close, volume, turnover]
    return [str(start_ms), "1", str(close + 1), str(close - 1), str(close), "10", "100"]


def _parse_qs(url: str, key: str) -> str | None:
    from urllib.parse import parse_qs, urlparse
    q = parse_qs(urlparse(url).query)
    return q[key][0] if key in q else None


class _PagedKline:
    """Mock of /v5/market/kline?interval=D: serves daily rows newest-first, honoring the `end` cursor like
    Bybit (returns up to `limit` rows with startMs <= end, descending). When the caller walks `end` back past
    the oldest row, it returns an empty list (stops the loop)."""

    def __init__(self, rows: list[list[str]], *, limit: int = 1000) -> None:
        # rows given oldest-first; store descending (newest first) like the API.
        self._rows = sorted(rows, key=lambda r: int(r[0]), reverse=True)
        self._limit = limit
        self.calls: list[dict[str, str | None]] = []

    def fetch(self, url: str) -> dict:
        end = _parse_qs(url, "end")
        category = _parse_qs(url, "category")
        self.calls.append({"end": end, "category": category})
        end_i = int(end) if end is not None else None
        cand = self._rows if end_i is None else [r for r in self._rows if int(r[0]) <= end_i]
        page = cand[: self._limit]
        return {"retCode": 0, "result": {"list": page}}


def test_daily_ohlcv_parses_high_low_turnover():
    # _kline_row sets high=close+1, low=close-1, turnover=100. close = row idx 4.
    rows = [_kline_row(100 * _DAY_MS, 100.0), _kline_row(101 * _DAY_MS, 101.0)]
    paged = _PagedKline(rows, limit=1000)
    bb = BybitKlines(fetch=paged.fetch, sleep=lambda _: None)
    out = bb.daily_ohlcv("BTCUSDT", category="linear", start_ms=0)
    assert out == {
        100: {"close": 100.0, "high": 101.0, "low": 99.0, "turnover": 100.0},
        101: {"close": 101.0, "high": 102.0, "low": 100.0, "turnover": 100.0},
    }


def test_daily_ohlcv_paginates_backward_unchanged():
    # The pagination/dedupe behaviour is identical to daily_close (which now projects daily_ohlcv).
    rows = [
        _kline_row(100 * _DAY_MS, 100.0),
        _kline_row(101 * _DAY_MS, 101.0),
        _kline_row(102 * _DAY_MS, 102.0),
    ]
    paged = _PagedKline(rows, limit=2)
    bb = BybitKlines(fetch=paged.fetch, sleep=lambda _: None)
    out = bb.daily_ohlcv("BTCUSDT", category="linear", start_ms=0)
    assert set(out) == {100, 101, 102}
    assert len(paged.calls) >= 2
    ends = [c["end"] for c in paged.calls if c["end"] is not None]
    assert ends == sorted(ends, reverse=True)


def test_daily_close_is_projection_of_daily_ohlcv():
    rows = [_kline_row(100 * _DAY_MS, 100.0), _kline_row(101 * _DAY_MS, 101.0)]
    bb = BybitKlines(fetch=_PagedKline(rows, limit=1000).fetch, sleep=lambda _: None)
    full = bb.daily_ohlcv("BTCUSDT", category="linear", start_ms=0)
    closes = bb.daily_close("BTCUSDT", category="linear", start_ms=0)
    assert closes == {d: bar["close"] for d, bar in full.items()}


def test_daily_close_paginates_backward_and_keys_by_epoch_day():
    # closes on 3 consecutive days, limit=2 forces paging; close = row index 4.
    rows = [
        _kline_row(100 * _DAY_MS, 100.0),
        _kline_row(101 * _DAY_MS, 101.0),
        _kline_row(102 * _DAY_MS, 102.0),
    ]
    paged = _PagedKline(rows, limit=2)
    bb = BybitKlines(fetch=paged.fetch, sleep=lambda _: None)
    out = bb.daily_close("BTCUSDT", category="linear", start_ms=0)
    assert out == {100: 100.0, 101: 101.0, 102: 102.0}
    # It actually paginated (>1 fetch) and walked `end` backward.
    assert len(paged.calls) >= 2
    ends = [c["end"] for c in paged.calls if c["end"] is not None]
    assert ends == sorted(ends, reverse=True)  # strictly walking backward


def test_daily_close_dedupes_across_pages():
    # A row repeated across a page boundary must not double-count (it's keyed by epoch_day anyway).
    rows = [_kline_row(50 * _DAY_MS, 50.0), _kline_row(51 * _DAY_MS, 51.0)]
    paged = _PagedKline(rows, limit=1)
    bb = BybitKlines(fetch=paged.fetch, sleep=lambda _: None)
    out = bb.daily_close("BTCUSDT", category="linear", start_ms=0)
    assert out == {50: 50.0, 51: 51.0}


def test_daily_close_empty_page_stops_loop():
    paged = _PagedKline([], limit=1000)
    bb = BybitKlines(fetch=paged.fetch, sleep=lambda _: None)
    assert bb.daily_close("NEWUSDT", category="linear") == {}
    assert len(paged.calls) == 1  # one fetch, empty -> stop


def test_daily_close_respects_start_ms_floor():
    rows = [_kline_row(10 * _DAY_MS, 10.0), _kline_row(20 * _DAY_MS, 20.0)]
    paged = _PagedKline(rows, limit=1000)
    bb = BybitKlines(fetch=paged.fetch, sleep=lambda _: None)
    out = bb.daily_close("BTCUSDT", category="linear", start_ms=15 * _DAY_MS)
    assert out == {20: 20.0}  # only the row at/after the floor survives


def test_perp_vs_spot_category_in_url():
    seen: list[str | None] = []

    def fetch(url: str) -> dict:
        seen.append(_parse_qs(url, "category"))
        return {"retCode": 0, "result": {"list": []}}

    bb = BybitKlines(fetch=fetch, sleep=lambda _: None)
    bb.daily_close("BTCUSDT", category="linear")
    bb.daily_close("BTCUSDT", category="spot")
    assert seen == ["linear", "spot"]


def test_daily_kline_uses_interval_d_and_symbol():
    seen: list[str] = []

    def fetch(url: str) -> dict:
        seen.append(url)
        return {"retCode": 0, "result": {"list": []}}

    bb = BybitKlines(fetch=fetch, sleep=lambda _: None)
    bb.daily_close("ETHUSDT", category="linear")
    assert "interval=D" in seen[0]
    assert "symbol=ETHUSDT" in seen[0]


def test_default_floor_is_2020_01():
    import datetime as dt
    assert int(dt.datetime(2020, 1, 1, tzinfo=dt.UTC).timestamp() * 1000) == BYBIT_KLINE_FLOOR_MS


def test_daily_close_retries_on_rate_limit_retcode_then_succeeds():
    # Under higher ingest concurrency Bybit may return a rate-limit retCode (10018). The adapter must back
    # off (injected no-op sleep — no real wait) and retry, then return the data.
    real = _PagedKline([_kline_row(40 * _DAY_MS, 40.0)], limit=1000)
    state = {"n": 0}

    def flaky(url: str) -> dict:
        state["n"] += 1
        if state["n"] <= 2:                          # first two calls are rate-limited
            return {"retCode": 10018, "retMsg": "ip rate limit", "result": {}}
        return real.fetch(url)

    waited: list[float] = []
    bb = BybitKlines(fetch=flaky, sleep=waited.append)
    out = bb.daily_close("BTCUSDT", category="linear", start_ms=0)
    assert out == {40: 40.0}                          # data eventually returned
    assert state["n"] >= 3                            # retried past the 2 rate-limited responses
    assert waited[:2] == [0.5, 1.0]                   # exponential backoff (no real sleep)