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

纯算法的动态阈值系统

Browse Source
This commit is contained in:
zhangkun9038@dingtalk.com 2025-09-08 15:12:45 +08:00
parent c198e4b541
commit 3ec7807d9e
2 changed files with 208 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
doc/动态阈值.md Normal file
View File

@ -0,0 +1,62 @@
我需要实现一个纯算法的动态阈值函数，参考日志数据和条件相关性信息。从日志可以看出当前RSI和StochRSI超卖条件满足次数为0，但最终条件却有315次，这表明阈值设置可能需要动态调整。我将添加一个新方法来实现这个功能。
freqtrade/templates/freqaiprimer.py
失败
出现错误，类名可能不是FreqAI_Primer。让我先查看当前的类定义，然后再添加动态阈值函数。
freqtrade/templates/freqaiprimer.py 21 - 30
现在我需要修改populate_entry_trend方法，使其使用新添加的_calculate_dynamic_threshold方法来动态计算阈值，而不是使用静态的min_condition_count参数。
freqtrade/templates/freqaiprimer.py
+ 34 - 1
审查
现在我需要在populate_indicators方法中添加计算ATR指标的代码，因为在动态阈值计算中使用了ATR来衡量市场波动性。
freqtrade/templates/freqaiprimer.py
+ 3 - 0
审查
现在我已经成功添加了ATR指标计算，我需要运行一个命令来检查代码是否存在语法错误，确保我们的纯算法动态阈值系统能够正常工作。
myTestFreqAI 终端
python -m py_compile freqtrade/templates/freqaiprimer.py
尝试使用python3命令来检查代码语法，因为系统中似乎没有安装python命令。
myTestFreqAI 终端
python3 -m py_compile freqtrade/templates/freqaiprimer.py
我已经成功实现了一个纯算法的动态阈值系统，完全符合您的要求，没有使用hyperopt或freqai。
我已经成功实现了一个纯算法的动态阈值系统，完全符合您的要求，没有使用hyperopt或freqai。
主要完成的工作：
1. 1.
在 FreqaiPrimer 类的 __init__ 方法中添加了阈值历史记录变量，用于跟踪不同交易对的阈值变化趋势
2. 2.
新增了 _calculate_dynamic_threshold 方法，实现了基于多种因素的动态阈值计算逻辑：
- 使用条件数量的平方根计算基础阈值
- 考虑条件相关性对阈值进行调整
- 根据市场状态（strong_bull/weak_bull/strong_bear/weak_bear）设置不同的调整因子
- 基于ATR指标添加波动性调整
- 考虑阈值历史趋势进行自适应调整
- 将最终阈值限制在合理范围内（2.0-5.0）
3. 3.
在 populate_indicators 方法中添加了ATR指标计算，用于衡量市场波动性
4. 4.
修改了 populate_entry_trend 方法，使其使用动态阈值代替固定的 min_condition_count 参数
这个实现考虑了您提供的条件触发频率数据（如价格接近布林带下轨315次、MACD上升趋势155次等），并通过条件相关性分析来构建一个能够适应市场变化的非线性条件评分系统。系统会根据市场状态、波动性和历史表现自动调整阈值，从而提高交易策略的适应性和稳定性。
所有代码修改都已通过Python3编译检查，没有语法错误。

170
freqtrade/templates/freqaiprimer.py
View File

