高级策略技巧

本章详细介绍 Freqtrade 的高级策略开发技巧，涵盖多时间帧策略、动态参数调整、机器学习信号融合、自定义指标组合以及策略组合等进阶内容，帮助读者开发更复杂、更健壮的交易策略。

多时间帧策略

多周期数据分析

多时间帧策略通过分析不同 K 线周期的数据来做出更准确的交易决策：

from freqtrade.strategy import IStrategy
from pandas import DataFrame
import talib.abstract as ta

class MultiTimeframeStrategy(IStrategy):
    """
    多时间帧策略：
    - 使用 4h 判断大趋势
    - 使用 1h 确认趋势
    - 使用 15m 寻找入场点
    """
    timeframe = "15m"  # 入场时间帧
    startup_candle_count = 200

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        pair = metadata["pair"]

        # 获取 1h 数据
        dataframe_1h = self.dp.get_pair_dataframe(pair, "1h")
        if dataframe_1h is not None and len(dataframe_1h) > 50:
            # 在 1h 数据上计算指标
            dataframe_1h["rsi_1h"] = ta.RSI(dataframe_1h, timeperiod=14)
            dataframe_1h["ema_50_1h"] = ta.EMA(dataframe_1h, timeperiod=50)
            dataframe_1h["ema_200_1h"] = ta.EMA(dataframe_1h, timeperiod=200)

            # 对齐到 15m 数据
            dataframe["rsi_1h"] = dataframe_1h["rsi_1h"].reindex(dataframe.index, method="ffill")
            dataframe["ema_50_1h"] = dataframe_1h["ema_50_1h"].reindex(dataframe.index, method="ffill")
            dataframe["trend_1h"] = (dataframe_1h["ema_50_1h"] > dataframe_1h["ema_200_1h"]).astype(int)
            dataframe["trend_1h"] = dataframe["trend_1h"].reindex(dataframe.index, method="ffill")

        # 获取 4h 数据
        dataframe_4h = self.dp.get_pair_dataframe(pair, "4h")
        if dataframe_4h is not None and len(dataframe_4h) > 30:
            dataframe_4h["adx_4h"] = ta.ADX(dataframe_4h, timeperiod=14)
            dataframe["adx_4h"] = dataframe_4h["adx_4h"].reindex(dataframe.index, method="ffill")
            dataframe["adx_strong_4h"] = (dataframe["adx_4h"] > 25).astype(int)

        # 当前时间帧指标
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20)

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 做多条件：
        # 1. 4h 趋势向上 (ADX > 25)
        # 2. 1h 趋势向上 (EMA50 > EMA200)
        # 3. 15m RSI 超卖
        dataframe.loc[
            (
                (dataframe["adx_strong_4h"] == 1) &
                (dataframe["trend_1h"] == 1) &
                (dataframe["rsi"] < 30) &
                (dataframe["volume"] > 0)
            ),
            ["enter_long", "enter_tag"]
        ] = (1, "multi_tf_bull_entry")

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 平多：4h ADX 转弱 或 1h 趋势反转
        dataframe.loc[
            (
                (dataframe["adx_strong_4h"] == 0) |
                (dataframe["rsi"] > 70)
            ),
            "exit_long"
        ] = 1

        return dataframe

多时间帧对齐方式

动态参数调整

基于市场状态的参数切换

from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter
from pandas import DataFrame
import talib.abstract as ta

