phyer/myTestFreqAI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加1小时EMA5向上穿越EMA20的入场限制条件

Browse Source
This commit is contained in:
phyer 2026-01-14 01:34:25 +08:00
parent ec0555773b
commit 9e07b8913d
2 changed files with 159 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
freqtrade/templates/freqaiprimer.py
View File

@ -544,6 +544,16 @@ class FreqaiPrimer(IStrategy):
df_1h['bb_lower_1h'] = bb_ma_1h - (bb_std_value * bb_std_1h)
df_1h['bb_upper_1h'] = bb_ma_1h + (bb_std_value * bb_std_1h)
# 添加 EMA5 和 EMA20 用于入场过滤
df_1h['ema_5_1h'] = df_1h['close'].ewm(span=5, adjust=False).mean()
df_1h['ema_20_1h'] = df_1h['close'].ewm(span=20, adjust=False).mean()
# 检测 EMA5 向上穿越 EMA20
df_1h['ema5_cross_above_ema20'] = (
(df_1h['ema_5_1h'] > df_1h['ema_20_1h']) &
(df_1h['ema_5_1h'].shift(1) <= df_1h['ema_20_1h'].shift(1))
)
# 使用 rolling 计算 RSI（减少看前偏差）
delta_1h = df_1h['close'].diff()
gain_1h = delta_1h.where(delta_1h > 0, 0).rolling(window=rsi_length_value).mean()
@ -575,12 +585,12 @@ class FreqaiPrimer(IStrategy):
df_1h = df_1h.set_index('date').reindex(dataframe['date']).ffill().bfill().reset_index()
df_1h = df_1h.rename(columns={'index': 'date'})
# Include macd_1h and macd_signal_1h in the column selection
df_1h = df_1h[['date', 'rsi_1h', 'trend_1h', 'ema_50_1h', 'ema_200_1h', 'bb_lower_1h', 'bb_upper_1h', 'stochrsi_k_1h', 'stochrsi_d_1h', 'macd_1h', 'macd_signal_1h']].ffill()
df_1h = df_1h[['date', 'rsi_1h', 'trend_1h', 'ema_50_1h', 'ema_200_1h', 'bb_lower_1h', 'bb_upper_1h', 'stochrsi_k_1h', 'stochrsi_d_1h', 'macd_1h', 'macd_signal_1h', 'ema_5_1h', 'ema_20_1h', 'ema5_cross_above_ema20']].ffill()
# Validate that all required columns are present
required_columns = ['date', 'rsi_1h', 'trend_1h', 'ema_50_1h', 'ema_200_1h',
'bb_lower_1h', 'bb_upper_1h', 'stochrsi_k_1h', 'stochrsi_d_1h',
'macd_1h', 'macd_signal_1h']
'macd_1h', 'macd_signal_1h', 'ema_5_1h', 'ema_20_1h', 'ema5_cross_above_ema20']
missing_columns = [col for col in required_columns if col not in df_1h.columns]
if missing_columns:
logger.error(f"[{metadata['pair']}] 缺少以下列: {missing_columns}")
@ -597,7 +607,7 @@ class FreqaiPrimer(IStrategy):
logger.error(f"[{metadata['pair']}] 缺少以下列: {missing_columns}")
raise KeyError(f"缺少以下列: {missing_columns}")
df_1h = df_1h[required_columns] # 确保包含 macd_1h 和 macd_signal_1h
df_1h = df_1h[required_columns] # 确保包含所有必需的列
# 合并 1h 数据
dataframe = dataframe.merge(df_1h, how='left', on='date').ffill()
@ -744,8 +754,11 @@ class FreqaiPrimer(IStrategy):
# 辅助条件: 3m 和 15m 趋势确认（允许部分时间框架不一致）
trend_confirmation = (dataframe['trend_3m'] == 1) | (dataframe['trend_15m'] == 1)
# 新增：EMA5向上穿越EMA20过滤条件（核心入场限制）
ema5_cross_filter = dataframe['ema5_cross_above_ema20'] == 1
# 合并所有条件（减少强制性条件）
# 至少满足多个条件中的一定数量
# 至少满足多个条件中的一定数量，并且必须满足EMA5向上穿越EMA20
condition_count = (
close_to_bb_lower_1h.astype(int) +
rsi_condition_1h.astype(int) +
@ -754,7 +767,9 @@ class FreqaiPrimer(IStrategy):
(volume_spike | bb_width_condition).astype(int) + # 成交量或布林带宽度满足其一即可
trend_confirmation.astype(int)
)
final_condition = condition_count >= self.min_condition_count.value
# 修改最终条件：必须同时满足最小条件数量和EMA5向上穿越EMA20
basic_condition = condition_count >= self.min_condition_count.value
final_condition = basic_condition & ema5_cross_filter
# 设置入场信号
dataframe.loc[final_condition, 'enter_long'] = 1
@ -828,14 +843,24 @@ class FreqaiPrimer(IStrategy):
dataframe.loc[final_condition_updated, 'enter_price'] = dataframe.loc[final_condition_updated, 'close'] * 0.9833
# 增强调试信息
#self.strategy_log(f"[{metadata['pair']}] 入场条件检查:")
#self.strategy_log(f" - 价格接近布林带下轨: {close_to_bb_lower_1h.sum()} 次")
#self.strategy_log(f" - RSI 超卖: {rsi_condition_1h.sum()} 次")
#self.strategy_log(f" - StochRSI 超卖: {stochrsi_condition_1h.sum()} 次")
#self.strategy_log(f" - MACD 上升趋势: {macd_condition_1h.sum()} 次")
#self.strategy_log(f" - 成交量或布林带宽度: {(volume_spike | bb_width_condition).sum()} 次")
#self.strategy_log(f" - 趋势确认: {trend_confirmation.sum()} 次")
#self.strategy_log(f" - 最终条件: {final_condition.sum()} 次")
ema5_cross_count = ema5_cross_filter.sum()
basic_condition_count = basic_condition.sum()
final_condition_count = final_condition.sum()
self.strategy_log(f"[{metadata['pair']}] 入场条件检查:")
self.strategy_log(f" - 价格接近布林带下轨: {close_to_bb_lower_1h.sum()} 次")
self.strategy_log(f" - RSI 超卖: {rsi_condition_1h.sum()} 次")
self.strategy_log(f" - StochRSI 超卖: {stochrsi_condition_1h.sum()} 次")
self.strategy_log(f" - MACD 上升趋势: {macd_condition_1h.sum()} 次")
self.strategy_log(f" - 成交量或布林带宽度: {(volume_spike | bb_width_condition).sum()} 次")
self.strategy_log(f" - 趋势确认: {trend_confirmation.sum()} 次")
self.strategy_log(f" - EMA5向上穿越EMA20: {ema5_cross_count} 次")
self.strategy_log(f" - 基本条件满足: {basic_condition_count} 次")
self.strategy_log(f" - 最终条件(基本+EMA穿越): {final_condition_count} 次")
# 如果有EMA穿越但最终条件未满足，输出详细信息
if ema5_cross_count > 0 and final_condition_count == 0:
self.strategy_log(f"[{metadata['pair']}] 注意：检测到 {ema5_cross_count} 次EMA5向上穿越，但由于其他条件不足未能生成入场信号")
# 在populate_entry_trend方法末尾添加
# 计算条件间的相关性
conditions = DataFrame({

121
test_ema_filter.py Normal file
View File

@ -0,0 +1,121 @@
#!/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测试完成！")