214 lines
7.7 KiB
Python
214 lines
7.7 KiB
Python
#!/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() |