#!/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)