#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 测试分型强度检测功能 """ from ChanKLC import ChanKLC import ChanKLU from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR import requests import json import numpy as np import matplotlib.pyplot as plt from collections import Counter def test_fx_strength(): """测试分型强度检测功能""" print('=== 分型强度检测功能测试 ===') # 创建一个简单的测试KLU，使用正确的构造函数参数 klu = ChanKLU.ChanKLU( time='2024-01-01 10:00:00', open=100.0, high=105.0, low=98.0, close=103.0, volume=1000 ) klu.rsi = 65.0 klu.volume_ratio = 1.2 klu.macdhist = 0.5 # 创建KLC对象 klc = ChanKLC(klu, 1, Chan_KLINE_DIR.UP) klc.fx = Chan_FX_TYPE.TOP # 测试强度计算 strength = klc.calculate_fx_strength() level = klc.get_fx_strength_level() is_strong = klc.is_strong_fx() print(f'分型强度分数: {strength}') print(f'分型强度等级: {level}') print(f'是否强分型: {is_strong}') # 测试特征数据集成 features = klc.get_feature_data() fx_features = {k: v for k, v in features.items() if 'fx_strength' in k} print('\n分型强度相关特征:') for key, value in fx_features.items(): print(f' {key}: {value}') print('\n✅ 分型强度检测功能正常工作!') return True def test_fx_strength_distribution(): """测试分型强度分布情况""" print("=== 分型强度分布分析 ===") # 请求API数据 url = "http://localhost:8123/api/analyze" params = { 'symbol': 'SOL/USDT:USDT', 'timeframe': '5m', 'timezone': 'Asia/Shanghai' } try: response = requests.get(url, params=params) response.raise_for_status() data = response.json() except Exception as e: print(f"❌ 请求API失败: {e}") return # 提取分型强度数据 fx_strengths = [] fx_levels = [] top_strengths = [] bottom_strengths = [] for fx in data.get('klc_fx_info', []): strength = fx.get('fx_strength', 0) level = fx.get('fx_strength_level', 'Unknown') is_bottom = fx.get('is_bottom_fx', False) fx_strengths.append(strength) fx_levels.append(level) if is_bottom: bottom_strengths.append(strength) else: top_strengths.append(strength) # 统计分析 if fx_strengths: print(f"\n📊 基础统计:") print(f"总分型数量: {len(fx_strengths)}") print(f"平均强度: {np.mean(fx_strengths):.2f}") print(f"强度中位数: {np.median(fx_strengths):.2f}") print(f"强度标准差: {np.std(fx_strengths):.2f}") print(f"最高强度: {np.max(fx_strengths):.2f}") print(f"最低强度: {np.min(fx_strengths):.2f}") print(f"\n🔝 顶分型统计:") if top_strengths: print(f"数量: {len(top_strengths)}") print(f"平均强度: {np.mean(top_strengths):.2f}") print(f"最高强度: {np.max(top_strengths):.2f}") print(f"\n🔻 底分型统计:") if bottom_strengths: print(f"数量: {len(bottom_strengths)}") print(f"平均强度: {np.mean(bottom_strengths):.2f}") print(f"最高强度: {np.max(bottom_strengths):.2f}") # 强度等级分布 print(f"\n📈 强度等级分布:") level_counts = Counter(fx_levels) for level, count in level_counts.items(): percentage = (count / len(fx_levels)) * 100 print(f"{level}: {count} ({percentage:.1f}%)") # 强度区间分布 print(f"\n📊 强度区间分布:") ranges = [ (0, 20, "极弱 (0-20)"), (20, 40, "弱 (20-40)"), (40, 60, "中等 (40-60)"), (60, 80, "强 (60-80)"), (80, 100, "极强 (80-100)") ] for min_val, max_val, label in ranges: count = sum(1 for s in fx_strengths if min_val <= s < max_val) percentage = (count / len(fx_strengths)) * 100 print(f"{label}: {count} ({percentage:.1f}%)") # 找出最强和最弱的分型 print(f"\n⭐ 最强分型 (Top 5):") sorted_fx = sorted(data.get('klc_fx_info', []), key=lambda x: x.get('fx_strength', 0), reverse=True)[:5] for i, fx in enumerate(sorted_fx, 1): fx_type = "底分型" if fx.get('is_bottom_fx', False) else "顶分型" print(f" {i}. {fx.get('time', 'N/A')} - {fx_type} - 强度: {fx.get('fx_strength', 0):.2f} - 等级: {fx.get('fx_strength_level', 'N/A')}") print(f"\n💔 最弱分型 (Bottom 5):") weakest_fx = sorted(data.get('klc_fx_info', []), key=lambda x: x.get('fx_strength', 0))[:5] for i, fx in enumerate(weakest_fx, 1): fx_type = "底分型" if fx.get('is_bottom_fx', False) else "顶分型" print(f" {i}. {fx.get('time', 'N/A')} - {fx_type} - 强度: {fx.get('fx_strength', 0):.2f} - 等级: {fx.get('fx_strength_level', 'N/A')}") # 生成直方图 try: plt.figure(figsize=(12, 8)) # 主强度分布图 plt.subplot(2, 2, 1) plt.hist(fx_strengths, bins=20, alpha=0.7, color='blue', edgecolor='black') plt.title('分型强度分布') plt.xlabel('强度分数') plt.ylabel('频次') plt.axvline(np.mean(fx_strengths), color='red', linestyle='--', label=f'平均值: {np.mean(fx_strengths):.2f}') plt.legend() # 顶分型 vs 底分型对比 plt.subplot(2, 2, 2) if top_strengths and bottom_strengths: plt.hist([top_strengths, bottom_strengths], bins=15, alpha=0.7, label=['顶分型', '底分型'], color=['red', 'green']) plt.title('顶分型 vs 底分型强度对比') plt.xlabel('强度分数') plt.ylabel('频次') plt.legend() # 强度等级饼图 plt.subplot(2, 2, 3) if level_counts: labels = list(level_counts.keys()) sizes = list(level_counts.values()) colors = ['red', 'orange', 'yellow', 'lightgreen', 'green'][:len(labels)] plt.pie(sizes, labels=labels, colors=colors, autopct='%1.1f%%') plt.title('强度等级分布') # 时间序列图 plt.subplot(2, 2, 4) x_vals = range(len(fx_strengths)) colors = ['red' if not fx.get('is_bottom_fx', False) else 'green' for fx in data.get('klc_fx_info', [])] plt.scatter(x_vals, fx_strengths, c=colors, alpha=0.6) plt.title('分型强度时间序列 (红=顶分型, 绿=底分型)') plt.xlabel('分型序号') plt.ylabel('强度分数') plt.tight_layout() plt.savefig('user_data/Chan/fx_strength_analysis.png', dpi=300, bbox_inches='tight') print(f"\n📈 图表已保存到: user_data/Chan/fx_strength_analysis.png") except ImportError: print("\n📈 matplotlib 未安装，跳过图表生成") except Exception as e: print(f"\n❌ 生成图表失败: {e}") else: print("❌ 未找到分型强度数据") if __name__ == "__main__": test_fx_strength() test_fx_strength_distribution()