8.3 KiB
8.3 KiB
class FreqaiExampleStrategy(IStrategy):
minimal_roi = {}
stoploss = -0.1
trailing_stop = True
process_only_new_candles = True
use_exit_signal = True
startup_candle_count: int = 100 # 增加数据需求
can_short = False
# Hyperopt 参数
buy_rsi = IntParameter(low=10, high=50, default=27, space="buy", optimize=False, load=True)
sell_rsi = IntParameter(low=50, high=90, default=59, space="sell", optimize=False, load=True)
roi_0 = DecimalParameter(low=0.01, high=0.2, default=0.038, space="roi", optimize=True, load=True)
roi_15 = DecimalParameter(low=0.005, high=0.1, default=0.027, space="roi", optimize=True, load=True)
roi_30 = DecimalParameter(low=0.001, high=0.05, default=0.009, space="roi", optimize=True, load=True)
stoploss_param = DecimalParameter(low=-0.25, high=-0.05, default=-0.1, space="stoploss", optimize=True, load=True)
# 保护机制
protections = [
{"method": "StoplossGuard", "stop_duration": 60, "lookback_period": 120},
{"method": "MaxDrawdown", "lookback_period": 120, "max_allowed_drawdown": 0.05}
]
# FreqAI 配置
freqai_info = {
"model": "LightGBMRegressor",
"feature_parameters": {
"include_timeframes": ["5m", "15m", "1h"],
"include_corr_pairlist": [],
"label_period_candles": 12,
"include_shifted_candles": 3,
},
"data_split_parameters": {
"test_size": 0.2,
"shuffle": False,
"n_splits": 5 # 添加交叉验证
},
"model_training_parameters": {
"n_estimators": 200,
"learning_rate": 0.05,
"num_leaves": 10,
"min_child_weight": 1,
"verbose": -1,
},
}
def set_freqai_targets(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame:
logger.info(f"设置 FreqAI 目标，交易对：{metadata['pair']}")
if "close" not in dataframe.columns:
logger.error("数据框缺少必要的 'close' 列")
raise ValueError("数据框缺少必要的 'close' 列")
try:
label_period = self.freqai_info["feature_parameters"]["label_period_candles"]
dataframe["%-volatility"] = dataframe["close"].pct_change().rolling(20).std()
dataframe["&-buy_rsi"] = (dataframe["close"].shift(-label_period) / dataframe["close"] - 1) * 100 # 修改目标为收益率
for col in ["&-buy_rsi", "%-volatility"]:
dataframe[col] = dataframe[col].replace([np.inf, -np.inf], 0)
dataframe[col] = dataframe[col].ffill()
if dataframe[col].isna().any():
logger.warning(f"目标列 {col} 包含 NaN，填充为 0")
dataframe[col] = dataframe[col].fillna(0)
logger.info(f"目标列 {col} 统计：\n{dataframe[col].describe().to_string()}")
except Exception as e:
logger.error(f"创建 FreqAI 目标失败：{str(e)}")
raise
return dataframe
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
logger.info(f"处理交易对：{metadata['pair']}, 数据形状：{dataframe.shape}")
dataframe = self.freqai.start(dataframe, metadata, self)
# 计算传统指标
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2)
dataframe["bb_lowerband"] = bollinger["lower"]
dataframe["bb_middleband"] = bollinger["mid"]
dataframe["bb_upperband"] = bollinger["upper"]
dataframe["tema"] = ta.TEMA(dataframe, timeperiod=9)
# 检查FreqAI预测列
if "&-buy_rsi_pred" in dataframe.columns:
logger.info(f"&-buy_rsi_pred 统计：均值={dataframe['&-buy_rsi_pred'].mean():.2f}, 标准差={dataframe['&-buy_rsi_pred'].std():.2f}")
dataframe["buy_rsi_trend"] = np.where(
dataframe["&-buy_rsi_pred"] > dataframe["&-buy_rsi_pred"].shift(1), 1, 0
)
dataframe["&-sell_rsi_pred"] = dataframe["&-buy_rsi_pred"] + 30
dataframe["%-volatility"] = dataframe["close"].pct_change().rolling(20).std()
dataframe["&-stoploss_pred"] = -0.1 - (dataframe["%-volatility"] * 10).clip(0, 0.25)
dataframe["&-roi_0_pred"] = (dataframe["&-buy_rsi_pred"] / 1000).clip(0.01, 0.2)
for col in ["&-buy_rsi_pred", "&-sell_rsi_pred", "&-stoploss_pred", "&-roi_0_pred"]:
if dataframe[col].isna().any():
logger.warning(f"列 {col} 包含 NaN，填充为默认值")
mean_value = dataframe[col].mean()
if pd.isna(mean_value):
mean_value = {
"&-buy_rsi_pred": 30,
"&-sell_rsi_pred": 70,
"&-stoploss_pred": -0.1,
"&-roi_0_pred": 0.05
}.get(col, 0)
dataframe[col] = dataframe[col].fillna(mean_value)
else:
logger.warning(f"&-buy_rsi_pred 列缺失，使用默认值初始化")
dataframe["buy_rsi_trend"] = 0
dataframe["&-buy_rsi_pred"] = 30
dataframe["&-sell_rsi_pred"] = 70
dataframe["&-stoploss_pred"] = -0.1
dataframe["&-roi_0_pred"] = 0.05
# 动态参数设置
try:
last_valid_idx = dataframe["&-stoploss_pred"].last_valid_index()
if last_valid_idx is None:
raise ValueError("没有有效的预测数据")
self.stoploss = float(np.clip(dataframe["&-stoploss_pred"].iloc[last_valid_idx], -0.25, -0.05))
self.buy_rsi.value = int(np.clip(dataframe["&-buy_rsi_pred"].iloc[last_valid_idx], 10, 50))
self.sell_rsi.value = int(np.clip(dataframe["&-sell_rsi_pred"].iloc[last_valid_idx], 50, 90))
self.roi_0.value = float(np.clip(dataframe["&-roi_0_pred"].iloc[last_valid_idx], 0.01, 0.2))
self.minimal_roi = {
0: self.roi_0.value,
15: self.roi_15.value,
30: self.roi_30.value,
60: 0.0
}
logger.info(f"动态参数设置: buy_rsi={self.buy_rsi.value}, sell_rsi={self.sell_rsi.value}, stoploss={self.stoploss:.2%}")
except Exception as e:
logger.error(f"动态参数设置失败，使用默认值: {str(e)}")
self.stoploss = -0.1
self.buy_rsi.value = 27
self.sell_rsi.value = 59
self.minimal_roi = {0: 0.038, 15: 0.027, 30: 0.009, 60: 0.0}
dataframe = dataframe.replace([np.inf, -np.inf], 0)
dataframe = dataframe.ffill()
dataframe = dataframe.fillna(0)
logger.info(f"do_predict 分布：\n{dataframe['do_predict'].value_counts().to_string()}")
return dataframe
def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
enter_long_conditions = [
qtpylib.crossed_above(df["rsi"], df["&-buy_rsi_pred"] - 5), # 放宽RSI条件
df["tema"] > df["tema"].shift(1),
df["volume"] > 0,
df["do_predict"] == 1
]
if enter_long_conditions:
condition_met = reduce(lambda x, y: x & y, enter_long_conditions)
df.loc[condition_met, ["enter_long", "enter_tag"]] = (1, "long")
if condition_met.any():
logger.info(f"买入信号触发：{metadata['pair']}，时间={df.index[condition_met][-1]}")
else:
logger.debug(f"买入条件未满足：{metadata['pair']}，do_predict={df['do_predict'].iloc[-1]}, rsi={df['rsi'].iloc[-1]:.2f}, buy_rsi_pred={df['&-buy_rsi_pred'].iloc[-1]:.2f}")
return df
def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
exit_long_conditions = [
qtpylib.crossed_above(df["rsi"], df["&-sell_rsi_pred"]),
df["close"] < df["bb_lowerband"],
df["volume"] > 0,
df["do_predict"] == 1
]
time_exit = (df["date"] >= df["date"].shift(1) + pd.Timedelta(days=2))
df.loc[
(reduce(lambda x, y: x & y, exit_long_conditions)) | time_exit,
"exit_long"
] = 1
return df