jackyu66gitandClaude Opus 4.6
ebcb3dce73
- 新增 ChanZone.py: 从笔中枢/线段中枢/EMA52 提取价格区，聚类评分 - ChanLun.py 新增 get_structure_zones() 方法 - web/app.py: 独立拉取多周期数据 + 缓存 + limit 传参避免全量传输 - web/index.html: 结构区勾选框 + K线数量输入 + 半透明填充区绘制 - tests/test_chan_zone.py: 24 个单元测试 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
301 lines
12 KiB
Python
301 lines
12 KiB
Python
"""
|
||
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'])
|