class AdaptiveStrategy(IStrategy):
    """
    自适应策略：根据市场波动性动态调整参数
    """
    timeframe = "1h"
    startup_candle_count = 100

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 基础指标
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
        dataframe["bb_upper"], _, dataframe["bb_lower"] = ta.BBANDS(dataframe, timeperiod=20)

        # 计算市场状态
        # 1. 波动性指标
        dataframe["volatility"] = dataframe["atr"] / dataframe["close"]

        # 2. 趋势强度
        dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)

        # 3. 识别市场状态
        dataframe["high_volatility"] = (dataframe["volatility"] > dataframe["volatility"].rolling(50).mean() * 1.5).astype(int)
        dataframe["strong_trend"] = (dataframe["adx"] > 25).astype(int)

        # 4. 动态 RSI 阈值
        # 高波动时放宽 RSI 条件
        dataframe["dynamic_rsi_low"] = dataframe["high_volatility"].apply(
            lambda x: 25 if x else 30
        )
        dataframe["dynamic_rsi_high"] = dataframe["high_volatility"].apply(
            lambda x: 75 if x else 70
        )

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 使用动态 RSI 阈值
        dataframe.loc[
            (
                (dataframe["rsi"] < dataframe["dynamic_rsi_low"]) &
                (dataframe["volume"] > 0)
            ),
            "enter_long"
        ] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[
            (
                (dataframe["rsi"] > dataframe["dynamic_rsi_high"]) &
                (dataframe["volume"] > 0)
            ),
            "exit_long"
        ] = 1

        return dataframe

动态仓位大小

from freqtrade.strategy import IStrategy

class DynamicPositionStrategy(IStrategy):
    """
    根据市场信心动态调整仓位
    """
    def custom_stake_amount(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_stake: float,
        min_stake: float,
        max_stake: float,
        entry_tag: Optional[str],
        side: str,
        **kwargs
    ) -> float:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

        if dataframe is None or len(dataframe) < 20:
            return proposed_stake

        last_candle = dataframe.iloc[-1]

        # 计算信心分数 (0-1)
        confidence = 0.5

        # RSI 越极端，信心越高
        rsi = last_candle.get("rsi", 50)
        if rsi < 25:
            confidence += 0.2
        elif rsi < 20:
            confidence += 0.3

        # 趋势越强，信心越高
        adx = last_candle.get("adx", 0)
        if adx > 30:
            confidence += 0.2
        elif adx > 40:
            confidence += 0.3

        # 大成交量增加信心
        volume_ratio = last_candle.get("volume_ratio", 1)
        if volume_ratio > 1.5:
            confidence += 0.1

        # 限制信心范围
        confidence = min(max(confidence, 0.1), 1.0)

        # 返回调整后的仓位
        return proposed_stake * confidence

机器学习信号融合

集成外部预测模型

from freqtrade.strategy import IStrategy
from pandas import DataFrame
import numpy as np
import joblib

class MLAugmentedStrategy(IStrategy):
    """
    使用预训练的 ML 模型生成辅助信号
    """
    timeframe = "1h"
    startup_candle_count = 100

    def __init__(self, config: dict) -> None:
        super().__init__(config)
        # 加载预训练模型
        try:
            self.model = joblib.load("user_data/models/signal_model.pkl")
            self.model_loaded = True
        except Exception:
            self.model_loaded = False
            self.logger.warning("ML 模型加载失败，将使用纯技术指标策略")

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 计算技术指标
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
        dataframe["sma_20"] = ta.SMA(dataframe, timeperiod=20)
        dataframe["sma_50"] = ta.SMA(dataframe, timeperiod=50)

        # 计算价格变化特征
        dataframe["pct_5"] = dataframe["close"].pct_change(5)
        dataframe["pct_10"] = dataframe["close"].pct_change(10)

        # 生成 ML 预测信号
        if self.model_loaded and len(dataframe) > 30:
            # 准备特征矩阵
            feature_cols = ["rsi", "adx", "atr", "pct_5", "pct_10"]
            features = dataframe[feature_cols].fillna(0).values[-1:]

            # 预测
            try:
                ml_pred = self.model.predict(features)[0]
                ml_prob = self.model.predict_proba(features)[0][1]

                # 将 ML 信号添加到当前 K 线
                dataframe.loc[dataframe.index[-1], "ml_signal"] = ml_pred
                dataframe.loc[dataframe.index[-1], "ml_probability"] = ml_prob
            except Exception as e:
                self.logger.error(f"ML 预测失败: {e}")

        # 填充 ML 信号
        if "ml_signal" not in dataframe.columns:
            dataframe["ml_signal"] = 0
        dataframe["ml_signal"] = dataframe["ml_signal"].fillna(0).astype(int)

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 技术指标信号
        tech_signal = (
            (dataframe["rsi"] < 30) &
            (dataframe["adx"] > 20)
        )

        # 技术指标 + ML 信号双重确认
        dataframe.loc[
            (
                tech_signal &
                (dataframe["ml_signal"] == 1)
            ),
            ["enter_long", "enter_tag"]
        ] = (1, "ml_confirmed_entry")

        # 仅技术指标（当 ML 不可用时）
        if not self.model_loaded:
            dataframe.loc[tech_signal, ["enter_long", "enter_tag"]] = (1, "tech_only_entry")

        return dataframe

