""" StructureZone 系统单元测试 """ import sys import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import pytest from ChanZone import ( RawZonePoint, StructureZone, StructureZoneConfig, cluster_raw_points, build_structure_zones, _calc_strength, _calc_confidence, _calc_recency, extract_raw_points_from_serialized, analyze_structure_zones_from_serialized, ) class TestClusterRawPoints: """聚类算法测试""" def test_empty_points(self): config = StructureZoneConfig() result = cluster_raw_points([], config) assert result == [] def test_single_point_filtered(self): """单点被 min_overlap 过滤""" config = StructureZoneConfig(min_overlap_for_zone=2) points = [RawZonePoint(price=100, timeframe='5m', structure_type='bi_zhongshu', boundary_type='ZG', source_zs_id=0, is_sure=True)] result = cluster_raw_points(points, config) assert result == [] def test_two_nearby_points_merge(self): """相邻价格点归为一类""" config = StructureZoneConfig(cluster_radius_pct=1.0, min_overlap_for_zone=2) points = [ RawZonePoint(price=100, timeframe='5m', structure_type='bi_zhongshu', boundary_type='ZG', source_zs_id=0, is_sure=True), RawZonePoint(price=100.5, timeframe='15m', structure_type='xd_zhongshu', boundary_type='ZD', source_zs_id=0, is_sure=True), ] result = cluster_raw_points(points, config) assert len(result) == 1 assert len(result[0]) == 2 def test_two_distant_points_separate(self): """远离的价格点不归为一类""" config = StructureZoneConfig(cluster_radius_pct=0.1, min_overlap_for_zone=1) # 先用 1 看聚类 points = [ RawZonePoint(price=100, timeframe='5m', structure_type='bi_zhongshu', boundary_type='ZG', source_zs_id=0, is_sure=True), RawZonePoint(price=110, timeframe='15m', structure_type='xd_zhongshu', boundary_type='ZD', source_zs_id=0, is_sure=True), ] # 先用 min_overlap=2 确认被过滤 config2 = StructureZoneConfig(cluster_radius_pct=0.1, min_overlap_for_zone=2) result = cluster_raw_points(points, config2) assert result == [] # 两个单独点，都不够 min_overlap def test_multi_tf_convergence(self): """多个时间周期在相同价格区间聚合""" config = StructureZoneConfig(cluster_radius_pct=1.0, min_overlap_for_zone=2) points = [] for tf in ['5m', '15m', '30m', '1h']: for btype in ['ZG', 'ZD']: points.append(RawZonePoint(price=100 + abs(hash(tf + btype)) % 3 * 0.1, timeframe=tf, structure_type='bi_zhongshu', boundary_type=btype, source_zs_id=0, is_sure=True)) result = cluster_raw_points(points, config) assert len(result) >= 1 # 所有点应该聚合在一起（价差很小） total = sum(len(c) for c in result) assert total == len(points) class TestBuildStructureZones: """评分和构建测试""" def _make_cluster(self, prices, tf='5m', st='bi_zhongshu'): return [RawZonePoint(price=p, timeframe=tf, structure_type=st, boundary_type='ZG', source_zs_id=0, is_sure=True, candle_time='2025-01-01T00:00:00') for p in prices] def test_zone_type_support(self): """当前价上方区间是阻力，下方是支撑""" config = StructureZoneConfig() cluster = self._make_cluster([90, 92]) ema52 = {'5m': 100} zones = build_structure_zones([cluster], current_price=100, ema52_values=ema52, latest_candle_time='2025-01-01T01:00:00', config=config) assert len(zones) == 1 assert zones[0].zone_type == 'support' # 在价格下方 def test_zone_type_resistance(self): config = StructureZoneConfig() cluster = self._make_cluster([110, 112]) ema52 = {'5m': 100} zones = build_structure_zones([cluster], current_price=100, ema52_values=ema52, latest_candle_time='2025-01-01T01:00:00', config=config) assert len(zones) == 1 assert zones[0].zone_type == 'resistance' def test_zone_type_neutral(self): config = StructureZoneConfig() cluster = self._make_cluster([95, 105]) ema52 = {'5m': 100} zones = build_structure_zones([cluster], current_price=100, ema52_values=ema52, latest_candle_time='2025-01-01T01:00:00', config=config) assert len(zones) == 1 assert zones[0].zone_type == 'neutral' def test_strength_score_range(self): """评分在 0-100 之间""" config = StructureZoneConfig() cluster = self._make_cluster([100, 102, 104], '5m', 'bi_zhongshu') cluster += self._make_cluster([100.5, 102.5], '15m', 'xd_zhongshu') ema52 = {'5m': 0, '15m': 0} zones = build_structure_zones([cluster], current_price=110, ema52_values=ema52, latest_candle_time='2025-01-01T01:00:00', config=config) assert len(zones) == 1 assert 0 <= zones[0].strength_score <= 100 def test_ema52_aligned_true(self): """EMA52 落在区间内""" config = StructureZoneConfig() cluster = self._make_cluster([95, 105]) ema52 = {'5m': 100} zones = build_structure_zones([cluster], current_price=110, ema52_values=ema52, latest_candle_time='2025-01-01T01:00:00', config=config) assert zones[0].ema52_aligned is True def test_ema52_aligned_false(self): """EMA52 不在区间内""" config = StructureZoneConfig() cluster = self._make_cluster([95, 105]) ema52 = {'5m': 120} zones = build_structure_zones([cluster], current_price=110, ema52_values=ema52, latest_candle_time='2025-01-01T01:00:00', config=config) assert zones[0].ema52_aligned is False def test_confidence_range(self): """置信度在 0-1 之间""" config = StructureZoneConfig() cluster = self._make_cluster([100, 101, 102, 103]) ema52 = {} zones = build_structure_zones([cluster], current_price=110, ema52_values=ema52, latest_candle_time='2025-01-01T01:00:00', config=config) assert 0 <= zones[0].confidence <= 1 def test_max_zones_cap(self): """max_zones 限制返回数量""" config = StructureZoneConfig(max_zones=3) clusters = [self._make_cluster([100 + i * 10, 100 + i * 10 + 2]) for i in range(10)] ema52 = {} zones = build_structure_zones(clusters, current_price=150, ema52_values=ema52, latest_candle_time='2025-01-01T01:00:00', config=config) assert len(zones) <= 3 def test_sorted_by_strength(self): """按 strength 降序排列""" config = StructureZoneConfig(max_zones=0) # 创建一个有更多重叠的聚类（更强）和一个较弱的聚类 cluster_strong = self._make_cluster([100, 101, 102, 103, 104]) # 5 点 cluster_weak = self._make_cluster([200, 201]) # 2 点 ema52 = {'5m': 0} zones = build_structure_zones([cluster_weak, cluster_strong], current_price=150, ema52_values=ema52, latest_candle_time='2025-01-01T01:00:00', config=config) assert zones[0].strength_score >= zones[-1].strength_score class TestExtractFromSerialized: """从序列化数据提取测试""" def test_empty_analyses(self): config = StructureZoneConfig() points = extract_raw_points_from_serialized({}, {}, config) assert points == [] def test_basic_extraction(self): analyses = { '5m': { 'bi_zs_list': [ {'zg': 100, 'zd': 95, 'gg': 102, 'dd': 93, 'is_sure': True, 'end_time': '2025-01-01T00:00'}, ], 'zs_list': [], }, '15m': { 'bi_zs_list': [], 'zs_list': [ {'zg': 105, 'zd': 98, 'gg': 107, 'dd': 96, 'is_sure': True, 'end_time': '2025-01-01T00:00'}, ], }, } ema52 = {'5m': 101, '15m': 103} config = StructureZoneConfig(zone_timeframes=['5m', '15m']) points = extract_raw_points_from_serialized(analyses, ema52, config) # bi_zs: 4 points (ZG/ZD/GG/DD) + zs_list: 4 points + 2 ema52 = 10 assert len(points) == 10 def test_unsure_filtered(self): analyses = { '5m': { 'bi_zs_list': [ {'zg': 100, 'zd': 95, 'gg': 102, 'dd': 93, 'is_sure': False}, ], 'zs_list': [], }, } ema52 = {} config = StructureZoneConfig(zone_timeframes=['5m']) points = extract_raw_points_from_serialized(analyses, ema52, config) assert len(points) == 0 # is_sure=False 被过滤 def test_timeframe_filtering(self): """仅提取 config.zone_timeframes 中的周期""" analyses = { '5m': { 'bi_zs_list': [ {'zg': 100, 'zd': 95, 'gg': 102, 'dd': 93, 'is_sure': True}, ], 'zs_list': [], }, '1h': { 'bi_zs_list': [ {'zg': 200, 'zd': 190, 'gg': 205, 'dd': 188, 'is_sure': True}, ], 'zs_list': [], }, } ema52 = {'5m': 101, '1h': 195} config = StructureZoneConfig(zone_timeframes=['5m']) # 只取 5m points = extract_raw_points_from_serialized(analyses, ema52, config) # 只有 5m: 4 bi_zs + 1 ema52 = 5 assert len(points) == 5 assert all(p.timeframe == '5m' for p in points) class TestScoringHelpers: """评分辅助函数测试""" def test_calc_recency_same_time(self): """同一时间的 recency = 1.0""" score = _calc_recency('2025-01-01T00:00:00', '2025-01-01T00:00:00', 50) assert score == 1.0 def test_calc_recency_invalid(self): """无效时间的 recency = 0.5""" score = _calc_recency(None, '2025-01-01T00:00:00', 50) assert score == 0.5 def test_calc_confidence_high(self): """高重叠数 = 高置信度""" conf = _calc_confidence(6, 3, []) assert conf > 0.7 def test_calc_confidence_low(self): """低重叠数 = 低置信度""" conf = _calc_confidence(2, 1, []) assert conf < 0.7 class TestAnalyzeFromSerialized: """端到端测试（从序列化数据到 StructureZone）""" def test_end_to_end(self): analyses = { '5m': { 'bi_zs_list': [ {'zg': 100, 'zd': 95, 'gg': 102, 'dd': 93, 'is_sure': True, 'end_time': '2025-01-01T00:00'}, ], 'zs_list': [], }, '15m': { 'bi_zs_list': [ {'zg': 101, 'zd': 96, 'gg': 103, 'dd': 94, 'is_sure': True, 'end_time': '2025-01-01T00:01'}, ], 'zs_list': [], }, } ema52_dict = {'5m': 100.5, '15m': 99.5} config = StructureZoneConfig(zone_timeframes=['5m', '15m'], cluster_radius_pct=2.0) zones = analyze_structure_zones_from_serialized(analyses, ema52_dict, 110, config) # 两个 TF 的 BI_ZS 价格接近，应聚合成一个区间 assert len(zones) >= 1 zone = zones[0] assert zone.zone_type == 'support' # 价格在 93-103，current_price=110 assert '5m' in zone.timeframes assert '15m' in zone.timeframes assert zone.structure_types == ['bi_zhongshu'] assert 0 <= zone.strength_score <= 100 assert 0 <= zone.confidence <= 1 def test_empty_returns_empty(self): zones = analyze_structure_zones_from_serialized({}, {}, 100) assert zones == [] if __name__ == '__main__': pytest.main([__file__, '-v'])