dekun
910c938d0a
Add page pauses, exponential backoff, and cooldown before swap fallback to avoid rate limits. Co-authored-by: Cursor <cursoragent@cursor.com>
330 lines
13 KiB
Python
330 lines
13 KiB
Python
"""振幅统计核心逻辑单元测试(不打交易所)."""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import date, datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from lib.hub.amp_stats_lib import (
|
|
build_export_csv,
|
|
compute_amp_stats,
|
|
compute_day_row,
|
|
list_settlement_dates,
|
|
summarize_rows,
|
|
window_bounds_for_settlement,
|
|
)
|
|
|
|
import httpx
|
|
|
|
TZ = ZoneInfo("Asia/Shanghai")
|
|
|
|
|
|
def _bar(ts_ms: int, o: float, h: float, l: float, c: float) -> dict:
|
|
return {"ts": ts_ms, "o": o, "h": h, "l": l, "c": c}
|
|
|
|
|
|
class AmpStatsLibTests(unittest.TestCase):
|
|
def test_window_cross_day_22_to_16(self):
|
|
start, end = window_bounds_for_settlement(date(2026, 7, 22), 22)
|
|
self.assertEqual(start.strftime("%Y-%m-%d %H:%M"), "2026-07-21 22:00")
|
|
self.assertEqual(end.strftime("%Y-%m-%d %H:%M"), "2026-07-22 16:00")
|
|
|
|
def test_window_same_day_8_to_16(self):
|
|
start, end = window_bounds_for_settlement(date(2026, 7, 22), 8)
|
|
self.assertEqual(start.strftime("%Y-%m-%d %H:%M"), "2026-07-22 08:00")
|
|
self.assertEqual(end.strftime("%Y-%m-%d %H:%M"), "2026-07-22 16:00")
|
|
|
|
def test_settlement_excludes_incomplete_today(self):
|
|
now = datetime(2026, 7, 22, 10, 0, tzinfo=TZ)
|
|
days = list_settlement_dates(sample_days=3, now=now)
|
|
self.assertEqual(days[0].isoformat(), "2026-07-21")
|
|
self.assertEqual(len(days), 3)
|
|
|
|
def test_settlement_includes_today_after_1600(self):
|
|
now = datetime(2026, 7, 22, 16, 0, tzinfo=TZ)
|
|
days = list_settlement_dates(sample_days=1, now=now)
|
|
self.assertEqual(days[0].isoformat(), "2026-07-22")
|
|
|
|
def test_day_row_points(self):
|
|
# 22:00 D-1 → 16:00 D; O=2000 H=2500 L=1800 C=2100 → up500 down200 amp700
|
|
settlement = date(2026, 7, 22)
|
|
start, end = window_bounds_for_settlement(settlement, 22)
|
|
bar_map = {}
|
|
t = int(start.timestamp() * 1000)
|
|
last = int((end.replace(hour=15)).timestamp() * 1000)
|
|
# first bar
|
|
bar_map[t] = {"o": 2000.0, "h": 2100.0, "l": 1950.0, "c": 2050.0}
|
|
cur = t + 3600 * 1000
|
|
while cur < last:
|
|
bar_map[cur] = {"o": 2050.0, "h": 2200.0, "l": 1900.0, "c": 2100.0}
|
|
cur += 3600 * 1000
|
|
# peak and trough somewhere
|
|
mid = t + 5 * 3600 * 1000
|
|
bar_map[mid] = {"o": 2100.0, "h": 2500.0, "l": 1800.0, "c": 2000.0}
|
|
bar_map[last] = {"o": 2000.0, "h": 2150.0, "l": 1990.0, "c": 2100.0}
|
|
# fill any missing hours with flat
|
|
cur = t
|
|
while cur <= last:
|
|
if cur not in bar_map:
|
|
bar_map[cur] = {"o": 2000.0, "h": 2000.0, "l": 2000.0, "c": 2000.0}
|
|
cur += 3600 * 1000
|
|
row = compute_day_row(settlement, 22, bar_map)
|
|
self.assertIsNotNone(row)
|
|
self.assertEqual(row["open"], 2000.0)
|
|
self.assertEqual(row["high"], 2500.0)
|
|
self.assertEqual(row["low"], 1800.0)
|
|
self.assertEqual(row["up_points"], 500.0)
|
|
self.assertEqual(row["down_points"], 200.0)
|
|
self.assertEqual(row["amplitude"], 700.0)
|
|
self.assertEqual(row["change"], 100.0)
|
|
|
|
def test_summary_max_amplitude(self):
|
|
rows = [
|
|
{"amplitude": 100, "up_points": 40, "down_points": 60, "change": 10, "settlement_day": "2026-07-01"},
|
|
{"amplitude": 700, "up_points": 500, "down_points": 200, "change": -5, "settlement_day": "2026-07-02"},
|
|
{"amplitude": 200, "up_points": 50, "down_points": 150, "change": 20, "settlement_day": "2026-07-03"},
|
|
]
|
|
s = summarize_rows(rows)
|
|
self.assertEqual(s["max_amplitude"], 700)
|
|
self.assertEqual(s["max_amplitude_day"], "2026-07-02")
|
|
self.assertEqual(s["max_up_points"], 500)
|
|
self.assertEqual(s["max_down_points"], 200)
|
|
self.assertIsNone(s["straddle"])
|
|
|
|
def test_long_straddle_stats(self):
|
|
rows = [
|
|
# |chg|=40>30 win+10; up=40>30; down=10
|
|
{"up_points": 40, "down_points": 10, "change": 40, "amplitude": 50, "settlement_day": "2026-07-01"},
|
|
# |chg|=10 lose-20; up=5; down=35>30
|
|
{"up_points": 5, "down_points": 35, "change": -10, "amplitude": 40, "settlement_day": "2026-07-02"},
|
|
# |chg|=30 not >30 lose-30; boundary
|
|
{"up_points": 30, "down_points": 30, "change": 30, "amplitude": 60, "settlement_day": "2026-07-03"},
|
|
]
|
|
s = summarize_rows(rows, straddle_premium=30)
|
|
st = s["straddle"]
|
|
self.assertEqual(st["side"], "long_straddle")
|
|
self.assertEqual(st["premium"], 30)
|
|
self.assertEqual(st["up_exceed_days"], 1) # only 40
|
|
self.assertEqual(st["down_exceed_days"], 1) # only 35
|
|
self.assertEqual(st["abs_change_exceed_days"], 1) # only 40
|
|
self.assertAlmostEqual(st["pnl_total"], 40 - 30 + 10 - 30 + 30 - 30)
|
|
self.assertEqual(st["win_days"], 1)
|
|
self.assertEqual(st["win_ratio"], round(1 / 3, 4))
|
|
csv_text = build_export_csv(
|
|
{"exchange": "okx", "symbol_label": "ETH", "summary": s, "rows": rows, "start_hour": 22, "end_hour": 16}
|
|
)
|
|
self.assertIn("买跨对照", csv_text)
|
|
self.assertIn("买跨点数盈亏合计", csv_text)
|
|
|
|
def test_take_profit_and_weekend(self):
|
|
from lib.hub.amp_stats_lib import (
|
|
enrich_rows_pnl,
|
|
filter_weekend_rows,
|
|
reframe_amp_stats,
|
|
)
|
|
|
|
# Sat 2026-07-18, Sun 2026-07-19, Mon 2026-07-20
|
|
rows = [
|
|
{
|
|
"settlement_day": "2026-07-18",
|
|
"is_weekend": True,
|
|
"weekday_label": "六",
|
|
"up_points": 100,
|
|
"down_points": 10,
|
|
"change": -5,
|
|
"amplitude": 110,
|
|
},
|
|
{
|
|
"settlement_day": "2026-07-19",
|
|
"is_weekend": True,
|
|
"weekday_label": "日",
|
|
"up_points": 20,
|
|
"down_points": 15,
|
|
"change": 12,
|
|
"amplitude": 35,
|
|
},
|
|
{
|
|
"settlement_day": "2026-07-20",
|
|
"is_weekend": False,
|
|
"weekday_label": "",
|
|
"up_points": 50,
|
|
"down_points": 40,
|
|
"change": 8,
|
|
"amplitude": 90,
|
|
},
|
|
]
|
|
excl = filter_weekend_rows(rows, "exclude")
|
|
self.assertEqual(len(excl), 1)
|
|
self.assertEqual(excl[0]["settlement_day"], "2026-07-20")
|
|
only = filter_weekend_rows(rows, "only")
|
|
self.assertEqual(len(only), 2)
|
|
|
|
# TP=80: day1 hit → move 80; day2 no → |12|; day3 no → 8
|
|
enriched = enrich_rows_pnl(rows, straddle_premium=10, take_profit=80)
|
|
self.assertTrue(enriched[0]["take_profit_hit"])
|
|
self.assertEqual(enriched[0]["effective_move"], 80)
|
|
self.assertEqual(enriched[0]["profit"], 70)
|
|
self.assertFalse(enriched[1]["take_profit_hit"])
|
|
self.assertEqual(enriched[1]["effective_move"], 12)
|
|
self.assertEqual(enriched[1]["profit"], 2)
|
|
# TP empty → use |change|
|
|
no_tp = enrich_rows_pnl(rows[:1], straddle_premium=10, take_profit=None)
|
|
self.assertEqual(no_tp[0]["effective_move"], 5)
|
|
self.assertEqual(no_tp[0]["profit"], -5)
|
|
|
|
# TP boundary >= : up=80 counts as hit
|
|
edge = enrich_rows_pnl(
|
|
[{"up_points": 80, "down_points": 1, "change": 2, "settlement_day": "2026-07-20", "is_weekend": False}],
|
|
straddle_premium=10,
|
|
take_profit=80,
|
|
)
|
|
self.assertTrue(edge[0]["take_profit_hit"])
|
|
self.assertEqual(edge[0]["profit"], 70)
|
|
|
|
reframed = reframe_amp_stats(
|
|
rows_all=rows,
|
|
symbol="eth",
|
|
weekend_filter="exclude",
|
|
straddle_premium=10,
|
|
take_profit=80,
|
|
)
|
|
self.assertEqual(reframed["summary"]["sample_count"], 1)
|
|
# Mon: 未触达止盈 → |8|-10
|
|
self.assertEqual(reframed["rows"][0]["profit"], -2)
|
|
self.assertIn("收益", build_export_csv(reframed))
|
|
|
|
def test_fetch_switches_to_history_endpoint(self):
|
|
"""近期接口到头后应切 history 续拉."""
|
|
from lib.hub.amp_stats_lib import fetch_okx_candles
|
|
|
|
calls: list[str] = []
|
|
|
|
class FakeResp:
|
|
def __init__(self, data, url="https://x", status_code=200):
|
|
self._data = data
|
|
self.status_code = status_code
|
|
self.url = url
|
|
self.request = httpx.Request("GET", url)
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise httpx.HTTPStatusError(
|
|
"err", request=self.request, response=self
|
|
)
|
|
|
|
def json(self):
|
|
return {"code": "0", "data": self._data}
|
|
|
|
class FakeClient:
|
|
def get(self, url, params=None):
|
|
calls.append(url)
|
|
after = (params or {}).get("after")
|
|
# recent: only 2 pages then empty; history continues
|
|
if "history" not in url:
|
|
if after is None:
|
|
return FakeResp([["2000", "1", "2", "0.5", "1.5"], ["1900", "1", "2", "0.5", "1.5"]], url=url)
|
|
if after == "1900":
|
|
return FakeResp([], url=url) # recent exhausted
|
|
return FakeResp([], url=url)
|
|
# history
|
|
if after == "1900":
|
|
return FakeResp([["1800", "1", "2", "0.5", "1.5"], ["1000", "1", "2", "0.5", "1.5"]], url=url)
|
|
return FakeResp([], url=url)
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
bars = fetch_okx_candles(
|
|
url="https://www.okx.com/api/v5/market/index-candles",
|
|
history_url="https://www.okx.com/api/v5/market/history-index-candles",
|
|
inst_id="ETH-USD",
|
|
since_ms=1000,
|
|
until_ms=3000,
|
|
client=FakeClient(),
|
|
max_pages=10,
|
|
page_pause_sec=0,
|
|
history_page_pause_sec=0,
|
|
)
|
|
self.assertTrue(any("history-index-candles" in u for u in calls))
|
|
self.assertGreaterEqual(len(bars), 3)
|
|
self.assertEqual(bars[0]["ts"], 1000)
|
|
|
|
def test_fetch_retries_on_429(self):
|
|
from lib.hub.amp_stats_lib import fetch_okx_candles
|
|
import httpx as _httpx
|
|
|
|
hits = {"n": 0}
|
|
|
|
class FakeResp:
|
|
def __init__(self, status_code, data=None):
|
|
self.status_code = status_code
|
|
self.url = "https://www.okx.com/api/v5/market/history-candles"
|
|
self.request = _httpx.Request("GET", self.url)
|
|
self._data = data or []
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise _httpx.HTTPStatusError("429", request=self.request, response=self)
|
|
|
|
def json(self):
|
|
return {"code": "0", "data": self._data}
|
|
|
|
class FakeClient:
|
|
def get(self, url, params=None):
|
|
hits["n"] += 1
|
|
if hits["n"] < 3:
|
|
return FakeResp(429)
|
|
return FakeResp(200, [["1000", "1", "2", "0.5", "1.5"]])
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
bars = fetch_okx_candles(
|
|
url="https://www.okx.com/api/v5/market/candles",
|
|
history_url=None,
|
|
inst_id="ETH-USDT-SWAP",
|
|
since_ms=1000,
|
|
until_ms=2000,
|
|
client=FakeClient(),
|
|
max_pages=3,
|
|
page_pause_sec=0,
|
|
history_page_pause_sec=0,
|
|
)
|
|
self.assertGreaterEqual(hits["n"], 3)
|
|
self.assertEqual(len(bars), 1)
|
|
|
|
def test_compute_with_mock_fetch(self):
|
|
now = datetime(2026, 7, 22, 18, 0, tzinfo=TZ)
|
|
|
|
def fetch_fn(*, inst_id, since_ms, until_ms):
|
|
bars = []
|
|
t = since_ms - (since_ms % (3600 * 1000))
|
|
while t <= until_ms:
|
|
# synthetic: open 2000, one spike day
|
|
o = 2000.0
|
|
h = 2500.0 if t == since_ms + 5 * 3600 * 1000 else 2050.0
|
|
l = 1800.0 if t == since_ms + 5 * 3600 * 1000 else 1950.0
|
|
c = 2020.0
|
|
bars.append(_bar(t, o, h, l, c))
|
|
t += 3600 * 1000
|
|
return bars
|
|
|
|
result = compute_amp_stats(
|
|
symbol="eth",
|
|
start_hour=16,
|
|
period="custom",
|
|
custom_days=7,
|
|
now=now,
|
|
fetch_fn=fetch_fn,
|
|
)
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(result["exchange"], "okx")
|
|
self.assertGreaterEqual(result["summary"]["sample_count"], 1)
|
|
csv_text = build_export_csv(result)
|
|
self.assertIn("最大振幅", csv_text)
|
|
self.assertIn("日表明细", csv_text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|