feat: market divergence scan tabs (4h/1d/1w) with Top20 MACD confluence colors

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-06 19:58:30 +08:00
parent 81f7a7c451
commit e62a3f3351
7 changed files with 999 additions and 48 deletions
@@ -0,0 +1,101 @@
+import unittest
+
+from lib.hub.hub_divergence_scan_lib import (
+    analyze_ohlcv_bars,
+    build_symbol_scan_row,
+    compute_confluence,
+    detect_latest_macd_divergence,
+    filter_tab_items,
+)
+
+
+def _synthetic_bull_div_closes(n: int = 120) -> list[float]:
+    """价格双底 + MACD 抬高 → 底背离。"""
+    closes = [100.0] * n
+    # 下跌
+    for i in range(20, 40):
+        closes[i] = 100 - (i - 20) * 0.8
+    # 反弹
+    for i in range(40, 55):
+        closes[i] = closes[39] + (i - 40) * 0.5
+    # 再跌略破前低
+    for i in range(55, 75):
+        closes[i] = closes[54] - (i - 55) * 0.35
+    # 末尾企稳略抬
+    for i in range(75, n):
+        closes[i] = closes[74] + (i - 75) * 0.02
+    return closes
+
+
+class TestHubDivergenceScanLib(unittest.TestCase):
+    def test_compute_confluence_three_same(self):
+        tf = {
+            "4h": {"direction": "bull"},
+            "1d": {"direction": "bull"},
+            "1w": {"direction": "bull"},
+        }
+        c = compute_confluence(tf)
+        self.assertEqual(c["confluence"], 3)
+        self.assertEqual(c["confluence_css"], "c3")
+        self.assertFalse(c["is_split"])
+
+    def test_compute_confluence_split(self):
+        tf = {
+            "4h": {"direction": "bull"},
+            "1d": {"direction": "bear"},
+            "1w": {"direction": None},
+        }
+        c = compute_confluence(tf)
+        self.assertTrue(c["is_split"])
+        self.assertEqual(c["confluence_kind"], "分歧")
+        self.assertEqual(c["confluence_css"], "split")
+        self.assertIn("4h底", c["split_detail"])
+
+    def test_filter_tab_items_only_matching_tf(self):
+        items = [
+            build_symbol_scan_row(
+                rank=1,
+                symbol="AAA/USDT",
+                volume_label="1M",
+                tf_hits={
+                    "4h": {"direction": "bull", "bars_ago": 2, "open_time_ms": 1},
+                    "1d": {"direction": None},
+                    "1w": {"direction": None},
+                },
+            ),
+            build_symbol_scan_row(
+                rank=2,
+                symbol="BBB/USDT",
+                volume_label="2M",
+                tf_hits={
+                    "4h": {"direction": None},
+                    "1d": {"direction": "bear", "bars_ago": 1, "open_time_ms": 2},
+                    "1w": {"direction": None},
+                },
+            ),
+        ]
+        f4 = filter_tab_items(items, "4h")
+        self.assertEqual(len(f4), 1)
+        self.assertEqual(f4[0]["symbol"], "AAA/USDT")
+        f1d = filter_tab_items(items, "1d")
+        self.assertEqual(len(f1d), 1)
+        self.assertEqual(f1d[0]["symbol"], "BBB/USDT")
+
+    def test_detect_macd_divergence_may_hit_on_synthetic(self):
+        closes = _synthetic_bull_div_closes()
+        hit = detect_latest_macd_divergence(closes)
+        # 合成数据不保证必中，但函数应正常返回
+        self.assertIn(hit.get("direction"), (None, "bull", "bear"))
+
+    def test_analyze_ohlcv_bars_from_rows(self):
+        closes = [float(100 + i * 0.1) for i in range(80)]
+        bars = [
+            {"open_time_ms": i * 3600000, "close": c, "open": c, "high": c, "low": c}
+            for i, c in enumerate(closes)
+        ]
+        out = analyze_ohlcv_bars(bars)
+        self.assertIn("direction", out)
+
+
+if __name__ == "__main__":
+    unittest.main()