Freqtrade 策略开发基础

本章详细介绍 Freqtrade 策略开发的核心知识，涵盖 IStrategy 接口、指标定义、买卖信号逻辑、止损策略以及完整的自定义策略示例，帮助读者系统掌握策略开发技能。

策略文件结构

IStrategy 接口概述

Freqtrade 中的所有策略都必须继承 IStrategy 基类并实现其核心方法。一个标准的策略文件包含以下组成部分：

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

class MyStrategy(IStrategy):
    # --- 必填参数 ---
    timeframe = "5m"          # K 线周期
    can_short = False         # 是否允许做空

    # --- 可选参数 ---
    startup_candle_count = 20  # 预热 K 线数
    process_only_new_candles = True
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    # --- 止损设置 ---
    stoploss = -0.10  # -10%

    # --- ROI 表 ---
    minimal_roi = {
        "0": 0.10,     # 持仓后立即有 10% 利润则卖出
        "30": 0.05,    # 30 分钟后 5% 利润卖出
        "60": 0.01,    # 60 分钟后 1% 利润卖出
        "120": 0       # 120 分钟后保本卖出
    }

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 在此方法中定义技术指标
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 在此方法中定义买入信号
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 在此方法中定义卖出信号
        return dataframe

核心方法说明

策略参数定义

Freqtrade 支持在策略中定义可调参数，方便在回测和实盘中灵活调整：

class MyStrategy(IStrategy):
    # 可调参数：可以通过 --strategy-parameter 覆盖
    buy_rsi_threshold = IntParameter(20, 40, default=30, space="buy")
    sell_rsi_threshold = IntParameter(60, 80, default=70, space="sell")

    # 布尔参数
    use_volume_filter = BooleanParameter(default=True, space="buy")

    # 枚举参数
    entry_method = EnumParameter(
        ["rsi", "macd", "bb"],
        default="rsi",
        space="buy"
    )

指标定义与使用

常用 TA-Lib 指标

def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    # 趋势指标
    dataframe["sma_20"] = ta.SMA(dataframe, timeperiod=20)
    dataframe["ema_12"] = ta.EMA(dataframe, timeperiod=12)
    dataframe["ema_26"] = ta.EMA(dataframe, timeperiod=26)

    # MACD
    macd = ta.MACD(dataframe)
    dataframe["macd"] = macd["macd"]
    dataframe["macd_signal"] = macd["macdsignal"]
    dataframe["macd_hist"] = macd["macdhist"]

    # RSI
    dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)

    # 布林带
    bollinger = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2, nbdevdn=2)
    dataframe["bb_upper"] = bollinger["upperband"]
    dataframe["bb_middle"] = bollinger["middleband"]
    dataframe["bb_lower"] = bollinger["lowerband"]

    # ATR（平均真实波幅）
    dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)

    # ADX（平均趋向指数）
    dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)

    # OBV（能量潮）
    dataframe["obv"] = ta.OBV(dataframe)

    return dataframe

自定义指标计算

除了 TA-Lib 内置指标，你可以在策略中定义任意自定义指标：

def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    # 价格变化率
    dataframe["pct_change_1"] = dataframe["close"].pct_change(1)
    dataframe["pct_change_5"] = dataframe["close"].pct_change(5)

    # 成交量比率
    dataframe["volume_ma"] = dataframe["volume"].rolling(window=20).mean()
    dataframe["volume_ratio"] = dataframe["volume"] / dataframe["volume_ma"]

    # 价格通道
    dataframe["highest_20"] = dataframe["high"].rolling(20).max()
    dataframe["lowest_20"] = dataframe["low"].rolling(20).min()

    # 价差百分比
    dataframe["spread"] = (dataframe["high"] - dataframe["low"]) / dataframe["close"]

    # 自定义复合指标：RSI 与价格的背离
    dataframe["rsi_ma"] = dataframe["rsi"].rolling(5).mean()
    dataframe["price_ma"] = dataframe["close"].rolling(5).mean()

    return dataframe

DataProvider 获取外部数据

def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    # 获取其他交易对的数据
    btc_data = self.dp.get_pair_dataframe("BTC/USDT", self.timeframe)
    if btc_data is not None:
        dataframe["btc_rsi"] = btc_data["rsi"].reindex(dataframe.index, method="ffill")

    # 获取 ticker 数据
    ticker = self.dp.ticker()
    if ticker:
        dataframe["current_bid"] = ticker.get("bid", 0)
        dataframe["current_ask"] = ticker.get("ask", 0)

    return dataframe