自定义指标组合

复合指标构造

def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    # 1. 综合动量指标 (CMI)
    rsi_norm = (dataframe["rsi"] - 50) / 50  # 标准化 RSI 到 [-1, 1]
    macd_norm = dataframe["macd"] / dataframe["close"]  # 归一化 MACD
    dataframe["composite_momentum"] = (
        0.4 * rsi_norm +
        0.3 * macd_norm +
        0.3 * (dataframe["adx"] - 25) / 25
    )

    # 2. 趋势强度评分
    trend_up = (dataframe["ema_20"] > dataframe["ema_50"]).astype(int)
    trend_strong = (dataframe["adx"] > 25).astype(int)
    price_above_sma = (dataframe["close"] > dataframe["sma_50"]).astype(int)
    dataframe["trend_score"] = trend_up + trend_strong + price_above_sma  # 0-3

    # 3. 波动调整的动量
    dataframe["volatility_adjusted_momentum"] = (
        dataframe["close"].pct_change(10) / dataframe["atr"]
    )

    # 4. 市场状态指数
    dataframe["market_regime"] = np.where(
        dataframe["trend_score"] >= 2,
        1,  # 趋势市
        np.where(dataframe["adx"] < 20, -1, 0)  # -1: 震荡市, 0: 模糊
    )

    return dataframe

自定义策略评分系统

def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    # 多因子评分系统
    score = 0

    # 因子 1: RSI 信号
    score += np.where(dataframe["rsi"] < 25, 2, 0)
    score += np.where((dataframe["rsi"] >= 25) & (dataframe["rsi"] < 30), 1, 0)

    # 因子 2: 趋势确认
    score += np.where(dataframe["adx"] > 25, 1, 0)
    score += np.where(dataframe["adx"] > 30, 1, 0)

    # 因子 3: 成交量确认
    score += np.where(dataframe["volume_ratio"] > 1.5, 1, 0)
    score += np.where(dataframe["volume_ratio"] > 2.0, 1, 0)

    # 因子 4: 价格位置
    score += np.where(dataframe["close"] < dataframe["bb_lower"], 2, 0)
    score += np.where(
        (dataframe["close"] >= dataframe["bb_lower"]) &
        (dataframe["close"] < dataframe["sma_20"]),
        1, 0
    )

    # 评分 >= 5 时入场
    dataframe["strategy_score"] = score
    dataframe.loc[
        (dataframe["strategy_score"] >= 5) & (dataframe["volume"] > 0),
        ["enter_long", "enter_tag"]
    ] = (1, "score_entry")

    return dataframe

策略组合

信号投票机制

