Skip to content

多时间框架策略

多时间框架(Multi-Timeframe, MTF)策略是一种使用不同时间周期的数据来做出交易决策的高级策略。通过结合长周期趋势和短周期信号,可以提高交易决策的准确性。

核心概念

什么是多时间框架策略

多时间框架策略同时分析多个时间周期的数据:

  • 长周期:判断主要趋势方向(如 1小时、4小时、日线)
  • 短周期:寻找具体入场时机(如 5分钟、15分钟)

优势

  • 减少假信号
  • 提高胜率
  • 更好的风险收益比
  • 趋势确认机制

Freqtrade 实现方法

方法一:使用 informative_pairs

这是 Freqtrade 推荐的多时间框架实现方式。

python
import numpy as np
import pandas as pd
from pandas import DataFrame
from datetime import datetime, timedelta

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


class MultiTimeframeStrategy(IStrategy):
    """
    多时间框架策略示例:
    - 使用 1小时 K线判断大趋势
    - 使用 5分钟 K线寻找入场时机
    - RSI + MACD 组合信号
    """
    INTERFACE_VERSION = 3

    # 主时间框架(用于交易执行)
    timeframe = "5m"
    
    # 启用做空
    can_short: bool = True

    # ROI 和止损设置
    minimal_roi = {
        "0": 0.08,  # 8% 收益
        "20": 0.05, # 20分钟后 5% 收益
        "40": 0.02  # 40分钟后 2% 收益
    }
    
    stoploss = -0.04  # 4% 止损
    trailing_stop = True
    trailing_stop_positive = 0.01
    trailing_stop_positive_offset = 0.02

    # 只处理新K线
    process_only_new_candles = True

    # 使用退出信号
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    # 启动需要的K线数量
    startup_candle_count: int = 200

    # 订单类型
    order_types = {
        "entry": "limit",
        "exit": "limit",
        "stoploss": "market",
        "stoploss_on_exchange": False,
        "emergency_exit": "market",
        "force_exit": "market",
    }

    def informative_pairs(self):
        """
        定义需要获取的额外时间框架数据
        """
        # 获取 1小时数据用于趋势判断
        pairs = []
        for pair in self.dp.current_whitelist():
            pairs.append((pair, "1h", self.dp.get_pair_dataframe(pair, "1h")))
        return pairs

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        主时间框架(5分钟)指标计算
        """
        # RSI
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["rsi_fast"] = ta.RSI(dataframe, timeperiod=7)
        
        # MACD
        macd = ta.MACD(dataframe)
        dataframe["macd"] = macd['macd']
        dataframe["macd_signal"] = macd['macdsignal']
        dataframe["macd_hist"] = macd['macdhist']
        
        # 移动平均线
        dataframe["sma_20"] = ta.SMA(dataframe, timeperiod=20)
        dataframe["sma_50"] = ta.SMA(dataframe, timeperiod=50)
        dataframe["ema_9"] = ta.EMA(dataframe, timeperiod=9)
        
        # 布林带
        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"]
        
        return dataframe

    def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        1小时时间框架指标计算(用于趋势判断)
        """
        # 1小时 RSI
        dataframe["rsi_1h"] = ta.RSI(dataframe, timeperiod=14)
        
        # 1小时 MACD
        macd_1h = ta.MACD(dataframe)
        dataframe["macd_1h"] = macd_1h['macd']
        dataframe["macd_signal_1h"] = macd_1h['macdsignal']
        
        # 1小时 移动平均线
        dataframe["sma_20_1h"] = ta.SMA(dataframe, timeperiod=20)
        dataframe["sma_50_1h"] = ta.SMA(dataframe, timeperiod=50)
        dataframe["ema_200_1h"] = ta.EMA(dataframe, timeperiod=200)
        
        # 1小时 趋势判断
        dataframe["trend_1h"] = 0
        dataframe.loc[dataframe["sma_20_1h"] > dataframe["sma_50_1h"], "trend_1h"] = 1  # 上升趋势
        dataframe.loc[dataframe["sma_20_1h"] < dataframe["sma_50_1h"], "trend_1h"] = -1  # 下降趋势
        
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        入场信号:结合多时间框架信息
        """
        # 获取 1小时数据
        informative = self.dp.get_pair_dataframe(metadata["pair"], "1h")
        informative_1h = self.populate_indicators_1h(informative, metadata)
        
        # 将 1小时数据合并到 5分钟数据
        dataframe = merge_informative_pair(dataframe, informative_1h, "1h", ffill=True)
        
        # 重置信号列
        dataframe["enter_long"] = 0
        dataframe["enter_short"] = 0
        
        # 做多入场条件
        long_conditions = (
            # 1小时趋势:必须为上升趋势
            (dataframe["trend_1h_1h"] == 1) &
            # 1小时 RSI 不超买
            (dataframe["rsi_1h_1h"] < 70) &
            # 5分钟 RSI 超卖
            (dataframe["rsi"] < 30) &
            # 5分钟 MACD 金叉
            (dataframe["macd"] > dataframe["macd_signal"]) &
            (dataframe["macd"].shift(1) <= dataframe["macd_signal"].shift(1)) &
            # 价格接近布林带下轨
            (dataframe["close"] < dataframe["bb_lowerband"] * 1.02) &
            # 成交量确认
            (dataframe["volume"] > dataframe["volume"].rolling(20).mean() * 0.8)
        )
        
        # 做空入场条件
        short_conditions = (
            # 1小时趋势:必须为下降趋势
            (dataframe["trend_1h_1h"] == -1) &
            # 1小时 RSI 不超卖
            (dataframe["rsi_1h_1h"] > 30) &
            # 5分钟 RSI 超买
            (dataframe["rsi"] > 70) &
            # 5分钟 MACD 死叉
            (dataframe["macd"] < dataframe["macd_signal"]) &
            (dataframe["macd"].shift(1) >= dataframe["macd_signal"].shift(1)) &
            # 价格接近布林带上轨
            (dataframe["close"] > dataframe["bb_upperband"] * 0.98) &
            # 成交量确认
            (dataframe["volume"] > dataframe["volume"].rolling(20).mean() * 0.8)
        )
        
        # 设置入场信号
        dataframe.loc[long_conditions, "enter_long"] = 1
        dataframe.loc[short_conditions, "enter_short"] = 1
        
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        出场信号:结合多时间框架信息
        """
        # 重置信号列
        dataframe["exit_long"] = 0
        dataframe["exit_short"] = 0
        
        # 做多出场条件
        exit_long_conditions = (
            # 5分钟 RSI 超买
            (dataframe["rsi"] > 75) |
            # MACD 死叉
            ((dataframe["macd"] < dataframe["macd_signal"]) & 
             (dataframe["macd"].shift(1) >= dataframe["macd_signal"].shift(1))) |
            # 价格跌破短期均线
            (dataframe["close"] < dataframe["ema_9"] * 0.98)
        )
        
        # 做空出场条件
        exit_short_conditions = (
            # 5分钟 RSI 超卖
            (dataframe["rsi"] < 25) |
            # MACD 金叉
            ((dataframe["macd"] > dataframe["macd_signal"]) & 
             (dataframe["macd"].shift(1) <= dataframe["macd_signal"].shift(1))) |
            # 价格突破短期均线
            (dataframe["close"] > dataframe["ema_9"] * 1.02)
        )
        
        # 设置出场信号
        dataframe.loc[exit_long_conditions, "exit_long"] = 1
        dataframe.loc[exit_short_conditions, "exit_short"] = 1
        
        return dataframe

    @property
    def plot_config(self):
        return {
            "main_plot": {
                "sma_20": {"color": "blue"},
                "sma_50": {"color": "orange"},
                "bb_upperband": {"color": "gray"},
                "bb_lowerband": {"color": "gray"},
            },
            "subplots": {
                "RSI": {
                    "rsi": {"color": "red"},
                    "rsi_fast": {"color": "orange"},
                },
                "MACD": {
                    "macd": {"color": "blue"},
                    "macd_signal": {"color": "red"},
                    "macd_hist": {"type": "bar", "color": "green"},
                }
            }
        }