买卖信号逻辑

买入信号 (Entry Signal)

在 populate_entry_trend 方法中设置 enter_long 或 enter_short 列为 1 表示触发信号：

def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    # 做多信号：RSI 超卖 + 价格触及布林带下轨
    dataframe.loc[
        (
            (dataframe["rsi"] < 30) &
            (dataframe["close"] < dataframe["bb_lower"]) &
            (dataframe["volume"] > 0)
        ),
        "enter_long"
    ] = 1

    # 做空信号（需 can_short = True）
    dataframe.loc[
        (
            (dataframe["rsi"] > 70) &
            (dataframe["close"] > dataframe["bb_upper"]) &
            (dataframe["volume"] > 0)
        ),
        "enter_short"
    ] = 1

    return dataframe

卖出信号 (Exit Signal)

在 populate_exit_trend 方法中设置 exit_long 或 exit_short 列：

def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    # 做多平仓：RSI 超买
    dataframe.loc[
        (
            (dataframe["rsi"] > 70) &
            (dataframe["close"] > dataframe["bb_upper"])
        ),
        "exit_long"
    ] = 1

    # 做空平仓
    dataframe.loc[
        (
            (dataframe["rsi"] < 30) &
            (dataframe["close"] < dataframe["bb_lower"])
        ),
        "exit_short"
    ] = 1

    return dataframe

信号标签约定

信号确认回调

使用 confirm_trade_entry 和 confirm_trade_exit 做二次确认：

def confirm_trade_entry(
    self,
    pair: str,
    order_type: str,
    amount: float,
    rate: float,
    time_in_force: str,
    current_time: datetime,
    entry_tag: Optional[str],
    side: str,
    **kwargs
) -> bool:
    # 自定义逻辑：例如只在特定时间段交易
    if current_time.hour < 8 or current_time.hour > 20:
        return False
    return True

def confirm_trade_exit(
    self,
    pair: str,
    trade: Trade,
    order_type: str,
    amount: float,
    rate: float,
    time_in_force: str,
    exit_reason: str,
    current_time: datetime,
    **kwargs
) -> bool:
    # 可以在卖出前检查是否有更好的价格
    return True

止损策略

静态止损

最简单的止损方式，在策略中直接设置固定比例：

class MyStrategy(IStrategy):
    stoploss = -0.05  # -5%，当价格下跌 5% 时止损

自定义动态止损

通过 custom_stoploss 方法实现动态调整：

def custom_stoploss(
    self,
    pair: str,
    trade: Trade,
    current_time: datetime,
    current_rate: float,
    current_profit: float,
    after_fill: bool,
    **kwargs
) -> Optional[float]:
    # 初始止损 -10%
    if current_profit < 0:
        return -0.10

    # 当利润超过 5% 时，将止损移至保本
    if current_profit > 0.05:
        return -0.01  # -1%，接近保本

    # 当利润超过 10% 时，锁定至少 5% 的利润
    if current_profit > 0.10:
        return current_profit - 0.05

    # 使用 trailing stop
    return -0.10

三种止损模式对比

追踪止损配置

class MyStrategy(IStrategy):
    # 追踪止损配置
    trailing_stop = True
    trailing_stop_positive = -0.02       # 利润为正后，止损为 -2%
    trailing_stop_positive_offset = 0.03 # 利润超过 3% 时激活追踪
    trailing_only_offset_is_reached = True  # 仅在达到偏移量后启用追踪

完整策略示例

双均线交叉策略

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

