crypto_monitor/tests/test_hub_ohlcv_lib.py

"""hub_ohlcv_lib：分页拉取（Gate 等单次不足 chunk 时仍继续）。"""
from __future__ import annotations

import unittest

from hub_ohlcv_lib import (
    aggregate_ohlcv_bars,
    bars_spacing_matches_timeframe,
    fetch_ohlcv_for_hub,
    normalize_price_tick,
)


class _FakeExchange:
    def __init__(self, pages, *, timeframes=None):
        self.pages = list(pages)
        self.calls = []
        self.markets = {}
        self.timeframes = timeframes if timeframes is not None else {}

    def fetch_ohlcv(self, symbol, timeframe=None, since=None, limit=None):
        self.calls.append(
            {"symbol": symbol, "since": since, "limit": limit, "timeframe": timeframe}
        )
        if not self.pages:
            return []
        page = self.pages.pop(0)
        if since is None:
            return page
        return [b for b in page if b[0] >= since]


class TestHubOhlcvLib(unittest.TestCase):
    def test_normalize_price_tick_snaps_powers_of_ten(self):
        self.assertAlmostEqual(normalize_price_tick(0.00001), 0.00001)
        self.assertAlmostEqual(normalize_price_tick(0.001), 0.001)
        self.assertIsNone(normalize_price_tick(0))

    def test_price_tick_from_decimal_precision(self):
        class _Ex:
            markets = {"BTC/USDT:USDT": {"precision": {"price": 2}, "info": {}, "limits": {}}}

            def load_markets(self):
                return self.markets

            def market(self, sym):
                return self.markets[sym]

            def price_to_precision(self, sym, price):
                return "12345.67"

        tick = __import__("hub_ohlcv_lib", fromlist=["price_tick_from_market"]).price_tick_from_market(
            _Ex(), "BTC/USDT:USDT"
        )
        self.assertAlmostEqual(tick, 0.01)

    def test_price_tick_from_binance_price_filter(self):
        class _Ex:
            markets = {
                "BTC/USDT:USDT": {
                    "precision": {"price": 2},
                    "info": {
                        "filters": [
                            {"filterType": "PRICE_FILTER", "tickSize": "0.10"},
                            {"filterType": "LOT_SIZE", "stepSize": "0.001"},
                        ]
                    },
                    "limits": {},
                }
            }

            def load_markets(self):
                return self.markets

            def market(self, sym):
                return self.markets[sym]

            def price_to_precision(self, sym, price):
                return "12345.6"

        from hub_ohlcv_lib import price_tick_from_market

        tick = price_tick_from_market(_Ex(), "BTC/USDT:USDT")
        self.assertAlmostEqual(tick, 0.10)

    def test_price_tick_from_info_tick_size(self):
        class _Ex:
            markets = {
                "INJ/USDT:USDT": {
                    "precision": {"price": 4},
                    "info": {"tickSize": "0.001"},
                    "limits": {},
                }
            }

            def load_markets(self):
                return self.markets

            def market(self, sym):
                return self.markets[sym]

            def price_to_precision(self, sym, price):
                return "7.123"

        from hub_ohlcv_lib import price_tick_from_market

        tick = price_tick_from_market(_Ex(), "INJ/USDT:USDT")
        self.assertAlmostEqual(tick, 0.001)

    def test_full_fetch_without_since_paginates_okx_style(self):
        """OKX 等无 since 单次约 300 根，须分页至 limit。"""
        from hub_ohlcv_lib import TIMEFRAME_MS

        step = TIMEFRAME_MS["1h"]
        want = 1000
        base = max(0, int(__import__("time").time() * 1000) - want * step)
        pages = [
            [[base + i * step, 1.0, 1.1, 0.9, 1.05, 100.0] for i in range(300)],
            [[base + (300 + i) * step, 2.0, 2.1, 1.9, 2.05, 200.0] for i in range(300)],
            [[base + (600 + i) * step, 3.0, 3.1, 2.9, 3.05, 300.0] for i in range(300)],
            [[base + (900 + i) * step, 4.0, 4.1, 3.9, 4.05, 400.0] for i in range(100)],
        ]
        ex = _FakeExchange(pages)

        out = fetch_ohlcv_for_hub(
            symbol="ONDO/USDT",
            timeframe="1h",
            since_ms=None,
            limit=want,
            normalize_symbol_input=lambda s: str(s).strip().upper(),
            normalize_exchange_symbol=lambda s: f"{s}:USDT" if ":" not in s else s,
            ensure_markets_loaded=lambda: None,
            exchange=ex,
        )
        self.assertTrue(out.get("ok"))
        self.assertEqual(len(out.get("bars") or []), 1000)
        self.assertGreaterEqual(len(ex.calls), 4)
        self.assertAlmostEqual(out["bars"][-1]["close"], 4.05)

    def test_pagination_continues_when_page_smaller_than_chunk(self):
        """Gate 等常返回 299 根/次，不应误判为已到末尾。"""
        base = 1_700_000_000_000
        step = 4 * 60 * 60 * 1000
        page1 = [
            [base + i * step, 1.0, 1.1, 0.9, 1.05, 100.0] for i in range(299)
        ]
        page2 = [
            [base + (299 + i) * step, 2.0, 2.1, 1.9, 2.05, 200.0] for i in range(299)
        ]
        page3 = [
            [base + (598 + i) * step, 3.0, 3.1, 2.9, 3.05, 300.0] for i in range(50)
        ]
        ex = _FakeExchange([page1, page2, page3])

        out = fetch_ohlcv_for_hub(
            symbol="INJ/USDT",
            timeframe="4h",
            since_ms=base,
            limit=600,
            normalize_symbol_input=lambda s: str(s).strip().upper(),
            normalize_exchange_symbol=lambda s: f"{s}:USDT" if ":" not in s else s,
            ensure_markets_loaded=lambda: None,
            exchange=ex,
        )
        self.assertTrue(out.get("ok"))
        self.assertEqual(len(out.get("bars") or []), 600)
        self.assertGreaterEqual(len(ex.calls), 3)
        self.assertAlmostEqual(out["bars"][-1]["close"], 3.05)

    def test_pagination_stops_when_next_since_reaches_now(self):
        """Gate 等：分页 since 不得越过当前时间，避免 from>to。"""
        from hub_ohlcv_lib import TIMEFRAME_MS

        step = TIMEFRAME_MS["1d"]
        now_ms = int(__import__("time").time() * 1000)
        # 最后一页最后一根 K 的 next_since 将 >= now_ms，应停止不再请求
        last_open = ((now_ms // step) - 2) * step
        page = [
            [last_open - step, 1.0, 1.1, 0.9, 1.0, 10.0],
            [last_open, 1.1, 1.2, 1.0, 1.1, 11.0],
        ]
        ex = _FakeExchange([page])

        out = fetch_ohlcv_for_hub(
            symbol="ONDO/USDT",
            timeframe="1d",
            since_ms=last_open - step * 5,
            limit=10,
            normalize_symbol_input=lambda s: str(s).strip().upper(),
            normalize_exchange_symbol=lambda s: f"{s}:USDT" if ":" not in s else s,
            ensure_markets_loaded=lambda: None,
            exchange=ex,
        )
        self.assertTrue(out.get("ok"))
        self.assertGreaterEqual(len(out.get("bars") or []), 2)
        self.assertLessEqual(len(ex.calls), 4)

    def test_aggregate_ohlcv_bars_buckets(self):
        from hub_ohlcv_lib import TIMEFRAME_MS

        h1 = TIMEFRAME_MS["1h"]
        h4 = TIMEFRAME_MS["4h"]
        base = (1_700_000_000_000 // h4) * h4
        src = [
            {
                "open_time_ms": base + i * h1,
                "open": 1.0,
                "high": 2.0,
                "low": 0.5,
                "close": 1.5,
                "volume": 1.0,
            }
            for i in range(4)
        ]
        out = aggregate_ohlcv_bars(src, "4h")
        self.assertEqual(len(out), 1)
        self.assertEqual(out[0]["volume"], 4.0)
        self.assertEqual(out[0]["high"], 2.0)
        self.assertEqual(out[0]["low"], 0.5)


if __name__ == "__main__":
    unittest.main()