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

freqtrade/templates/freqaiprimer.py

@@ -554,11 +554,16 @@ class FreqaiPrimer(IStrategy):
         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))
-        )
+        # 检测 EMA5 向上穿越 EMA20（添加安全检查）
+        # 确保有足够的数据进行shift操作
+        if len(df_1h) >= 2:
+            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))
+            )
+        else:
+            # 数据不足时，默认为False
+            df_1h['ema5_cross_above_ema20'] = False
         
         # 使用 rolling 计算 RSI（减少看前偏差）
         delta_1h = df_1h['close'].diff()
@@ -761,7 +766,13 @@ class FreqaiPrimer(IStrategy):
         trend_confirmation = (dataframe['trend_3m'] == 1) | (dataframe['trend_15m'] == 1)
         
         # 新增：EMA5向上穿越EMA20过滤条件（核心入场限制）
-        ema5_cross_filter = dataframe['ema5_cross_above_ema20'] == 1
+        # 添加防御性检查，防止列不存在
+        if 'ema5_cross_above_ema20' in dataframe.columns:
+            ema5_cross_filter = dataframe['ema5_cross_above_ema20'] == 1
+        else:
+            # 如果列不存在，默认为False（不允许入场）
+            self.strategy_log(f"[{metadata['pair']}] 警告：ema5_cross_above_ema20列不存在，过滤条件设为False")
+            ema5_cross_filter = False
         
         # 合并所有条件（减少强制性条件）
         # 至少满足多个条件中的一定数量，并且必须满足EMA5向上穿越EMA20

test_ema_safety.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""
+测试EMA交叉过滤条件的安全性修复
+"""
+
+import pandas as pd
+import numpy as np
+
+def test_ema_cross_safety():
+    """测试EMA交叉检测的安全性"""
+    print("=== 测试EMA交叉安全性 ===")
+    
+    # 测试1：正常数据
+    print("测试1：正常数据（100根K线）")
+    dates = pd.date_range(start='2024-01-01', periods=100, freq='1H')
+    prices = np.random.random(100) * 100 + 50
+    
+    df_normal = pd.DataFrame({
+        'date': dates,
+        'close': prices
+    })
+    
+    # 计算EMA
+    df_normal['ema_5'] = df_normal['close'].ewm(span=5, adjust=False).mean()
+    df_normal['ema_20'] = df_normal['close'].ewm(span=20, adjust=False).mean()
+    
+    # 安全的交叉检测
+    if len(df_normal) >= 2:
+        df_normal['ema5_cross_above_ema20'] = (
+            (df_normal['ema_5'] > df_normal['ema_20']) & 
+            (df_normal['ema_5'].shift(1) <= df_normal['ema_20'].shift(1))
+        )
+    else:
+        df_normal['ema5_cross_above_ema20'] = False
+    
+    cross_count = df_normal['ema5_cross_above_ema20'].sum()
+    print(f"  正常数据交叉次数: {cross_count}")
+    
+    # 测试2：数据不足
+    print("测试2：数据不足（只有1根K线）")
+    df_insufficient = pd.DataFrame({
+        'date': [pd.Timestamp('2024-01-01')],
+        'close': [100.0]
+    })
+    
+    df_insufficient['ema_5'] = df_insufficient['close'].ewm(span=5, adjust=False).mean()
+    df_insufficient['ema_20'] = df_insufficient['close'].ewm(span=20, adjust=False).mean()
+    
+    # 应该触发安全检查
+    if len(df_insufficient) >= 2:
+        df_insufficient['ema5_cross_above_ema20'] = (
+            (df_insufficient['ema_5'] > df_insufficient['ema_20']) & 
+            (df_insufficient['ema_5'].shift(1) <= df_insufficient['ema_20'].shift(1))
+        )
+    else:
+        df_insufficient['ema5_cross_above_ema20'] = False
+    
+    print(f"  数据不足时的值: {df_insufficient['ema5_cross_above_ema20'].iloc[0]}")
+    
+    # 测试3：边界情况（刚好2根K线）
+    print("测试3：边界情况（2根K线）")
+    df_boundary = pd.DataFrame({
+        'date': pd.date_range(start='2024-01-01', periods=2, freq='1H'),
+        'close': [100.0, 105.0]
+    })
+    
+    df_boundary['ema_5'] = df_boundary['close'].ewm(span=5, adjust=False).mean()
+    df_boundary['ema_20'] = df_boundary['close'].ewm(span=20, adjust=False).mean()
+    
+    if len(df_boundary) >= 2:
+        df_boundary['ema5_cross_above_ema20'] = (
+            (df_boundary['ema_5'] > df_boundary['ema_20']) & 
+            (df_boundary['ema_5'].shift(1) <= df_boundary['ema_20'].shift(1))
+        )
+    else:
+        df_boundary['ema5_cross_above_ema20'] = False
+    
+    print(f"  边界情况的值: {df_boundary['ema5_cross_above_ema20'].tolist()}")
+    
+    print("✓ 所有安全性测试通过！")
+
+if __name__ == "__main__":
+    test_ema_cross_safety()