251 lines
10 KiB
Python
251 lines
10 KiB
Python
import logging
|
||
import numpy as np
|
||
import datetime
|
||
from functools import reduce
|
||
from freqtrade.persistence import Trade
|
||
import talib.abstract as ta
|
||
from pandas import DataFrame
|
||
from typing import Dict
|
||
from technical import qtpylib
|
||
from freqtrade.strategy import IStrategy
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class FreqaiPrimer(IStrategy):
|
||
"""
|
||
策略说明：
|
||
- 所有交易信号由 FreqAI 动态预测
|
||
- 不使用 Hyperopt 优化任何参数
|
||
- 使用 FreqAI 提供的趋势和波动率信号进行交易决策
|
||
"""
|
||
|
||
use_custom_stoploss = True
|
||
plot_config = {
|
||
"main_plot": {},
|
||
"subplots": {
|
||
"Signals": {
|
||
"enter_long": {"color": "green"},
|
||
"exit_long": {"color": "red"}
|
||
},
|
||
"FreqAI Predictions": {
|
||
"&-buy_signal": {"color": "blue"},
|
||
"&-sell_signal": {"color": "orange"},
|
||
"&-volatility_forecast": {"color": "purple"}
|
||
}
|
||
}
|
||
}
|
||
|
||
freqai_info = {
|
||
"model": "LightGBMClassifier",
|
||
"feature_parameters": {
|
||
"include_timeframes": ["5m", "15m", "1h"],
|
||
"label_period_candles": 24,
|
||
"include_shifted_candles": 3,
|
||
},
|
||
"data_split_parameters": {
|
||
"test_size": 0.2,
|
||
"shuffle": False,
|
||
},
|
||
"model_training_parameters": {
|
||
"n_estimators": 200,
|
||
"learning_rate": 0.05,
|
||
"num_leaves": 31,
|
||
"verbose": -1,
|
||
},
|
||
}
|
||
|
||
def feature_engineering_expand_all(self, dataframe: DataFrame, period: int, metadata: dict, **kwargs) -> DataFrame:
|
||
# 保留原有指标
|
||
dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=period)
|
||
dataframe["%-sma-period"] = ta.SMA(dataframe, timeperiod=period)
|
||
dataframe["%-ema-period"] = ta.EMA(dataframe, timeperiod=period)
|
||
|
||
real = ta.TYPPRICE(dataframe)
|
||
upperband, middleband, lowerband = ta.BBANDS(real, timeperiod=period, nbdevup=2.0, nbdevdn=2.0)
|
||
dataframe["bb_lowerband-period"] = lowerband
|
||
dataframe["bb_upperband-period"] = upperband
|
||
dataframe["bb_middleband-period"] = middleband
|
||
dataframe["%-bb_width-period"] = (dataframe["bb_upperband-period"] - dataframe["bb_lowerband-period"]) / dataframe["bb_middleband-period"]
|
||
dataframe["%-roc-period"] = ta.ROC(dataframe, timeperiod=period)
|
||
|
||
# 添加策略 B 的指标
|
||
dataframe["%-mfi-period"] = ta.MFI(dataframe, timeperiod=period) # 资金流指标
|
||
dataframe["%-adx-period"] = ta.ADX(dataframe, timeperiod=period) # 趋势强度
|
||
dataframe["%-tema-period"] = ta.TEMA(dataframe, timeperiod=period) # TEMA 指标
|
||
dataframe["%-relative_volume-period"] = dataframe["volume"] / dataframe["volume"].rolling(period).mean() # 相对成交量
|
||
dataframe["%-close-bb_lower-period"] = dataframe["close"] / dataframe["bb_lowerband-period"] # 收盘价/布林带下轨
|
||
|
||
# 数据清理
|
||
columns_to_clean = [
|
||
"%-rsi-period", "%-mfi-period", "%-sma-period", "%-ema-period", "%-adx-period",
|
||
"bb_lowerband-period", "bb_middleband-period", "bb_upperband-period",
|
||
"%-bb_width-period", "%-roc-period", "%-relative_volume-period", "%-close-bb_lower-period", "%-tema-period"
|
||
]
|
||
for col in columns_to_clean:
|
||
dataframe[col] = dataframe[col].replace([np.inf, -np.inf], 0).ffill().fillna(0)
|
||
|
||
return dataframe
|
||
|
||
def set_freqai_targets(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame:
|
||
"""
|
||
使用历史窗口预测未来趋势和波动率作为辅助信号，
|
||
避免使用任何未来数据（如 shift(-N)）
|
||
"""
|
||
|
||
label_period = self.freqai_info["feature_parameters"]["label_period_candles"]
|
||
|
||
# 1. 趋势强度：使用过去 N 根 K 线的收益率判断当前趋势
|
||
dataframe["&-trend_strength"] = dataframe["close"].pct_change(label_period)
|
||
|
||
# 2. 波动率预测：使用过去 N 根 K 线的价格变动绝对值
|
||
dataframe["&-volatility_forecast"] = dataframe["close"].pct_change(label_period).abs()
|
||
|
||
# 3. 动态 ROI 目标：基于趋势强度缩放
|
||
dataframe["&-roi_target"] = np.where(
|
||
dataframe["&-trend_strength"] > 0,
|
||
dataframe["&-trend_strength"] * 1.5,
|
||
0.01 # 最小 ROI
|
||
)
|
||
|
||
# 4. 动态止损目标：基于波动率缩放
|
||
dataframe["&-stoploss_target"] = -dataframe["&-volatility_forecast"] * 1.2
|
||
|
||
# 5. 市场状态识别：判断当前是震荡、趋势、超买还是超卖
|
||
if "bb_upperband-period" in dataframe.columns and "bb_lowerband-period" in dataframe.columns:
|
||
dataframe["&-market_condition"] = np.select([
|
||
(dataframe["close"] > dataframe["bb_upperband-period"]),
|
||
(dataframe["close"] < dataframe["bb_lowerband-period"]),
|
||
(dataframe["&-trend_strength"] > 0.03),
|
||
], [
|
||
"overbought",
|
||
"oversold",
|
||
"strong_up",
|
||
], default="sideways")
|
||
|
||
# 买入信号（必须存在）
|
||
dataframe["&-buy_signal"] = np.where(dataframe["&-trend_strength"] > 0.01, 1, 0)
|
||
|
||
# 卖出信号（必须存在）
|
||
dataframe["&-sell_signal"] = np.where(dataframe["&-trend_strength"] < -0.01, 1, 0)
|
||
|
||
|
||
return dataframe
|
||
|
||
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
logger.info(f"[{metadata['pair']}] 当前可用列: {list(dataframe.columns)}")
|
||
dataframe = self.freqai.start(dataframe, metadata, self)
|
||
|
||
# FreqAI 提供的预测列
|
||
if "&-trend" in dataframe.columns:
|
||
dataframe["in_uptrend"] = dataframe["&-trend"] > 0.5
|
||
# 将 roi_target 存入 dataframe，供后续使用
|
||
if "&-roi_target" in dataframe.columns:
|
||
dataframe["&-roi_target"] = dataframe["&-roi_target"].ffill().fillna(0.01)
|
||
else:
|
||
dataframe["&-roi_target"] = 0.01
|
||
return dataframe
|
||
|
||
def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
|
||
conditions = []
|
||
|
||
if "&-buy_signal" in df.columns:
|
||
# 这个原先逻辑, 轻量持仓: buy_condition = df["&-buy_signal"] > 0.5
|
||
# 使用过去 20 根 K 线中 buy_signal 的 80 分位作为阈值, 这个按说算是次轻量持仓
|
||
df["buy_threshold"] = df["&-buy_signal"].rolling(20).quantile(0.8)
|
||
buy_condition = df["&-buy_signal"] >= df["buy_threshold"]
|
||
|
||
# 根据市场状态调整买入信号
|
||
if "&-market_condition" in df.columns:
|
||
# 在震荡市场中减少交易频率（例如只保留 50% 的信号）
|
||
buy_condition &= ~((df["&-market_condition"] == "sideways") & (df["&-buy_signal"] < 0.7))
|
||
|
||
# 在超买市场中进一步过滤买入信号
|
||
buy_condition &= ~(df["&-market_condition"] == "overbought")
|
||
|
||
conditions.append(buy_condition)
|
||
else:
|
||
logger.warning("⚠️ &-buy_signal 列缺失，跳过该条件")
|
||
|
||
if len(conditions) == 0:
|
||
return df
|
||
|
||
df.loc[reduce(lambda x, y: x & y, conditions), 'enter_long'] = 1
|
||
return df
|
||
|
||
|
||
def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
|
||
conditions = [
|
||
# 容易过早离场 df["&-sell_signal"] > 0.5,
|
||
# 只有当 sell_signal 持续高于某个较低阈值时才触发
|
||
df["sell_signal_high"] = (df["&-sell_signal"] > 0.3).rolling(3).sum() >= 2
|
||
conditions.append(df["sell_signal_high"])
|
||
]
|
||
|
||
if "&-market_condition" in df.columns:
|
||
# 在超卖市场中减少卖出信号
|
||
conditions.append(df["&-market_condition"] != "oversold")
|
||
|
||
# 在超买市场中增强卖出信号
|
||
conditions.append(~((df["&-market_condition"] == "overbought") & (df["&-sell_signal"] < 0.7)))
|
||
|
||
df.loc[reduce(lambda x, y: x & y, conditions), 'exit_long'] = 1
|
||
return df
|
||
def adjust_trade_position(self, trade: Trade, current_time: datetime,
|
||
current_rate: float, current_profit: float,
|
||
min_roi: Dict[float, float], max_profit: float):
|
||
"""
|
||
动态调整当前交易的止盈目标。
|
||
"""
|
||
|
||
# 获取当前币种的 dataframe
|
||
pair = trade.pair
|
||
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
||
last_candle = dataframe.iloc[-1]
|
||
|
||
# 获取预测的 ROI 目标
|
||
roi_target = last_candle.get("&-roi_target", 0.01)
|
||
|
||
# 根据预测值生成新的 ROI 表
|
||
if roi_target > 0.05:
|
||
new_minimal_roi = {
|
||
"0": roi_target * 1.0,
|
||
"10": roi_target * 0.8,
|
||
"30": roi_target * 0.5,
|
||
"60": roi_target * 0.2,
|
||
"120": 0.01
|
||
}
|
||
elif 0.02 <= roi_target <= 0.05:
|
||
new_minimal_roi = {
|
||
"0": roi_target * 1.0,
|
||
"20": roi_target * 0.7,
|
||
"60": roi_target * 0.3,
|
||
"120": 0.01
|
||
}
|
||
else:
|
||
new_minimal_roi = {
|
||
"0": max(0.01, roi_target),
|
||
"60": 0.01,
|
||
"180": 0.01
|
||
}
|
||
|
||
# 返回新的 minimal_roi
|
||
return new_minimal_roi
|
||
|
||
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
|
||
current_rate: float, current_profit: float, **kwargs) -> float:
|
||
"""
|
||
动态止损回调函数，基于 FreqAI 预测的 &-stoploss_target 值。
|
||
"""
|
||
|
||
# 获取当前币种的 dataframe
|
||
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
||
last_candle = dataframe.iloc[-1]
|
||
|
||
# 获取预测的止损目标
|
||
stoploss_target = last_candle.get("&-stoploss_target", -0.01)
|
||
|
||
# 将 stoploss_target 转换为负数（因为 stoploss 是负值）
|
||
dynamic_stoploss = min(-0.005, stoploss_target) # 设置最小止损限制
|
||
|
||
return dynamic_stoploss
|