Day 1 — 多周期 merge（resample / agg）

import pandas as pd

# 假设已加载 1H 数据 df_1h
# 示例：把 1H 汇总成 4H
df_4h = df_1h.resample("4H").agg({
    "open": "first",
    "high": "max",
    "low": "min",
    "close": "last",
    "volume": "sum"
}).dropna()

# 合并 1H + 4H（右对齐）
merged = df_1h.join(df_4h, rsuffix="_4h", how="left").ffill()

merged.tail()

示意图（可在 Notebook 画）：

from IPython.display import Markdown
Markdown("""
**Multi-Timeframe Logic**
- Higher TF → 市场趋势方向 (4H)
- Lower TF → 精准入场 (1H)
""")

Day 2 — Backtrader 多周期实现

import backtrader as bt

class MTFStrategy(bt.Strategy):

    def __init__(self):
        # lower timeframe data0 (1H)
        self.data_close = self.datas[0].close

        # higher timeframe data1 (4H)
        self.data_close_4h = self.datas[1].close

    def next(self):
        # 过滤条件：4H 趋势向上
        trend_up = self.data_close_4h[0] > self.data_close_4h[-1]

        # 入场条件：1H 突破
        breakout = self.data_close[0] > max(self.data_close[-3:-1])

        if trend_up and breakout:
            self.buy()

# === 注册数据 ===
cerebro = bt.Cerebro()
data1h = bt.feeds.PandasData(dataname=df_1h)
data4h = bt.feeds.PandasData(dataname=df_4h)

cerebro.adddata(data1h)
cerebro.adddata(data4h)

cerebro.addstrategy(MTFStrategy)
cerebro.run()

Day 3 — 回测 & 对比单周期性能

# 回测结果
res = cerebro.run()[0]
cerebro.plot()

# 手动计算比较
perf = {
    "mtf_pnl": res.broker.getvalue(),
}

print(perf)

时间戳对齐检查：

print(df_1h.index[-5:])
print(df_4h.index[-5:])

Day 4 — 加入 volume surge + breakout filter

import numpy as np

df_1h['volume_surge'] = df_1h['volume'] > df_1h['volume'].rolling(20).mean() * 1.5
df_1h['breakout'] = df_1h['close'] > df_1h['high'].rolling(20).max()

# backtrader 用 indicator 版本（简单示例）
class MTFStrategy(bt.Strategy):

    def next(self):
        trend_up = self.data_close_4h[0] > self.data_close_4h[-1]
        breakout = self.data_close[0] > max(self.data_close[-20:-1])
        vol_surge = self.datas[0].volume[0] > bt.indicators.SMA(self.datas[0].volume, 20)[0] * 1.5

        if trend_up and breakout and vol_surge:
            self.buy()

Day 5 — 延迟 / 漏单模拟

import random

ORDER_DELAY = 2    # 2 根K线延迟
MISS_PROB = 0.1    # 10% 漏单概率

class DelayStrategy(bt.Strategy):

    def next(self):

        # 延迟触发
        if len(self) >= ORDER_DELAY:
            delayed_price = self.data_close[-ORDER_DELAY]
        else:
            delayed_price = self.data_close[0]

        # 漏单
        if random.random() < MISS_PROB:
            return

        # 示例：在突破时用延迟价格入场
        breakout = self.data_close[0] > max(self.data_close[-3:-1])
        if breakout:
            self.buy(price=delayed_price)

Day 6 — 策略模块化（插拔 timeframes）