dekun
53863559f4
Retarget git remote, install path, and deploy docs from crypto_monitor to crypto_monitor_user. Co-authored-by: Cursor <cursoragent@cursor.com>
102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
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()
|