dekun
bfffc7d984
- fetch_ohlcv_for_hub：无 since 时按目标根数分页拉取（OKX/Gate 单次约 300） - hub_kline_store 全量补拉传 fetch_start_ms - 行情区：数字键切换周期、Ctrl+空格全屏、Esc 退出全屏 Co-authored-by: Cursor <cursoragent@cursor.com>
131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
"""hub_ohlcv_lib：分页拉取（Gate 等单次不足 chunk 时仍继续）。"""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from hub_ohlcv_lib import fetch_ohlcv_for_hub
|
|
|
|
|
|
class _FakeExchange:
|
|
def __init__(self, pages):
|
|
self.pages = list(pages)
|
|
self.calls = []
|
|
self.markets = {}
|
|
|
|
def fetch_ohlcv(self, symbol, timeframe=None, since=None, limit=None):
|
|
self.calls.append({"symbol": symbol, "since": since, "limit": limit})
|
|
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_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_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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|