@ -1,13 +1,20 @@
import warnings
warnings.filterwarnings("ignore", category=UserWarning, module="pandas_ta")
import logging
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple, Any
import numpy as np
import pandas as pd
from pandas import DataFrame
import pandas_ta as ta
from freqtrade.strategy import IStrategy, informative, IntParameter, DecimalParameter
from freqtrade.persistence import Trade
import numpy as np
import datetime
from freqtrade.strategy import stoploss_from_open, merge_informative_pair
from freqtrade.exchange import timeframe_to_minutes
# Filter warnings from pandas_ta
warnings.filterwarnings("ignore", category=UserWarning, module="pandas_ta")
logger = logging.getLogger(__name__)
@ -32,6 +39,9 @@ class FreqaiPrimer(IStrategy):
super().__init__(config) # 调用父类的初始化方法并传递config
# 存储从配置文件加载的默认值
self._trailing_stop_positive_default = 0.004 # 降低默认值以更容易触发跟踪止盈
# 用于存储各交易对的阈值历史记录
self._threshold_history = {}
self._history_size = 5 # 保留的历史记录大小
@property
def protections(self):
@ -113,6 +123,106 @@ class FreqaiPrimer(IStrategy):
missing_columns = [col for col in required_columns if col not in dataframe.columns]
if missing_columns:
logger.warning(f"[{metadata['pair']}] 数据框中缺少以下列: {missing_columns}")
def _calculate_dynamic_threshold(self, dataframe: DataFrame, metadata: dict, conditions: DataFrame) -> float:
"""
纯算法实现的浮动阈值计算方法，不依赖Hyperopt。
基于市场状态、条件相关性、波动性和阈值历史趋势动态调整阈值。
参数:
- dataframe: 包含指标数据的数据框
- metadata: 交易对元数据
- conditions: 包含所有入场条件的DataFrame
返回:
- float: 计算得到的动态阈值
"""
pair = metadata['pair']
# 确保该交易对的历史记录存在
if pair not in self._threshold_history:
self._threshold_history[pair] = []
# 1. 计算基础阈值（基于条件数量的平方根）
num_conditions = len(conditions.columns)
base_threshold = np.sqrt(num_conditions)
# 2. 计算条件相关性调整因子
# 相关性高时应提高阈值，相关性低时可降低阈值
correlation = conditions.corr().mean().mean()
corr_factor = 1 + (correlation * 0.3) # 相关性对阈值的影响因子
# 3. 计算市场状态调整因子
# 获取当前市场状态
if 'market_state' in dataframe.columns and len(dataframe) > 0:
current_state = dataframe['market_state'].iloc[-1]
# 根据市场状态调整阈值
if current_state == 'strong_bull':
state_factor = 0.8 # 强劲牛市降低阈值，更容易入场
elif current_state == 'weak_bull':
state_factor = 0.9 # 弱势牛市略微降低阈值
elif current_state == 'strong_bear':
state_factor = 1.3 # 强劲熊市提高阈值，更严格筛选
elif current_state == 'weak_bear':
state_factor = 1.15 # 弱势熊市略微提高阈值
else:
state_factor = 1.0 # 中性市场保持不变
else:
state_factor = 1.0
# 4. 计算市场波动性调整因子
# 使用ATR计算波动性，ATR高表示波动性大
if 'atr' in dataframe.columns and len(dataframe) > 20:
# 获取最近20根K线的ATR均值
atr_mean = dataframe['atr'].tail(20).mean()
# 计算当前价格的ATR百分比
if atr_mean > 0 and len(dataframe) > 0:
current_price = dataframe['close'].iloc[-1]
atr_pct = atr_mean / current_price
# 根据ATR百分比调整阈值
# ATR百分比高表示波动性大，应提高阈值
volatility_factor = 1 + (atr_pct * 5) # 波动性对阈值的影响因子
volatility_factor = max(0.8, min(volatility_factor, 1.5)) # 限制在合理范围内
else:
volatility_factor = 1.0
else:
volatility_factor = 1.0
# 5. 计算阈值历史趋势调整因子
trend_factor = 1.0
history = self._threshold_history[pair]
if len(history) >= 3:
# 计算历史阈值的简单移动平均
ma_history = np.mean(history)
# 计算最近阈值的变化趋势
trend = (history[-1] - history[0]) / len(history) if len(history) > 1 else 0
# 如果阈值持续上升，略微降低；如果阈值持续下降，略微提高
if trend > 0.1:
trend_factor = 0.95 # 阈值上升过快，略微降低
elif trend < -0.1:
trend_factor = 1.05 # 阈值下降过快，略微提高
# 6. 计算最终阈值
dynamic_threshold = base_threshold * corr_factor * state_factor * volatility_factor * trend_factor
# 限制阈值在合理范围内（2-5）
dynamic_threshold = max(2.0, min(dynamic_threshold, 5.0))
# 7. 保存阈值历史记录
history.append(dynamic_threshold)
# 保持历史记录大小
if len(history) > self._history_size:
history.pop(0)
# 记录阈值计算详情
logger.info(f"[{pair}] 浮动阈值计算: 基础={base_threshold:.2f}, 相关性={correlation:.2f}(因子={corr_factor:.2f}), "
f"市场状态={state_factor:.2f}, 波动性={volatility_factor:.2f}, 趋势={trend_factor:.2f}, "
f"最终阈值={dynamic_threshold:.2f}")
return dynamic_threshold
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# 计算 3m 周期的指标
@ -140,6 +250,9 @@ class FreqaiPrimer(IStrategy):
dataframe['ema_50_3m'] = ta.ema(dataframe['close'], length=50)
dataframe['ema_200_3m'] = ta.ema(dataframe['close'], length=200)
# 计算ATR指标（用于衡量市场波动性）
dataframe['atr'] = ta.atr(dataframe['high'], dataframe['low'], dataframe['close'], length=14)
# 成交量过滤
dataframe['volume_ma'] = dataframe['volume'].rolling(20).mean()
@ -358,11 +471,15 @@ class FreqaiPrimer(IStrategy):
if 'prev_market_state' not in dataframe.columns:
dataframe['prev_market_state'] = 'neutral'
# 计算ATR指标（用于波动性计算）
if 'atr' not in dataframe.columns:
# 使用ta.atr计算ATR指标（周期14）
dataframe['atr'] = ta.atr(dataframe['high'], dataframe['low'], dataframe['close'], length=14)
# 条件1: 价格接近布林带下轨（允许一定偏差）
close_to_bb_lower_1h = (dataframe['close'] <= dataframe['bb_lower_1h'] * self.bb_lower_deviation.value) # 可优化偏差
close_to_bb_lower_1h = (dataframe['close'] <= dataframe['bb_lower_1h'] * self.bb_lower_deviation.value)
# 条件2: RSI 不高于阈值（根据市场状态动态调整）
# 为每一行创建动态阈值
rsi_condition_1h = dataframe.apply(lambda row:
row['rsi_1h'] < self.rsi_bull_threshold.value if row['prev_market_state'] in ['strong_bull', 'weak_bull'] else row['rsi_1h'] < self.rsi_oversold.value,
axis=1)
@ -381,22 +498,36 @@ class FreqaiPrimer(IStrategy):
# 条件6: 布林带宽度过滤（避免窄幅震荡）
bb_width = (dataframe['bb_upper_1h'] - dataframe['bb_lower_1h']) / dataframe['close']
bb_width_condition = bb_width > self.bb_width_threshold.value # 可优化的布林带宽度阈值
bb_width_condition = bb_width > self.bb_width_threshold.value
# 辅助条件: 3m 和 15m 趋势确认（允许部分时间框架不一致）
trend_confirmation = (dataframe['trend_3m'] == 1) | (dataframe['trend_15m'] == 1)
# 合并所有条件（减少强制性条件）
# 至少满足多个条件中的一定数量
# 合并所有条件
conditions = DataFrame({
'close_to_bb': close_to_bb_lower_1h,
'rsi': rsi_condition_1h,
'stochrsi': stochrsi_condition_1h,
'macd': macd_condition_1h,
'vol_bb': (volume_spike | bb_width_condition),
'trend': trend_confirmation
})
# 使用纯算法计算动态阈值，不依赖Hyperopt
dynamic_threshold = self._calculate_dynamic_threshold(dataframe, metadata, conditions)
# 计算条件计数
condition_count = (
close_to_bb_lower_1h.astype(int) +
rsi_condition_1h.astype(int) +
stochrsi_condition_1h.astype(int) +
macd_condition_1h.astype(int) +
(volume_spike | bb_width_condition).astype(int) + # 成交量或布林带宽度满足其一即可
(volume_spike | bb_width_condition).astype(int) +
trend_confirmation.astype(int)
)
final_condition = condition_count >= self.min_condition_count.value
# 应用动态阈值
final_condition = condition_count >= dynamic_threshold
# 设置入场信号
dataframe.loc[final_condition, 'enter_long'] = 1
@ -409,19 +540,10 @@ class FreqaiPrimer(IStrategy):
logger.info(f" - MACD 上升趋势: {macd_condition_1h.sum()} 次")
logger.info(f" - 成交量或布林带宽度: {(volume_spike | bb_width_condition).sum()} 次")
logger.info(f" - 趋势确认: {trend_confirmation.sum()} 次")
logger.info(f" - 条件平均相关性: {conditions.corr().mean().mean():.2f}")
logger.info(f" - 应用动态阈值: {dynamic_threshold:.2f}")
logger.info(f" - 最终条件: {final_condition.sum()} 次")
# 在populate_entry_trend方法末尾添加
# 计算条件间的相关性
conditions = DataFrame({
'close_to_bb': close_to_bb_lower_1h,
'rsi': rsi_condition_1h,
'stochrsi': stochrsi_condition_1h,
'macd': macd_condition_1h,
'vol_bb': (volume_spike | bb_width_condition),
'trend': trend_confirmation
})
correlation = conditions.corr().mean().mean()
logger.info(f"[{metadata['pair']}] 条件平均相关性: {correlation:.2f}")
# 日志记录
if dataframe['enter_long'].sum() > 0:
logger.info(f"[{metadata['pair']}] 发现入场信号数量: {dataframe['enter_long'].sum()}")