Add hub-only OKX amp stats for ETH/BTC session windows.

Read-only 1H index candles, point amplitude metrics, history save and CSV export; no order-path changes.

Co-authored-by: Cursor <cursoragent@cursor.com>

tests/test_amp_stats_lib.py

@@ -0,0 +1,124 @@
+"""振幅统计核心逻辑单元测试(不打交易所)."""
+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,
+)
+
+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)
+
+    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()