Day 1 — Backtrader 深入：Cerebro / Analyzer / Observer

# day1_backtrader_analyzer_demo.py
import backtrader as bt

# --- 基础策略（可复用） ---
class TestStrategy(bt.Strategy):
    params = dict(
        fast=10,
        slow=20,
    )

    def __init__(self):
        self.sma_fast = bt.ind.SMA(period=self.p.fast)
        self.sma_slow = bt.ind.SMA(period=self.p.slow)

    def next(self):
        if not self.position:
            if self.sma_fast > self.sma_slow:
                self.buy()
        else:
            if self.sma_fast < self.sma_slow:
                self.sell()

# --- Loader ---
def load_data():
    data = bt.feeds.YahooFinanceCSVData(
        dataname='data/BTC_USDT.csv',
        fromdate=datetime(2021,1,1),
        todate=datetime(2023,12,31),
    )
    return data

# --- Main ---
if __name__ == "__main__":
    cerebro = bt.Cerebro()

    cerebro.addstrategy(TestStrategy)
    cerebro.adddata(load_data())

    # 添加分析器 analyzer
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='dd')

    result = cerebro.run()
    strat = result[0]

    print("Sharpe Ratio:", strat.analyzers.sharpe.get_analysis())
    print("Max Drawdown:", strat.analyzers.dd.get_analysis())

    cerebro.plot()

📘 Day 2 — 实作 RSI+SMA 混合策略

# day2_rsi_sma_strategy.py
import backtrader as bt

class RSISMAStrategy(bt.Strategy):
    params = dict(
        sma_period=20,
        rsi_period=14,
        rsi_buy=30,
        rsi_sell=70,
    )

    def __init__(self):
        self.sma = bt.ind.SMA(period=self.p.sma_period)
        self.rsi = bt.ind.RSI(period=self.p.rsi_period)

    def next(self):
        # Buy signal: price > SMA AND RSI < 30
        if not self.position:
            if self.data.close > self.sma and self.rsi < self.p.rsi_buy:
                self.buy()

        else:
            if self.data.close < self.sma or self.rsi > self.p.rsi_sell:
                self.sell()

# ——— 数据加载 & 回测略（与昨日相同） ———

📘 Day 3 — 手动参数搜索（10 组）

# day3_manual_param_search.py
import backtrader as bt
import pandas as pd

from strategy_rsi_sma import RSISMAStrategy

param_grid = [
    {"sma_period": 10, "rsi_period": 14},
    {"sma_period": 20, "rsi_period": 14},
    {"sma_period": 30, "rsi_period": 14},
    {"sma_period": 10, "rsi_period": 7},
    {"sma_period": 20, "rsi_period": 7},
    {"sma_period": 30, "rsi_period": 7},
    {"sma_period": 15, "rsi_period": 10},
    {"sma_period": 25, "rsi_period": 10},
    {"sma_period": 35, "rsi_period": 10},
    {"sma_period": 50, "rsi_period": 14},
]

results = []

def run_backtest(params):
    cerebro = bt.Cerebro()
    cerebro.adddata(bt.feeds.YahooFinanceCSVData(dataname="data/BTC_USDT.csv"))
    cerebro.addstrategy(RSISMAStrategy, **params)
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='dd')

    res = cerebro.run()[0]
    sharpe = res.analyzers.sharpe.get_analysis().get('sharperatio', 0)
    maxdd = res.analyzers.dd.get_analysis()['max']['drawdown']

    return sharpe, maxdd

for p in param_grid:
    sharpe, maxdd = run_backtest(p)
    results.append([p['sma_period'], p['rsi_period'], sharpe, maxdd])

df = pd.DataFrame(results, columns=["SMA", "RSI", "Sharpe", "MaxDD"])
print(df)

df.to_csv("week3/manual_param_search.csv", index=False)

📘 Day 4 — Optuna 自动优化示例

# day4_optuna_optimize.py
import optuna
import backtrader as bt
import pandas as pd

from strategy_rsi_sma import RSISMAStrategy

def objective(trial):
    # 搜索范围
    sma_period = trial.suggest_int("sma", 5, 50)
    rsi_period = trial.suggest_int("rsi", 5, 20)

    cerebro = bt.Cerebro()
    cerebro.adddata(bt.feeds.YahooFinanceCSVData(dataname="data/BTC_USDT.csv"))

    cerebro.addstrategy(
        RSISMAStrategy,
        sma_period=sma_period,
        rsi_period=rsi_period,
    )

    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
    result = cerebro.run()[0]

    sharpe = result.analyzers.sharpe.get_analysis().get('sharperatio', -1)
    return sharpe

# 运行优化
study = optuna.create_study(direction="maximize")
study.optimize(objective, n_trials=50)

print("最佳参数:", study.best_params)
print("最佳 Sharpe:", study.best_value)

# 保存可视化图
fig = optuna.visualization.plot_optimization_history(study)
fig.write_image("week3/optuna_history.png")

📘 Day 5 — 添加交易费用与滑点

# day5_commission_slippage.py
import backtrader as bt

class RSISMAStrategy(bt.Strategy):
    # ... 与前一致 ...

cerebro = bt.Cerebro()

data = bt.feeds.YahooFinanceCSVData(dataname="data/BTC_USDT.csv")
cerebro.adddata(data)

# 关键：加入交易费 & 滑点
cerebro.broker.setcommission(commission=0.0004)  # 交易费 0.04%
cerebro.broker.set_slippage_perc(0.0005)  # 滑点 0.05%

cerebro.addstrategy(RSISMAStrategy)
cerebro.run()

📘 Day 6 — 写成脚本 run_backtest.py

# run_backtest.py
import json
import backtrader as bt
from strategy_rsi_sma import RSISMAStrategy

def run(config_path):
    with open(config_path, 'r') as f:
        cfg = json.load(f)

    cerebro = bt.Cerebro()

    data = bt.feeds.YahooFinanceCSVData(dataname=cfg["data"])
    cerebro.adddata(data)

    cerebro.broker.setcommission(commission=cfg["commission"])
    cerebro.broker.set_slippage_perc(cfg["slippage"])

    cerebro.addstrategy(
        RSISMAStrategy,
        sma_period=cfg["sma"],
        rsi_period=cfg["rsi"],
    )

    result = cerebro.run()
    cerebro.plot()

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", type=str, default="config.json")
    args = parser.parse_args()

    run(args.config)

config.json 模板：

{
    "data": "data/BTC_USDT.csv",
    "commission": 0.0004,
    "slippage": 0.0005,
    "sma": 20,
    "rsi": 14
}