add fx strentth

2025-05-23 21:19:50 +08:00
parent a1e033432b
commit a9d24233b0
14 changed files with 2067 additions and 399 deletions
@@ -0,0 +1,214 @@
+#!/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()