import logging import numpy as np from functools import reduce import talib.abstract as ta from pandas import DataFrame from technical import qtpylib from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter logger = logging.getLogger(__name__) class MyStrategy(IStrategy): minimal_roi = { 0: 0.135, 9: 0.052, 15: 0.007, 60: 0 } stoploss = -0.263 trailing_stop = True trailing_stop_positive = 0.324 trailing_stop_positive_offset = 0.411 trailing_only_offset_is_reached = False max_open_trades = 4 process_only_new_candles = True use_exit_signal = True startup_candle_count: int = 40 can_short = False buy_rsi = IntParameter(low=10, high=50, default=30, space="buy", optimize=False, load=True) sell_rsi = IntParameter(low=50, high=90, default=70, space="sell", optimize=False, load=True) roi_0 = DecimalParameter(low=0.01, high=0.2, default=0.135, space="roi", optimize=True, load=True) roi_15 = DecimalParameter(low=0.005, high=0.1, default=0.052, space="roi", optimize=True, load=True) roi_30 = DecimalParameter(low=0.001, high=0.05, default=0.007, space="roi", optimize=True, load=True) stoploss_param = DecimalParameter(low=-0.35, high=-0.1, default=-0.263, space="stoploss", optimize=True, load=True) trailing_stop_positive_param = DecimalParameter(low=0.1, high=0.5, default=0.324, space="trailing", optimize=True, load=True) trailing_stop_positive_offset_param = DecimalParameter(low=0.2, high=0.6, default=0.411, space="trailing", optimize=True, load=True) 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, }, "model_training_parameters": { "n_estimators": 200, "learning_rate": 0.05, "num_leaves": 31, "verbose": -1, }, } plot_config = { "main_plot": {}, "subplots": { "&-buy_rsi": {"&-buy_rsi": {"color": "green"}}, "&-sell_rsi": {"&-sell_rsi": {"color": "red"}}, "&-stoploss": {"&-stoploss": {"color": "purple"}}, "&-roi_0": {"&-roi_0": {"color": "orange"}}, "do_predict": {"do_predict": {"color": "brown"}}, }, } def feature_engineering_expand_all(self, dataframe: DataFrame, period: int, metadata: dict, **kwargs) -> DataFrame: dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=period) dataframe["%-mfi-period"] = ta.MFI(dataframe, timeperiod=period) dataframe["%-sma-period"] = ta.SMA(dataframe, timeperiod=period) dataframe["%-ema-period"] = ta.EMA(dataframe, timeperiod=period) dataframe["%-adx-period"] = ta.ADX(dataframe, timeperiod=period) bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=period, stds=2.2) dataframe["bb_lowerband-period"] = bollinger["lower"] dataframe["bb_middleband-period"] = bollinger["mid"] dataframe["bb_upperband-period"] = bollinger["upper"] dataframe["%-bb_width-period"] = ( dataframe["bb_upperband-period"] - dataframe["bb_lowerband-period"] ) / dataframe["bb_middleband-period"] dataframe["%-close-bb_lower-period"] = dataframe["close"] / dataframe["bb_lowerband-period"] dataframe["%-roc-period"] = ta.ROC(dataframe, timeperiod=period) dataframe["%-relative_volume-period"] = ( dataframe["volume"] / dataframe["volume"].rolling(period).mean() ) dataframe = dataframe.replace([np.inf, -np.inf], 0) dataframe = dataframe.ffill() dataframe = dataframe.fillna(0) return dataframe def feature_engineering_expand_basic(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame: dataframe["%-pct-change"] = dataframe["close"].pct_change() dataframe["%-raw_volume"] = dataframe["volume"] dataframe["%-raw_price"] = dataframe["close"] dataframe = dataframe.replace([np.inf, -np.inf], 0) dataframe = dataframe.ffill() dataframe = dataframe.fillna(0) return dataframe def feature_engineering_standard(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame: if len(dataframe["close"]) < 20: logger.warning(f"数据不足 {len(dataframe)} 根 K 线，%-volatility 可能不完整") dataframe["%-day_of_week"] = dataframe["date"].dt.dayofweek dataframe["%-hour_of_day"] = dataframe["date"].dt.hour dataframe["%-volatility"] = dataframe["close"].pct_change().rolling(20, min_periods=1).std() dataframe["%-volatility"] = dataframe["%-volatility"].replace([np.inf, -np.inf], 0) dataframe["%-volatility"] = dataframe["%-volatility"].ffill() dataframe["%-volatility"] = dataframe["%-volatility"].fillna(0) return dataframe 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"] if "%-volatility" not in dataframe.columns: logger.warning("缺少 %-volatility 列，强制重新生成") dataframe["%-volatility"] = dataframe["close"].pct_change().rolling(20, min_periods=1).std() dataframe["%-volatility"] = dataframe["%-volatility"].replace([np.inf, -np.inf], 0) dataframe["%-volatility"] = dataframe["%-volatility"].ffill() dataframe["%-volatility"] = dataframe["%-volatility"].fillna(0) # 移除 shift(-label_period)，改为使用当前及过去的数据 dataframe["&-buy_rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["&-buy_rsi"] = dataframe["&-buy_rsi"].rolling(window=label_period).mean().ffill().bfill() for col in ["&-buy_rsi", "%-volatility"]: dataframe[col] = dataframe[col].replace([np.inf, -np.inf], 0) dataframe[col] = dataframe[col].ffill() dataframe[col] = dataframe[col].fillna(0) if dataframe[col].isna().any(): logger.warning(f"目标列 {col} 仍包含 NaN，数据预览：\n{dataframe[col].tail(10)}") except Exception as e: logger.error(f"创建 FreqAI 目标失败：{str(e)}") raise logger.info(f"目标列预览：\n{dataframe[['&-buy_rsi']].head().to_string()}") return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: logger.info(f"处理交易对：{metadata['pair']}") logger.debug(f"输入特征列：{list(dataframe.columns)}") dataframe = self.freqai.start(dataframe, metadata, self) logger.debug(f"FreqAI 输出特征列：{list(dataframe.columns)}") 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) label_period = self.freqai_info["feature_parameters"]["label_period_candles"] # 使用滚动窗口而非未来函数来生成 up_or_down 列 dataframe["up_or_down"] = np.where( dataframe["close"].rolling(window=label_period).mean() > dataframe["close"], 1, 0 ) if "&-buy_rsi" in dataframe.columns: if "%-volatility" not in dataframe.columns: logger.warning("缺少 %-volatility 列，强制重新生成") dataframe["%-volatility"] = dataframe["close"].pct_change().rolling(20, min_periods=1).std() dataframe["%-volatility"] = dataframe["%-volatility"].replace([np.inf, -np.inf], 0) dataframe["%-volatility"] = dataframe["%-volatility"].ffill() dataframe["%-volatility"] = dataframe["%-volatility"].fillna(0) dataframe["&-sell_rsi"] = dataframe["&-buy_rsi"] + 30 dataframe["&-stoploss"] = self.stoploss - (dataframe["%-volatility"] * 5).clip(-0.05, 0.05) dataframe["&-roi_0"] = (dataframe["close"].rolling(window=label_period).mean() / dataframe["close"] - 1).clip(0, 0.2) for col in ["&-buy_rsi", "&-sell_rsi", "&-stoploss", "&-roi_0"]: dataframe[col] = dataframe[col].replace([np.inf, -np.inf], 0) dataframe[col] = dataframe[col].ffill() dataframe[col] = dataframe[col].fillna(0) dataframe["buy_rsi_pred"] = dataframe["&-buy_rsi"].rolling(5).mean().clip(10, 50) dataframe["sell_rsi_pred"] = dataframe["&-sell_rsi"].rolling(5).mean().clip(50, 90) dataframe["stoploss_pred"] = dataframe["&-stoploss"].clip(-0.35, -0.1) dataframe["roi_0_pred"] = dataframe["&-roi_0"].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，填充为默认值") dataframe[col] = dataframe[col].ffill() dataframe[col] = dataframe[col].fillna(dataframe[col].mean()) dataframe["trailing_stop_positive"] = (dataframe["roi_0_pred"] * 0.5).clip(0.01, 0.3) dataframe["trailing_stop_positive_offset"] = (dataframe["roi_0_pred"] * 0.75).clip(0.02, 0.4) self.buy_rsi.value = float(dataframe["buy_rsi_pred"].iloc[-1]) self.sell_rsi.value = float(dataframe["sell_rsi_pred"].iloc[-1]) self.stoploss = float(self.stoploss_param.value) self.minimal_roi = { 0: float(self.roi_0.value), 15: float(self.roi_15.value), 30: float(self.roi_30.value), 60: 0 } self.trailing_stop_positive = float(self.trailing_stop_positive_param.value) self.trailing_stop_positive_offset = float(self.trailing_stop_positive_offset_param.value) logger.info(f"动态参数：buy_rsi={self.buy_rsi.value}, sell_rsi={self.sell_rsi.value}, " f"stoploss={self.stoploss}, trailing_stop_positive={self.trailing_stop_positive}") else: logger.warning(f"&-buy_rsi 列缺失，跳过 FreqAI 预测逻辑，检查 freqai.start 输出") dataframe = dataframe.replace([np.inf, -np.inf], 0) dataframe = dataframe.ffill() dataframe = dataframe.fillna(0) logger.info(f"up_or_down 值统计：\n{dataframe['up_or_down'].value_counts().to_string()}") logger.info(f"do_predict 值统计：\n{dataframe['do_predict'].value_counts().to_string()}") logger.debug(f"最终特征列：{list(dataframe.columns)}") return dataframe def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame: enter_long_conditions = [ qtpylib.crossed_above(df["rsi"], df["buy_rsi_pred"]), df["tema"] > df["tema"].shift(1), df["volume"] > 0, df["do_predict"] == 1, df["up_or_down"] == 1 ] if enter_long_conditions: df.loc[ reduce(lambda x, y: x & y, enter_long_conditions), ["enter_long", "enter_tag"] ] = (1, "long") 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["close"].shift(1) * 0.97), df["volume"] > 0, df["do_predict"] == 1, df["up_or_down"] == 0 ] if exit_long_conditions: df.loc[ reduce(lambda x, y: x & y, exit_long_conditions), "exit_long" ] = 1 return df def confirm_trade_entry( self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time, entry_tag, side: str, **kwargs ) -> bool: try: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df is None or df.empty: logger.warning(f"无法获取 {pair} 的分析数据，拒绝交易") return False last_candle = df.iloc[-1].squeeze() if "close" not in last_candle or np.isnan(last_candle["close"]): logger.warning(f"{pair} 的最新 K 线缺少有效 close 价格，拒绝交易") return False if side == "long": max_rate = last_candle["close"] * (1 + 0.0025) # 0.25% 滑点阈值 if rate > max_rate: logger.debug(f"拒绝 {pair} 的买入，价格 {rate} 超过最大允许价格 {max_rate}") return False elif side == "short": logger.warning(f"{pair} 尝试做空，但策略不支持做空 (can_short={self.can_short})") return False logger.debug(f"确认 {pair} 的交易：side={side}, rate={rate}, close={last_candle['close']}") return True except Exception as e: logger.error(f"确认 {pair} 交易时出错：{str(e)}") return False