"""振幅统计核心逻辑单元测试(不打交易所).""" 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"]) self.assertIsNone(s["perp_hedge"]) def test_perp_hedge_hit_and_pnl(self): from lib.hub.amp_stats_lib import perp_hedge_day_pnl # 开盘=1800 optLev=100 → prem/coin=18; 1:2 → premium=36 # A move ≈ 52.83; B portfolio move = 51 rows = [ {"open": 1800, "close": 1860, "change": 60, "up_points": 60, "down_points": 0, "amplitude": 60, "settlement_day": "2026-07-01"}, {"open": 1800, "close": 1740, "change": -60, "up_points": 0, "down_points": 60, "amplitude": 60, "settlement_day": "2026-07-02"}, {"open": 1800, "close": 1820, "change": 20, "up_points": 20, "down_points": 0, "amplitude": 20, "settlement_day": "2026-07-03"}, ] hedge = { "target_profit_u": 15, "perp_leverage": 10, "option_leverage": 100, "ratio_perp": 1, "ratio_opt": 2, "ct_mult": 0.01, } s = summarize_rows(rows, perp_hedge=hedge) ph = s["perp_hedge"] self.assertIsNotNone(ph) self.assertEqual(ph["entry"], "open") self.assertEqual(ph["spot"], 1800.0) self.assertEqual(ph["opt_coins"], 2.0) self.assertEqual(ph["premium_total"], 36.0) self.assertAlmostEqual(ph["move_b"], 51.0, places=4) self.assertEqual(ph["hit_a_days"], 1) # only +60 self.assertEqual(ph["hit_b_days"], 1) # only -60 up_pnl = perp_hedge_day_pnl( change=60, open_px=1800, close_px=1860, option_leverage=100, opt_coins=2 ) down_pnl = perp_hedge_day_pnl( change=-60, open_px=1800, close_px=1740, option_leverage=100, opt_coins=2 ) self.assertAlmostEqual(down_pnl, 60 * (2 - 1) - 36, places=4) # 24 self.assertAlmostEqual(ph["down_pnl_total"], down_pnl, places=4) self.assertGreater(up_pnl, 0) self.assertEqual(ph["up_days"], 2) self.assertEqual(ph["down_days"], 1) # 不同开盘 → 不同权利金 hi_open_pnl = perp_hedge_day_pnl( change=-60, open_px=2000, close_px=1940, option_leverage=100, opt_coins=2 ) self.assertAlmostEqual(hi_open_pnl, 60 - 40, places=4) # prem=40 csv_text = build_export_csv( { "exchange": "okx", "symbol_label": "ETH", "summary": s, "rows": rows, "start_hour": 16, "end_hour": 16, } ) self.assertIn("永期对冲对照", csv_text) self.assertIn("永期盈亏", csv_text) self.assertIn("按日开盘", csv_text) 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()