dekun
96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
"""行情区 chart 后台轮询订阅。"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
sys.path.insert(0, str(ROOT / "manual_trading_hub"))
|
|
|
|
from hub_chart_cache import ChartPollStore, series_key # noqa: E402
|
|
|
|
|
|
class TestHubChartCache(unittest.TestCase):
|
|
def test_series_key(self) -> None:
|
|
self.assertEqual(series_key("Gate_X", "hype/usdt", "5m"), "gate_x|HYPE/USDT|5m")
|
|
|
|
def test_position_and_watch_keys(self) -> None:
|
|
store = ChartPollStore()
|
|
store.sync_positions_from_rows(
|
|
[
|
|
{
|
|
"key": "okx_auto",
|
|
"agent": {
|
|
"ok": True,
|
|
"positions": [{"symbol": "BTC/USDT"}, {"symbol": "ETH/USDT"}],
|
|
},
|
|
}
|
|
]
|
|
)
|
|
store.touch_watch("gate_trend", "HYPE/USDT", "5m")
|
|
keys = store.active_series_keys()
|
|
self.assertIn(series_key("okx_auto", "BTC/USDT", "5m"), keys)
|
|
self.assertIn(series_key("gate_trend", "HYPE/USDT", "5m"), keys)
|
|
|
|
def test_note_series_result_pushes_tail_candles(self) -> None:
|
|
store = ChartPollStore()
|
|
key = series_key("binance", "BTC/USDT", "15m")
|
|
candles = [
|
|
{"time": 1_700_000_000 + i * 900, "open": 1, "high": 2, "low": 0.5, "close": 1.5, "volume": 10}
|
|
for i in range(40)
|
|
]
|
|
store.note_series_result(
|
|
"binance",
|
|
"BTC/USDT",
|
|
"15m",
|
|
ok=True,
|
|
fetched=3,
|
|
candles=candles,
|
|
price_tick=0.01,
|
|
)
|
|
ev = store.event_dict()
|
|
self.assertIn("tails", ev)
|
|
self.assertIn(key, ev["tails"])
|
|
tail = ev["tails"][key]
|
|
self.assertEqual(len(tail["candles"]), 30)
|
|
self.assertEqual(tail["price_tick"], 0.01)
|
|
self.assertGreater(tail["series_version"], 0)
|
|
|
|
def test_broadcast_clears_pending_tails(self) -> None:
|
|
store = ChartPollStore()
|
|
store.note_series_result(
|
|
"gate",
|
|
"ONDO/USDT",
|
|
"5m",
|
|
ok=True,
|
|
candles=[{"time": 100, "open": 1, "high": 1, "low": 1, "close": 1, "volume": 1}],
|
|
)
|
|
store._broadcast()
|
|
ev = store.event_dict()
|
|
self.assertNotIn("tails", ev)
|
|
|
|
def test_poll_increments_version(self) -> None:
|
|
async def run() -> None:
|
|
store = ChartPollStore()
|
|
n = 0
|
|
|
|
async def poll():
|
|
nonlocal n
|
|
n += 1
|
|
store.touch_watch("binance", "BTC/USDT", "1d")
|
|
return {"ok": True, "n": n}
|
|
|
|
await store.start(poll)
|
|
await asyncio.sleep(0.05)
|
|
self.assertGreaterEqual(store.version, 1)
|
|
await store.stop()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|