import logging import numpy as np import pandas as pd from pandas import DataFrame from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter import pandas_ta as ta logger = logging.getLogger(__name__) class FreqaiPrimerML(IStrategy): """ 基于 FreqAI ML 的交易策略 - 单目标回归模型 预测未来价格上升幅度，用于入场决策 """ # FreqAI 配置 - 直接在类中定义 freqai_info = { "identifier": "freqai_primer_ml", "model": "LightGBMRegressor", "feature_parameters": { "include_timeframes": ["3m", "15m", "1h"], "include_shifted_candles": 3, "label_period_candles": 6, }, "data_split_parameters": { "test_size": 0.2, "shuffle": False, }, "model_training_parameters": { "n_estimators": 200, "learning_rate": 0.05, "num_leaves": 31, "verbose": -1, }, "fit_live_predictions_candles": 100, "live_retrain_candles": 100, } # 策略参数 stoploss = -0.15 trailing_stop = True trailing_stop_positive_offset = 0.005 timeframe = "3m" can_short = False # 可优化参数 bb_std = DecimalParameter(2.0, 5.0, decimals=1, default=3.5, optimize=True, space='buy') rsi_length = IntParameter(10, 30, default=14, optimize=True, space='buy') rsi_oversold = IntParameter(20, 50, default=30, optimize=True, space='buy') def informative_pairs(self): pairs = self.dp.current_whitelist() return [(pair, '15m') for pair in pairs] + [(pair, '1h') for pair in pairs] def feature_engineering_expand_all(self, dataframe: DataFrame, period, metadata, **kwargs) -> DataFrame: """ 为 FreqAI 生成扩展特征 """ # RSI 特征 dataframe[f"%-rsi-{period}"] = ta.rsi(dataframe["close"], length=period) # 成交量特征 dataframe[f"%-volume_ratio-{period}"] = dataframe["volume"] / dataframe["volume"].rolling(period).mean() # ADX 趋势强度 adx_result = ta.adx(dataframe["high"], dataframe["low"], dataframe["close"], length=period) dataframe[f"%-adx-{period}"] = adx_result[f"ADX_{period}"] # 价格位置特征 low = dataframe["low"].rolling(period).min() high = dataframe["high"].rolling(period).max() dataframe[f"%-price_position-{period}"] = (dataframe["close"] - low) / (high - low + 1e-8) return dataframe def feature_engineering_standard(self, dataframe: DataFrame, metadata, **kwargs) -> DataFrame: """ 标准特征（不需要自动扩展） """ # 小时时间特征 dataframe["%-hour_of_day"] = (dataframe["date"].dt.hour + 1) / 25 # ATR 波动性 atr_14 = ta.atr(dataframe["high"], dataframe["low"], dataframe["close"], length=14) dataframe["%-atr_14"] = atr_14 / dataframe["close"] # EMA 比率 ema_5 = ta.ema(dataframe["close"], length=5) ema_10 = ta.ema(dataframe["close"], length=10) dataframe["%-ema_ratio"] = ema_5 / (ema_10 + 1e-8) return dataframe def set_freqai_targets(self, dataframe: DataFrame, metadata, **kwargs) -> DataFrame: """ 为 FreqAI 生成回归目标：预测未来价格上升幅度 """ pair = metadata.get('pair', 'Unknown') # 获取标签周期 label_period = self.freqai_info["feature_parameters"]["label_period_candles"] # 计算未来 label_period 根 K 线内的最大上升幅度 dataframe["&-price_rise"] = 0.0 for i in range(len(dataframe) - label_period): future_high = dataframe["high"].iloc[i:i+label_period].max() current_price = dataframe["close"].iloc[i] # 计算上升幅度（百分比） max_rise_pct = (future_high - current_price) / current_price if current_price != 0 else 0 dataframe.loc[i, "&-price_rise"] = max(0, max_rise_pct) # 只记录上升，下跌记为0 # 平滑处理 dataframe["&-price_rise"] = dataframe["&-price_rise"].rolling(3).mean().fillna(0) logger.debug(f"[{pair}] 目标标签统计: 均值={dataframe['&-price_rise'].mean():.6f}, " f"最小={dataframe['&-price_rise'].min():.6f}, " f"最大={dataframe['&-price_rise'].max():.6f}") return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 主指标计算函数 - FreqAI 在最开始调用 """ pair = metadata.get('pair', 'Unknown') # 🔧 第一步：调用 FreqAI 进行预测和训练 if hasattr(self, 'freqai') and self.freqai is not None: dataframe = self.freqai.start(dataframe, metadata, self) logger.debug(f"[{pair}] FreqAI 预测已生成，列: {list(dataframe.columns)}") else: logger.warning(f"[{pair}] FreqAI 未初始化") # 计算基础指标 rsi_length = self.rsi_length.value bb_std = self.bb_std.value bb_length = 40 # 布林带 bb = ta.bbands(dataframe['close'], length=bb_length, std=bb_std) dataframe['bb_lower'] = bb[f'BBL_{bb_length}_{bb_std}'] dataframe['bb_upper'] = bb[f'BBU_{bb_length}_{bb_std}'] dataframe['bb_middle'] = bb[f'BBM_{bb_length}_{bb_std}'] # RSI dataframe['rsi'] = ta.rsi(dataframe['close'], length=rsi_length) # EMA dataframe['ema_50'] = ta.ema(dataframe['close'], length=50) dataframe['ema_200'] = ta.ema(dataframe['close'], length=200) # 成交量均线 dataframe['volume_ma'] = dataframe['volume'].rolling(20).mean() # MACD macd = ta.macd(dataframe['close']) dataframe['macd'] = macd['MACD_12_26_9'] dataframe['macd_signal'] = macd['MACDs_12_26_9'] # 填充缺失值 dataframe = dataframe.fillna(0).bfill() return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 入场逻辑 - 使用 ML 预测 """ conditions = [] # 条件1: 价格接近下轨 conditions.append(dataframe['close'] <= dataframe['bb_lower'] * 1.02) # 条件2: RSI 超卖 conditions.append(dataframe['rsi'] < self.rsi_oversold.value) # 条件3: MACD 底部 conditions.append(dataframe['macd'] < dataframe['macd_signal']) # 条件4: EMA 多头排列 conditions.append(dataframe['ema_50'] > dataframe['ema_200']) # 传统条件 traditional_signal = (conditions[0] & conditions[1] & conditions[2] & conditions[3]) # ML 增强：使用 FreqAI 预测的价格上升幅度 if '&-price_rise' in dataframe.columns: ml_signal = dataframe['&-price_rise'] > 0.01 # 预测上升 > 1% final_signal = traditional_signal & ml_signal logger.debug(f"[{metadata['pair']}] 使用 ML 增强的入场信号") else: final_signal = traditional_signal logger.debug(f"[{metadata['pair']}] 仅使用传统入场信号（ML 未初始化）") dataframe.loc[final_signal, 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 出场逻辑 """ conditions = [] # 条件1: 价格突破上轨 conditions.append(dataframe['close'] >= dataframe['bb_upper'] * 0.99) # 条件2: RSI 超买 conditions.append(dataframe['rsi'] > 70) # 条件3: MACD 下降 conditions.append(dataframe['macd'] > dataframe['macd_signal']) # 至少满足 2 个条件 condition_count = conditions[0].astype(int) + conditions[1].astype(int) + conditions[2].astype(int) final_signal = condition_count >= 2 dataframe.loc[final_signal, 'exit_long'] = 1 return dataframe