方法二:手动获取多时间框架数据

如果需要更灵活的控制,可以手动获取多时间框架数据:

python
class AdvancedMultiTimeframeStrategy(IStrategy):
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 主时间框架指标
        dataframe["rsi_5m"] = ta.RSI(dataframe, timeperiod=14)
        
        # 手动获取 15分钟数据
        dataframe_15m = self.dp.get_pair_dataframe(metadata["pair"], "15m")
        dataframe_15m["rsi_15m"] = ta.RSI(dataframe_15m, timeperiod=14)
        
        # 手动获取 1小时数据
        dataframe_1h = self.dp.get_pair_dataframe(metadata["pair"], "1h")
        dataframe_1h["rsi_1h"] = ta.RSI(dataframe_1h, timeperiod=14)
        dataframe_1h["sma_trend_1h"] = ta.SMA(dataframe_1h, timeperiod=50)
        
        # 合并数据
        dataframe = dataframe.copy()
        dataframe["rsi_15m"] = dataframe_15m["rsi_15m"].reindex(dataframe.index, method='ffill')
        dataframe["rsi_1h"] = dataframe_1h["rsi_1h"].reindex(dataframe.index, method='ffill')
        dataframe["sma_trend_1h"] = dataframe_1h["sma_trend_1h"].reindex(dataframe.index, method='ffill')
        
        return dataframe

