```python 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 ``` ``` ``` ``` ``` ``` ``` ```