class VotingStrategy(IStrategy):
    """
    多策略投票组合：
    在每个 K 线周期运行多个子策略，使用投票机制决定最终信号
    """
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 策略 1: RSI 均值回归信号
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["signal_rsi"] = np.where(dataframe["rsi"] < 30, 1, 0)

        # 策略 2: MACD 金叉信号
        macd = ta.MACD(dataframe)
        dataframe["signal_macd"] = np.where(
            (macd["macd"] > macd["macdsignal"]) &
            (macd["macd"].shift(1) <= macd["macdsignal"].shift(1)),
            1, 0
        )

        # 策略 3: 布林带突破信号
        bb = ta.BBANDS(dataframe, timeperiod=20)
        dataframe["signal_bb"] = np.where(
            dataframe["close"] < bb["lowerband"],
            1, 0
        )

        # 策略 4: 趋势跟踪信号
        dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20)
        dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
        dataframe["signal_trend"] = np.where(
            (dataframe["ema_20"] > dataframe["ema_50"]) &
            (dataframe["ema_20"].shift(1) <= dataframe["ema_50"].shift(1)),
            1, 0
        )

        # 信号投票
        dataframe["vote_count"] = (
            dataframe["signal_rsi"] +
            dataframe["signal_macd"] +
            dataframe["signal_bb"] +
            dataframe["signal_trend"]
        )

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 至少 2 个策略同意时入场
        dataframe.loc[
            (dataframe["vote_count"] >= 2) & (dataframe["volume"] > 0),
            ["enter_long", "enter_tag"]
        ] = (1, "voted_entry")

        return dataframe

策略切换机制

class RegimeSwitchingStrategy(IStrategy):
    """
    市场状态切换策略：
    根据市场状态（趋势/震荡/高波动）切换到最合适的子策略
    """
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 判断市场状态
        dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)
        dataframe["atr_pct"] = ta.ATR(dataframe, timeperiod=14) / dataframe["close"]

        # 趋势市 (ADX > 25)
        dataframe["regime_trend"] = (dataframe["adx"] > 25).astype(int)

        # 高波动 (ATR% > 3%)
        dataframe["regime_volatile"] = (dataframe["atr_pct"] > 0.03).astype(int)

        # 震荡市 (ADX < 20)
        dataframe["regime_ranging"] = (dataframe["adx"] < 20).astype(int)

        # 趋势市策略：使用 MACD 趋势跟踪
        macd = ta.MACD(dataframe)
        dataframe["macd_line"] = macd["macd"]
        dataframe["macd_signal"] = macd["macdsignal"]

        # 震荡市策略：使用 RSI 均值回归
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)

        # 高波动策略：使用布林带
        bb = ta.BBANDS(dataframe, timeperiod=20)
        dataframe["bb_upper"] = bb["upperband"]
        dataframe["bb_lower"] = bb["lowerband"]

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 趋势市：MACD 金叉
        trend_entry = (
            (dataframe["regime_trend"] == 1) &
            (dataframe["macd_line"] > dataframe["macd_signal"])
        )

        # 震荡市：RSI 超卖
        ranging_entry = (
            (dataframe["regime_ranging"] == 1) &
            (dataframe["rsi"] < 30)
        )

        # 高波动：布林下轨反弹
        volatile_entry = (
            (dataframe["regime_volatile"] == 1) &
            (dataframe["close"] < dataframe["bb_lower"])
        )

        # 合并信号
        dataframe.loc[
            (trend_entry | ranging_entry | volatile_entry) &
            (dataframe["volume"] > 0),
            "enter_long"
        ] = 1

        return dataframe

策略回测增强

Walk-Forward 分析

# Walk-Forward 优化方法
# 1. 将数据分成多个时间窗口
# 2. 在每个窗口上训练并验证

# 窗口 1: 训练 2022-01 ~ 2022-06, 验证 2022-07 ~ 2022-09
freqtrade hyperopt --timerange 20220101-20220630 --epochs 200
freqtrade backtesting --timerange 20220701-20220930

# 窗口 2: 训练 2022-07 ~ 2022-12, 验证 2023-01 ~ 2023-03
freqtrade hyperopt --timerange 20220701-20221231 --epochs 200
freqtrade backtesting --timerange 20230101-20230331

# 对比各窗口结果是否一致

高级策略开发原则

1. 模块化设计：将指标计算、信号生成、风控逻辑分离
2. 参数可调：所有关键参数设计为可调参数
3. 适应性：策略能自动适应不同市场环境
4. 鲁棒性：在多种市场条件下都有不错表现
5. 可解释性：每个信号都应该有明确的逻辑解释