Day 1 — 趋势跟随 vs 均值回归 + 逻辑图

# day1_flowchart_template.ipynb （示意代码）
# —— 这里主要是画流程图，给你可执行模板 ——

from graphviz import Digraph

dot = Digraph(comment='Strategy Logic')

dot.node('A', 'Start')
dot.node('B', 'Fetch OHLCV')
dot.node('C', 'Compute Indicators (SMA/RSI etc)')
dot.node('D', 'Check Trend / Mean Reversion')
dot.node('E', 'Generate Signal')
dot.node('F', 'Risk Control')
dot.node('G', 'Execute Order')

dot.edges([
    ('A','B'),
    ('B','C'),
    ('C','D'),
    ('D','E'),
    ('E','F'),
    ('F','G'),
])

dot.render('week4/strategy_flowchart', format='png')

📘 Day 2 — 策略伪代码 → Backtrader 骨架

🧠 伪代码模板（你可改写）

# day2_pseudocode.py

"""
伪代码示例（趋势策略）：

IF price > SMA(50):
    IF RSI < 30 AND no position:
        BUY
IF price < SMA(50):
    IF RSI > 70 AND position exists:
        SELL

加入风控：
- 每单 max 5% 仓位
- 最大总仓位 20%
- 单日亏损 > 3% 则停止交易（flat）
"""

Backtrader 策略骨架

# day2_bt_skeleton.py
import backtrader as bt

class StrategyV1(bt.Strategy):
    params = dict(
        sma_period=50,
        rsi_period=14,
        pos_size=0.05,        # 单笔仓位
        max_total_exposure=0.2,
        max_daily_loss=0.03,
    )

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

    def prenext(self): self.next()

    def next(self):
        # —— 日损失计算（风控会在 day4 完整写） ——
        if self.day_start_value is None:
            self.day_start_value = self.broker.getvalue()

        # —— 信号逻辑占位 ——
        # 你可以在 Day3、4 完善
        pass

📘 Day 3 — 回测、性能指标、调参

# day3_backtest_base.py
import backtrader as bt
from strategy_v1 import StrategyV1

cerebro = bt.Cerebro()

data = bt.feeds.YahooFinanceCSVData(
    dataname='data/BTC_USDT.csv',
    fromdate=datetime(2023,1,1),
    todate=datetime(2024,12,31)
)

cerebro.adddata(data)
cerebro.addstrategy(StrategyV1)

# 分析器
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe")
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="dd")
cerebro.addanalyzer(bt.analyzers.Returns, _name="returns")

result = cerebro.run()[0]

print("Sharpe:", result.analyzers.sharpe.get_analysis())
print("Drawdown:", result.analyzers.dd.get_analysis())
print("Returns:", result.analyzers.returns.get_analysis())

cerebro.plot()

📘 Day 4 — 基本风控

# day4_risk_control_add.py
import backtrader as bt

class StrategyV1(bt.Strategy):
    params = dict(
        sma_period=50,
        rsi_period=14,
        pos_size=0.05,
        max_total_exposure=0.2,
        max_daily_loss=0.03,
    )

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

    def next(self):
        # 初始化当天起始资金
        if self.day_start_value is None or self.data.datetime.date(0) != self.data.datetime.date(-1):
            self.day_start_value = self.broker.getvalue()

        # --- 风控 1：单日最大亏损 ---
        cur_value = self.broker.getvalue()
        if (cur_value - self.day_start_value) / self.day_start_value < -self.p.max_daily_loss:
            if self.position:
                self.close()
            return  # 停止交易

        # --- 风控 2：总仓位限制 ---
        total_value = self.broker.getvalue()
        current_exposure = abs(self.position.size * self.data.close[0]) / total_value

        if current_exposure > self.p.max_total_exposure:
            return  # 暂停开仓

        # -------- 信号逻辑占位（例如 SMA + RSI） --------
        if not self.position:
            if self.data.close > self.sma and self.rsi < 30:
                self.buy(size=self.p.pos_size * total_value / self.data.close)

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

📘 Day 5 — 多周期对比

# day5_multi_timeframe_test.py
import backtrader as bt
from strategy_v1 import StrategyV1

periods = ["1h", "4h", "1d"]

results = []

for p in periods:
    cerebro = bt.Cerebro()
    data = bt.feeds.GenericCSVData(
        dataname=f"data/BTC_{p}.csv",
        dtformat=('%Y-%m-%d %H:%M:%S'),
        timeframe=bt.TimeFrame.Minutes if p in ['1h','4h'] else bt.TimeFrame.Days,
        compression=60 if p=='1h' else 240 if p=='4h' else 1
    )

    cerebro.adddata(data)
    cerebro.addstrategy(StrategyV1)
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe")
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name="dd")

    res = cerebro.run()[0]

    results.append([
        p,
        res.analyzers.sharpe.get_analysis().get('sharperatio', 0),
        res.analyzers.dd.get_analysis()['max']['drawdown']
    ])

# 输出 DataFrame
import pandas as pd
df = pd.DataFrame(results, columns=["Timeframe","Sharpe","MaxDD"])
df.to_csv("week4/timeframe_compare.csv", index=False)
print(df)

📘 Day 6 — 策略文档模板 + 打包结构

文档模板