121 lines
4.0 KiB
Python
121 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试EMA5向上穿越EMA20入场过滤条件的脚本
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from freqtrade.templates.freqaiprimer import FreqaiPrimer
|
||
import pandas as pd
|
||
import numpy as np
|
||
from datetime import datetime, timedelta
|
||
|
||
def create_test_dataframe():
|
||
"""创建测试用的模拟数据"""
|
||
# 创建时间序列
|
||
dates = pd.date_range(start='2024-01-01', periods=100, freq='1H')
|
||
|
||
# 创建价格数据（模拟趋势）
|
||
np.random.seed(42)
|
||
prices = []
|
||
current_price = 100.0
|
||
|
||
for i in range(100):
|
||
# 模拟EMA5向上穿越EMA20的情况
|
||
if i < 30:
|
||
# 前30个周期：下跌趋势
|
||
change = np.random.normal(-0.005, 0.01)
|
||
elif i < 50:
|
||
# 30-50周期：横盘整理
|
||
change = np.random.normal(0, 0.005)
|
||
else:
|
||
# 50-100周期：上涨趋势（EMA5应该向上穿越EMA20）
|
||
change = np.random.normal(0.01, 0.015)
|
||
|
||
current_price = current_price * (1 + change)
|
||
prices.append(max(current_price, 1)) # 确保价格为正数
|
||
|
||
# 创建DataFrame
|
||
df = pd.DataFrame({
|
||
'date': dates,
|
||
'open': prices,
|
||
'high': [p * (1 + abs(np.random.normal(0, 0.01))) for p in prices],
|
||
'low': [p * (1 - abs(np.random.normal(0, 0.01))) for p in prices],
|
||
'close': prices,
|
||
'volume': [np.random.uniform(1000, 5000) for _ in range(100)]
|
||
})
|
||
|
||
return df
|
||
|
||
def test_ema_calculation():
|
||
"""测试EMA计算和交叉检测"""
|
||
print("=== 测试EMA5和EMA20计算 ===")
|
||
|
||
# 创建测试数据
|
||
df = create_test_dataframe()
|
||
|
||
# 计算EMA
|
||
df['ema_5'] = df['close'].ewm(span=5, adjust=False).mean()
|
||
df['ema_20'] = df['close'].ewm(span=20, adjust=False).mean()
|
||
|
||
# 检测交叉
|
||
df['ema5_cross_above_ema20'] = (
|
||
(df['ema_5'] > df['ema_20']) &
|
||
(df['ema_5'].shift(1) <= df['ema_20'].shift(1))
|
||
)
|
||
|
||
# 显示关键点的数据
|
||
cross_points = df[df['ema5_cross_above_ema20'] == True]
|
||
print(f"检测到 {len(cross_points)} 次EMA5向上穿越EMA20")
|
||
|
||
if len(cross_points) > 0:
|
||
print("\n交叉点详情:")
|
||
for idx, row in cross_points.iterrows():
|
||
print(f" 时间: {row['date']}, 价格: {row['close']:.2f}, "
|
||
f"EMA5: {row['ema_5']:.2f}, EMA20: {row['ema_20']:.2f}")
|
||
|
||
# 验证交叉逻辑
|
||
print("\n=== 验证交叉检测逻辑 ===")
|
||
for i in range(1, min(10, len(df))):
|
||
current_ema5 = df['ema_5'].iloc[i]
|
||
current_ema20 = df['ema_20'].iloc[i]
|
||
prev_ema5 = df['ema_5'].iloc[i-1]
|
||
prev_ema20 = df['ema_20'].iloc[i-1]
|
||
cross_flag = df['ema5_cross_above_ema20'].iloc[i]
|
||
|
||
# 手动验证交叉条件
|
||
manual_cross = (current_ema5 > current_ema20) and (prev_ema5 <= prev_ema20)
|
||
|
||
if cross_flag or manual_cross:
|
||
status = "✓" if cross_flag == manual_cross else "✗"
|
||
print(f"{status} 索引{i}: 当前({current_ema5:.3f}>{current_ema20:.3f}), "
|
||
f"前一个({prev_ema5:.3f}<={prev_ema20:.3f}), 检测结果: {cross_flag}")
|
||
|
||
def test_strategy_integration():
|
||
"""测试策略集成"""
|
||
print("\n=== 测试策略集成 ===")
|
||
|
||
try:
|
||
# 创建策略实例
|
||
strategy = FreqaiPrimer(config={})
|
||
print("✓ 策略实例创建成功")
|
||
|
||
# 检查是否有必要的属性
|
||
required_attrs = ['ema5_cross_above_ema20']
|
||
print("策略属性检查:")
|
||
for attr in required_attrs:
|
||
if hasattr(strategy, attr):
|
||
print(f" ✓ {attr}")
|
||
else:
|
||
print(f" ✗ {attr}")
|
||
|
||
except Exception as e:
|
||
print(f"✗ 策略集成测试失败: {e}")
|
||
|
||
if __name__ == "__main__":
|
||
print("开始测试EMA5向上穿越EMA20过滤条件...")
|
||
test_ema_calculation()
|
||
test_strategy_integration()
|
||
print("\n测试完成！") |