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

最小化去除iloc[-1], 重构detect_h1_rapid_rise

Browse Source
This commit is contained in:
zhangkun9038@dingtalk.com 2025-09-07 22:27:26 +08:00
parent afcf7e51dc
commit 0b256e31d9
1 changed files with 46 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

119
freqtrade/templates/freqaiprimer.py
View File

@ -263,6 +263,11 @@ class FreqaiPrimer(IStrategy):
dataframe.loc[(dataframe['market_score'] >= 30) & (dataframe['market_score'] <= 50), 'market_state'] = 'neutral'
dataframe.loc[(dataframe['market_score'] > 10) & (dataframe['market_score'] < 30), 'market_state'] = 'weak_bear'
dataframe.loc[dataframe['market_score'] <= 10, 'market_state'] = 'strong_bear'
# 创建一个使用前一行市场状态的列，避免在populate_entry_trend中使用iloc[-1]
dataframe['prev_market_state'] = dataframe['market_state'].shift(1)
# 为第一行设置默认值
dataframe['prev_market_state'].fillna('neutral', inplace=True)
# 记录当前的市场状态
if len(dataframe) > 0:
@ -312,54 +317,6 @@ class FreqaiPrimer(IStrategy):
logger.info(f"[{metadata['pair']}] 发现入场信号数量: {dataframe['enter_long'].sum()}")
return dataframe
# 做多条件（放宽以增加入场信号）
# 1. 价格接近3m布林带下轨（放宽到5%偏差）
close_to_bb_lower = (dataframe['close'] <= dataframe['bb_lower_3m'] * 1.05)
# 2. 至少一个时间框架RSI处于超卖区域
rsi_oversold = (dataframe['rsi_3m'] < self.rsi_oversold) | \
(dataframe['rsi_15m'] < self.rsi_oversold)
# 3. 1小时趋势向上或横盘
trend_1h = dataframe['trend_1h'] | (dataframe['market_state'] == 'neutral')
# 4. 成交量至少达到平均水平的80%
volume_condition = dataframe['volume'] > dataframe['volume_ma'] * 0.8
# 5. 看涨吞没形态或RSI接近超卖
special_condition = dataframe['bullish_engulfing'] | (dataframe['rsi_3m'] < self.rsi_oversold + 3)
# 检查剧烈拉升情况 - 如果检测到剧烈拉升，则不产生入场信号
pair = metadata['pair']
is_unstable_region, _ = self.detect_h1_rapid_rise(pair, dataframe, metadata)
# 熊牛得分检查：当得分低于55时禁止入场（放宽门槛）
market_score_condition = dataframe['market_score'] >= 55
# 合并所有条件，并且确保不在不稳固区域且市场状态良好
final_condition = close_to_bb_lower & rsi_oversold & trend_1h & volume_condition & special_condition & (~is_unstable_region) & market_score_condition
# 所有原始条件（不包括剧烈拉升检查和熊牛得分检查）
original_conditions = close_to_bb_lower & rsi_oversold & trend_1h & volume_condition & special_condition & (~is_unstable_region)
dataframe.loc[final_condition, 'enter_long'] = 1
# 记录入场信号被阻止的情况
prevented_entries = dataframe[(~final_condition) & original_conditions]
# 统计因熊牛得分不足而被阻止的信号
low_score_prevented = dataframe[(~market_score_condition) & original_conditions]
if len(low_score_prevented) > 0:
logger.info(f"[{pair}] 由于熊牛得分低于60，阻止了 {len(low_score_prevented)} 个潜在的入场信号")
if len(prevented_entries) > 0:
logger.info(f"[{pair}] 总共阻止了 {len(prevented_entries)} 个潜在的入场信号")
# 调试：检查每个条件的触发情况（简化输出）
if dataframe['enter_long'].sum() > 0:
logger.info(f"[{pair}] 发现入场信号数量: {dataframe['enter_long'].sum()}")
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# 出场信号基于趋势和量价关系
@ -388,8 +345,8 @@ class FreqaiPrimer(IStrategy):
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# 获取当前市场状态
current_state = dataframe['market_state'].iloc[-1] if 'market_state' in dataframe.columns else 'neutral'
# 使用前一行的市场状态，避免未来数据泄露
current_state = dataframe['prev_market_state'] if 'prev_market_state' in dataframe.columns else 'neutral'
# 条件1: 价格接近布林带下轨（允许一定偏差）
close_to_bb_lower_1h = (dataframe['close'] <= dataframe['bb_lower_1h'] * 1.03) # 放宽到3%偏差
@ -446,15 +403,13 @@ class FreqaiPrimer(IStrategy):
return dataframe
def detect_h1_rapid_rise(self, pair: str, dataframe: DataFrame, metadata: dict) -> tuple[bool, float]:
def detect_h1_rapid_rise(self, pair: str) -> bool:
"""
检测1小时K线图上的剧烈拉升情况
检测1小时K线图上的剧烈拉升情况（轻量级版本，用于confirm_trade_entry）
参数:
- pair: 交易对
- dataframe: 数据框
- metadata: 元数据
返回:
- tuple: (是否处于不稳固区域, 当前价格在最近价格范围中的百分比)
- bool: 是否处于不稳固区域
"""
try:
# 获取1小时K线数据
@ -463,11 +418,10 @@ class FreqaiPrimer(IStrategy):
# 确保有足够的K线数据
if len(df_1h) < self.H1_MAX_CANDLES:
logger.warning(f"[{pair}] 1h K线数据不足 {self.H1_MAX_CANDLES} 根，当前只有 {len(df_1h)} 根，无法完整检测剧烈拉升")
return False, 0.0
return False
# 获取最近的K线
recent_data = df_1h.iloc[-self.H1_MAX_CANDLES:].copy()
current_price = recent_data['close'].iloc[-1]
# 检查连续最多3根K线内的最大涨幅
rapid_rise_detected = False
@ -490,26 +444,45 @@ class FreqaiPrimer(IStrategy):
logger.info(f"[{pair}] 检测到剧烈拉升: 从 {window_low:.2f} 到 {window_high:.2f} ({rise_percentage:.2%}) 在 {self.H1_MAX_CONSECUTIVE_CANDLES} 根K线内")
break
# 计算当前价格在最近价格范围中的位置百分比
recent_low = recent_data['low'].min()
recent_high = recent_data['high'].max()
current_price = recent_data['close'].iloc[-1]
logger.info(f"[{pair}] 剧烈拉升检测结果: {'不稳固' if rapid_rise_detected else '稳固'}")
logger.info(f"[{pair}] 最近最大涨幅: {max_rise:.2%}")
if recent_high > recent_low:
price_position_pct = (current_price - recent_low) / (recent_high - recent_low) * 100
else:
price_position_pct = 50.0 # 默认中间位置
# 判断是否处于不稳固区域
is_unstable_region = rapid_rise_detected
logger.info(f"[{pair}] 剧烈拉升检测结果: {'不稳固' if is_unstable_region else '稳固'}")
logger.info(f"[{pair}] 当前价格位置: {current_price:.2f} ({price_position_pct:.1f}%), 最近最大涨幅: {max_rise:.2%}")
return is_unstable_region, price_position_pct
return rapid_rise_detected
except Exception as e:
logger.error(f"[{pair}] 剧烈拉升检测过程中发生错误: {str(e)}")
return False, 0.0
return False
def confirm_trade_entry(
self,
pair: str,
order_type: str,
amount: float,
rate: float,
time_in_force: str,
current_time: datetime,
entry_tag: str | None,
side: str,
**kwargs,
) -> bool:
"""
交易买入前的确认函数，用于最终决定是否执行交易
此处实现剧烈拉升检查逻辑
"""
# 默认允许交易
allow_trade = True
# 仅对多头交易进行检查
if side == 'long':
# 检查是否处于剧烈拉升的不稳固区域
is_unstable_region = self.detect_h1_rapid_rise(pair)
if is_unstable_region:
logger.info(f"[{pair}] 由于检测到剧烈拉升，取消入场交易")
allow_trade = False
# 如果没有阻止因素，允许交易
return allow_trade
def custom_stoploss(self, pair: str, trade: 'Trade', current_time, current_rate: float,
current_profit: float, **kwargs) -> float: