223 lines
8.1 KiB
Python
223 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
KLU与KLC分型强度算法一致性测试
|
||
验证两种算法在相同数据下是否产生一致的结果
|
||
"""
|
||
|
||
from ChanKLU import ChanKLU
|
||
from ChanKLC import ChanKLC
|
||
from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR
|
||
import pandas as pd
|
||
from datetime import datetime, timedelta
|
||
|
||
def create_test_data():
|
||
"""创建测试用的K线数据"""
|
||
test_cases = [
|
||
# 测试用例1：标准顶分型
|
||
{
|
||
'name': '标准顶分型',
|
||
'data': [
|
||
{'open': 100, 'high': 102, 'low': 99, 'close': 101, 'volume': 1000}, # K1
|
||
{'open': 101, 'high': 105, 'low': 100, 'close': 103, 'volume': 1500}, # K2 (顶分型中心)
|
||
{'open': 103, 'high': 104, 'low': 98, 'close': 99, 'volume': 1200}, # K3
|
||
]
|
||
},
|
||
# 测试用例2：标准底分型
|
||
{
|
||
'name': '标准底分型',
|
||
'data': [
|
||
{'open': 100, 'high': 102, 'low': 99, 'close': 101, 'volume': 1000}, # K1
|
||
{'open': 101, 'high': 103, 'low': 95, 'close': 97, 'volume': 1500}, # K2 (底分型中心)
|
||
{'open': 97, 'high': 104, 'low': 96, 'close': 102, 'volume': 1200}, # K3
|
||
]
|
||
},
|
||
# 测试用例3：强势顶分型（放量+下影线）
|
||
{
|
||
'name': '强势顶分型',
|
||
'data': [
|
||
{'open': 100, 'high': 102, 'low': 99, 'close': 101, 'volume': 1000}, # K1
|
||
{'open': 101, 'high': 108, 'low': 100, 'close': 102, 'volume': 2500}, # K2 (强顶分型)
|
||
{'open': 102, 'high': 103, 'low': 95, 'close': 96, 'volume': 1800}, # K3 (大阴线确认)
|
||
]
|
||
}
|
||
]
|
||
return test_cases
|
||
|
||
def setup_klu_chain(data_list):
|
||
"""设置KLU链"""
|
||
klus = []
|
||
base_time = datetime.now()
|
||
|
||
for i, data in enumerate(data_list):
|
||
time_str = (base_time + timedelta(minutes=i)).strftime("%Y-%m-%d %H:%M:%S")
|
||
klu = ChanKLU(time_str, data['open'], data['high'], data['low'], data['close'], data['volume'])
|
||
klu.set_idx(i)
|
||
|
||
# 设置基础技术指标
|
||
indicators = {
|
||
'ma5': data['close'] + (i-1) * 0.1,
|
||
'ma10': data['close'] + (i-1) * 0.05,
|
||
'rsi': 50 + (i % 3 - 1) * 15,
|
||
'macd': (i % 3 - 1) * 0.01,
|
||
'macdhist': (i % 2) * 0.005,
|
||
'volume_ratio': 1.0 + (i % 2) * 0.3
|
||
}
|
||
klu.set_indicators(indicators)
|
||
klus.append(klu)
|
||
|
||
# 建立前后关系
|
||
for i in range(len(klus)):
|
||
if i > 0:
|
||
klus[i].set_pre(klus[i-1])
|
||
if i < len(klus) - 1:
|
||
klus[i].set_next(klus[i+1])
|
||
|
||
return klus
|
||
|
||
def setup_klc_chain(data_list):
|
||
"""设置KLC链（基于KLU）"""
|
||
klus = setup_klu_chain(data_list)
|
||
klcs = []
|
||
|
||
# 为简化测试，假设每个KLU对应一个KLC（无包含关系处理）
|
||
for i, klu in enumerate(klus):
|
||
klc = ChanKLC(klu, i, Chan_KLINE_DIR.UP)
|
||
klc.set_end_klu(klu)
|
||
klcs.append(klc)
|
||
|
||
# 建立前后关系
|
||
for i in range(len(klcs)):
|
||
if i > 0:
|
||
klcs[i].set_pre(klcs[i-1])
|
||
if i < len(klcs) - 1:
|
||
klcs[i].set_next(klcs[i+1])
|
||
|
||
# 设置分型类型
|
||
if len(klcs) >= 3:
|
||
middle_klc = klcs[1]
|
||
if (middle_klc.high > klcs[0].high and middle_klc.high > klcs[2].high):
|
||
middle_klc.set_fx(Chan_FX_TYPE.TOP)
|
||
elif (middle_klc.low < klcs[0].low and middle_klc.low < klcs[2].low):
|
||
middle_klc.set_fx(Chan_FX_TYPE.BOTTOM)
|
||
|
||
return klcs
|
||
|
||
def compare_algorithms(test_cases):
|
||
"""对比KLU和KLC算法"""
|
||
|
||
print("=" * 80)
|
||
print("KLU与KLC分型强度算法一致性测试")
|
||
print("=" * 80)
|
||
|
||
for case in test_cases:
|
||
print(f"\n🔍 测试用例: {case['name']}")
|
||
print("-" * 50)
|
||
|
||
# 准备数据
|
||
klus = setup_klu_chain(case['data'])
|
||
klcs = setup_klc_chain(case['data'])
|
||
|
||
if len(klus) >= 3 and len(klcs) >= 3:
|
||
middle_klu = klus[1]
|
||
middle_klc = klcs[1]
|
||
|
||
# KLU分析
|
||
middle_klu.update_realtime_analysis()
|
||
klu_fx_type = middle_klu.fx_type
|
||
klu_strength = middle_klu.fx_strength
|
||
klu_confirmed = middle_klu.fx_confirmed
|
||
|
||
# KLC分析
|
||
klc_fx_type = middle_klc.fx
|
||
klc_strength_raw = middle_klc.cal_fx_strength() # -3到3
|
||
klc_strength_converted = int((klc_strength_raw + 3) * 100 / 6) # 转换为0-100
|
||
|
||
# 输出对比结果
|
||
print(f"K线数据: {case['data'][1]}")
|
||
print(f"\nKLU算法结果:")
|
||
print(f" 分型类型: {klu_fx_type}")
|
||
print(f" 分型强度: {klu_strength}")
|
||
print(f" 是否确认: {klu_confirmed}")
|
||
|
||
print(f"\nKLC算法结果:")
|
||
print(f" 分型类型: {klc_fx_type}")
|
||
print(f" 分型强度(原始): {klc_strength_raw}")
|
||
print(f" 分型强度(转换): {klc_strength_converted}")
|
||
|
||
# 一致性检查
|
||
type_consistent = (klu_fx_type == klc_fx_type)
|
||
strength_diff = abs(klu_strength - klc_strength_converted)
|
||
strength_consistent = strength_diff <= 10 # 允许10分以内的差异
|
||
|
||
print(f"\n一致性检查:")
|
||
print(f" 分型类型一致: {'✅' if type_consistent else '❌'}")
|
||
print(f" 强度差异: {strength_diff}分 {'✅' if strength_consistent else '❌'}")
|
||
|
||
if not type_consistent or not strength_consistent:
|
||
print(f" ⚠️ 算法结果不一致!")
|
||
else:
|
||
print(f" ✅ 算法结果一致")
|
||
else:
|
||
print("❌ 数据不足，无法进行对比")
|
||
|
||
def detailed_strength_analysis():
|
||
"""详细的强度分析对比"""
|
||
print("\n" + "=" * 80)
|
||
print("详细强度分析对比")
|
||
print("=" * 80)
|
||
|
||
# 创建一个明确的强分型案例
|
||
strong_top_data = [
|
||
{'open': 100, 'high': 101, 'low': 99, 'close': 100, 'volume': 1000},
|
||
{'open': 100, 'high': 110, 'low': 99, 'close': 102, 'volume': 3000}, # 强顶分型
|
||
{'open': 102, 'high': 103, 'low': 92, 'close': 93, 'volume': 2000}, # 强确认
|
||
{'open': 93, 'high': 94, 'low': 90, 'close': 91, 'volume': 1500}, # 继续下跌
|
||
{'open': 91, 'high': 92, 'low': 88, 'close': 89, 'volume': 1200}, # 进一步确认
|
||
]
|
||
|
||
klus = setup_klu_chain(strong_top_data)
|
||
|
||
if len(klus) >= 5:
|
||
target_klu = klus[1] # 目标分型K线
|
||
|
||
print(f"分析目标: 第2根K线 (索引1)")
|
||
print(f"K线数据: {strong_top_data[1]}")
|
||
|
||
# 更新分析
|
||
target_klu.update_realtime_analysis()
|
||
|
||
print(f"\n分型检测结果:")
|
||
print(f" 分型类型: {target_klu.fx_type}")
|
||
print(f" 分型确认: {target_klu.fx_confirmed}")
|
||
print(f" 最终强度: {target_klu.fx_strength}")
|
||
|
||
# 显示中间计算过程（需要重新调用以获取详细信息）
|
||
if target_klu.fx_confirmed:
|
||
print(f"\n强度计算过程:")
|
||
is_bi_end = target_klu._check_if_bi_ending_fx()
|
||
post_confirmation = target_klu._check_post_fx_confirmation()
|
||
fx_quality = target_klu._check_fx_quality()
|
||
|
||
print(f" 笔终结判断: {is_bi_end}")
|
||
print(f" 后续确认: {post_confirmation}")
|
||
print(f" 分型质量: {fx_quality}")
|
||
|
||
raw_score = is_bi_end + post_confirmation + fx_quality
|
||
final_raw = max(-3, min(3, raw_score))
|
||
converted_score = int((final_raw + 3) * 100 / 6)
|
||
|
||
print(f" 原始总分: {raw_score} -> {final_raw}")
|
||
print(f" 转换分数: {converted_score}")
|
||
|
||
if __name__ == "__main__":
|
||
# 运行测试
|
||
test_cases = create_test_data()
|
||
compare_algorithms(test_cases)
|
||
|
||
# 详细分析
|
||
detailed_strength_analysis()
|
||
|
||
print("\n" + "=" * 80)
|
||
print("测试完成！")
|
||
print("=" * 80) |