配置说明

config.json 设置

json
{
    "trading_mode": "futures",
    "margin_mode": "cross",
    "stake_currency": "USDT",
    "stake_amount": 100,
    "fee": 0.001,
    
    "timeframe": "5m",
    
    "dry_run": true,
    "dry_run_wallet": 1000,
    
    "exchange": {
        "name": "binance",
        "key": "your_api_key",
        "secret": "your_api_secret",
        "ccxt_config": {},
        "ccxt_async_config": {}
    },
    
    "pairlists": [
        {
            "method": "StaticPairList"
        }
    ],
    
    "telegram": {
        "enabled": false
    }
}

策略逻辑详解

数据流图

1小时数据 ←───── 趋势判断(EMA交叉、RSI位置)

5分钟数据 ────→ 入场时机(RSI极值、MACD交叉)

交易执行 ←───── 风险控制(止损、追踪止损)

趋势过滤机制

  1. 上升趋势确认

    • 1小时 SMA20 > SMA50
    • 1小时 RSI 在合理范围(30-70)
    • 只寻找做多机会
  2. 下降趋势确认

    • 1小时 SMA20 < SMA50
    • 1小时 RSI 在合理范围(30-70)
    • 只寻找做空机会

入场信号过滤

大趋势过滤 + 短周期信号 = 高质量入场信号

上升趋势中:
    ├─ RSI_5m < 30 (超卖)
    ├─ MACD_5m 金叉
    ├─ 价格接近布林带下轨
    └─ 成交量确认

优化建议

参数调优

  1. 时间框架组合

    • 快速策略:1m + 5m + 15m
    • 中等策略:5m + 15m + 1h
    • 长期策略:15m + 1h + 4h
  2. 指标参数

    • RSI 周期:7-21
    • MACD 参数:12,26,9(标准)或 5,35,5(敏感)
    • 移动平均线:20,50,200
  3. 过滤条件

    • 趋势强度要求
    • RSI 阈值调整
    • 成交量倍数设置

风险管理

  1. 动态止损

    • 基于ATR的止损
    • 波动率调整止损
    • 时间止损
  2. 仓位管理

    • 趋势强度调整仓位
    • 波动率调整仓位
    • 相关性分散

回测验证

回测命令

bash
freqtrade backtesting --strategy MultiTimeframeStrategy --timeframe 5m --timerange 20240101-20240301

性能指标

  • 总收益率
  • 夏普比率
  • 最大回撤
  • 胜率
  • 平均持仓时间
  • 年化收益率

实盘部署注意事项

  1. 数据同步:确保多时间框架数据正确同步
  2. 网络延迟:考虑数据获取延迟对策略影响
  3. 资金管理:合理分配仓位,避免过度杠杆
  4. 监控告警:设置策略异常监控
  5. 定期复盘:根据市场变化调整参数

常见问题

Q: 如何选择合适的时间框架组合?

A: 根据交易风格选择:

  • 短线交易:1m+5m+15m
  • 中线交易:5m+15m+1h
  • 长线交易:15m+1h+4h

Q: 多时间框架数据延迟怎么办?

A:

  • 使用 process_only_new_candles = True
  • 设置合理的 startup_candle_count
  • 考虑使用缓存机制

Q: 如何避免未来函数?

A:

  • 确保长周期数据不包含未来信息
  • 使用正确的数据对齐方法
  • 严格测试时间序列逻辑

Q: 策略性能不佳如何优化?

A:

  • 调整时间框架组合
  • 优化指标参数
  • 增加过滤条件
  • 改进风险管理