class SmaCrossStrategy(IStrategy):
    """
    经典双均线交叉策略：
    - 短期均线上穿长期均线时买入
    - 短期均线下穿长期均线时卖出
    """
    # 基础配置
    timeframe = "1h"
    can_short = True
    startup_candle_count = 100

    # 止损
    stoploss = -0.08

    # ROI
    minimal_roi = {
        "0": 0.15,
        "60": 0.05,
        "120": 0.02,
        "240": 0
    }

    # 策略参数
    fast_ma_period = IntParameter(10, 30, default=20, space="buy")
    slow_ma_period = IntParameter(40, 60, default=50, space="sell")

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 计算快慢均线
        fast_period = self.fast_ma_period.value
        slow_period = self.slow_ma_period.value

        dataframe["fast_ma"] = ta.SMA(dataframe, timeperiod=fast_period)
        dataframe["slow_ma"] = ta.SMA(dataframe, timeperiod=slow_period)

        # 辅助指标
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 做多：快线上穿慢线，且 RSI 不过热
        dataframe.loc[
            (
                (dataframe["fast_ma"] > dataframe["slow_ma"]) &
                (dataframe["fast_ma"].shift(1) <= dataframe["slow_ma"].shift(1)) &
                (dataframe["rsi"] < 60)
            ),
            ["enter_long", "enter_tag"]
        ] = (1, "sma_cross_bull")

        # 做空：快线下穿慢线，且 RSI 不过冷
        dataframe.loc[
            (
                (dataframe["fast_ma"] < dataframe["slow_ma"]) &
                (dataframe["fast_ma"].shift(1) >= dataframe["slow_ma"].shift(1)) &
                (dataframe["rsi"] > 40)
            ),
            ["enter_short", "enter_tag"]
        ] = (1, "sma_cross_bear")

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 平多：快线下穿慢线
        dataframe.loc[
            (
                (dataframe["fast_ma"] < dataframe["slow_ma"]) &
                (dataframe["fast_ma"].shift(1) >= dataframe["slow_ma"].shift(1))
            ),
            ["exit_long", "exit_tag"]
        ] = (1, "exit_sma_cross")

        # 平空：快线上穿慢线
        dataframe.loc[
            (
                (dataframe["fast_ma"] > dataframe["slow_ma"]) &
                (dataframe["fast_ma"].shift(1) <= dataframe["slow_ma"].shift(1))
            ),
            ["exit_short", "exit_tag"]
        ] = (1, "exit_sma_cross")

        return dataframe

RSI 均值回归策略

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

class RsiReboundStrategy(IStrategy):
    """
    RSI 均值回归策略：
    - RSI 超卖时买入，回到中性时卖出
    - 结合布林带进行过滤
    """
    timeframe = "15m"
    can_short = True
    startup_candle_count = 40

    stoploss = -0.05

    minimal_roi = {
        "0": 0.03,
        "30": 0.01,
        "60": 0
    }

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)

        bb = ta.BBANDS(dataframe, timeperiod=20)
        dataframe["bb_upper"] = bb["upperband"]
        dataframe["bb_middle"] = bb["middleband"]
        dataframe["bb_lower"] = bb["lowerband"]

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 做多：RSI < 30 且价格接近布林下轨
        dataframe.loc[
            (
                (dataframe["rsi"] < 30) &
                (dataframe["close"] < dataframe["bb_lower"] * 1.01)
            ),
            "enter_long"
        ] = 1

        # 做空：RSI > 70 且价格接近布林上轨
        dataframe.loc[
            (
                (dataframe["rsi"] > 70) &
                (dataframe["close"] > dataframe["bb_upper"] * 0.99)
            ),
            "enter_short"
        ] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 平多：RSI 回到中性
        dataframe.loc[
            (dataframe["rsi"] > 50),
            "exit_long"
        ] = 1

        # 平空
        dataframe.loc[
            (dataframe["rsi"] < 50),
            "exit_short"
        ] = 1

        return dataframe

策略文件组织最佳实践

user_data/strategies/
├── __init__.py
├── MyStrategy.py
├── indicators.py          # 自定义指标函数
├── filters.py             # 过滤器函数
└── config.py              # 策略常量定义

调试技巧

# 在 populate_indicators 中添加日志输出
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)

    # 输出调试信息
    logger = self.logger
    last_rsi = dataframe["rsi"].iloc[-1]
    logger.info(f"Pair: {metadata['pair']}, Last RSI: {last_rsi:.2f}")

    return dataframe

常见陷阱与注意事项

1. 未来信息泄露：确保指标计算只使用截至当前周期的数据，不要使用未来数据
2. 数据对齐：使用 shift() 确保信号条件中的跨周期比较正确对齐
3. 预热 K 线：正确设置 startup_candle_count 确保指标有足够的计算历史
4. 信号闪烁：避免在 populate_entry_trend 中使用会对同一根 K 线反复变化的条件
5. 过度拟合：避免在策略中加入过多条件导致只在历史数据上表现良好