量化投资全景
从数据到决策

import backtrader as bt
from datetime import datetime

class DualMAStrategy(bt.Strategy):
    """双均线交叉策略：快均线上穿慢均线买入(金叉)，下穿卖出(死叉)"""
    params = (
        ('fast_period', 10),      # 快均线周期
        ('slow_period', 30),      # 慢均线周期
        ('position_size', 0.95),  # 仓位比例
    )

    def __init__(self):
        # 计算短期简单移动平均线
        self.fast_sma = bt.indicators.SMA(
            self.data.close, period=self.p.fast_period)
        # 计算长期简单移动平均线
        self.slow_sma = bt.indicators.SMA(
            self.data.close, period=self.p.slow_period)
        # 交叉信号：>0为金叉，<0为死叉
        self.crossover = bt.indicators.CrossOver(
            self.fast_sma, self.slow_sma)
        self.order = None

    def next(self):
        if self.order:
            return
        if not self.position:
            # 无持仓时：金叉买入
            if self.crossover > 0:
                self.order = self.buy(size=self.p.position_size)
        else:
            # 有持仓时：死叉卖出
            if self.crossover < 0:
                self.order = self.close()

# ===== 回测配置 =====
cerebro = bt.Cerebro()
cerebro.addstrategy(DualMAStrategy)
data = bt.feeds.YahooFinanceData(dataname='AAPL',
    fromdate=datetime(2020, 1, 1),
    todate=datetime(2024, 12, 31))
cerebro.adddata(data)
cerebro.broker.setcash(100000.0)
cerebro.broker.setcommission(commission=0.001)
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
results = cerebro.run()

import backtrader as bt

class MomentumStrategy(bt.Strategy):
    """动量策略: MACD金叉 + RSI强势区间(50-70)时买入"""
    params = (
        ('rsi_period', 14),
        ('macd_fast', 12),
        ('macd_slow', 26),
        ('macd_signal', 9),
    )

    def __init__(self):
        # 计算RSI指标
        self.rsi = bt.indicators.RSI(
            self.data.close, period=self.p.rsi_period)
        # 计算MACD指标
        self.macd = bt.indicators.MACD(
            self.data.close,
            period_me1=self.p.macd_fast,
            period_me2=self.p.macd_slow,
            period_signal=self.p.macd_signal)
        self.order = None

    def next(self):
        if self.order:
            return
        if not self.position:
            # 入场：MACD金叉 + RSI在50-70之间(强势但未超买)
            if (self.macd.macd[0] > self.macd.signal[0] and
                self.rsi[0] < 70 and self.rsi[0] > 50):
                self.order = self.buy()
        else:
            # 出场：MACD死叉 或 RSI超买(>70)
            if (self.macd.macd[0] < self.macd.signal[0] or
                self.rsi[0] > 70):
                self.order = self.close()

import backtrader as bt

class MeanReversionStrategy(bt.Strategy):
    """均值回归: 布林带下轨+RSI超卖买入，上轨+RSI超买卖出"""
    params = (
        ('period', 20),          # 布林带周期
        ('devfactor', 2),       # 标准差倍数
        ('position_size', 0.95),
    )

    def __init__(self):
        # 计算布林带指标
        self.bband = bt.indicators.BollingerBands(
            self.data.close,
            period=self.p.period,
            devfactor=self.p.devfactor)
        # RSI辅助确认
        self.rsi = bt.indicators.RSI(
            self.data.close, period=14)
        self.order = None

    def next(self):
        if self.order:
            return
        if not self.position:
            # 入场：价格跌破下轨 且 RSI超卖(<30)
            if (self.data.close[0] < self.bband.lines.bot[0] and
                self.rsi[0] < 30):
                self.order = self.buy(size=self.p.position_size)
        else:
            # 出场：价格突破上轨 或 RSI超买(>70)
            if (self.data.close[0] > self.bband.lines.top[0] or
                self.rsi[0] > 70):
                self.order = self.close()

import backtrader as bt

class GridTradingStrategy(bt.Strategy):
    """网格交易：基准价上下按固定间距设置网格买卖"""
    params = (
        ('grid_interval', 0.02),  # 网格间距(2%)
        ('grid_volume', 100),     # 每格交易量(股)
        ('max_grids', 10),        # 最大网格层数
    )

    def __init__(self):
        self.base_price = None      # 基准价格
        self.last_grid_level = 0    # 上次网格层级
        self.order = None

    def next(self):
        if self.order:
            return
        current_price = self.data.close[0]
        # 首次运行设置基准价格
        if self.base_price is None:
            self.base_price = current_price
            self.order = self.buy(size=self.p.grid_volume)
            return
        # 计算当前网格层级
        grid_level = int(
            (current_price - self.base_price) /
            (self.base_price * self.p.grid_interval))

        if not self.position:
            # 价格下跌到新网格层则买入
            if grid_level < self.last_grid_level:
                self.order = self.buy(size=self.p.grid_volume)
                self.last_grid_level = grid_level
        else:
            if grid_level > self.last_grid_level:
                # 价格上涨到新网格层则卖出
                self.order = self.sell(size=self.p.grid_volume)
                self.last_grid_level = grid_level
            elif grid_level < self.last_grid_level:
                # 价格继续下跌，加仓(限制最大层数)
                if abs(grid_level) < self.p.max_grids:
                    self.order = self.buy(size=self.p.grid_volume)
                    self.last_grid_level = grid_level

import backtrader as bt

class PairsTradingStrategy(bt.Strategy):
    """配对交易: Z-score超阈值开仓，回归0附近平仓"""
    params = dict(
        period=20,         # Z-score计算窗口
        threshold=2.0,     # 开仓阈值
        exit_threshold=0.5, # 平仓阈值
        stake=10,          # 每笔交易股数
    )

    def __init__(self):
        # data0=资产A, data1=资产B
        self.spread = self.data0.close - self.data1.close
        self.spread_sma = bt.indicators.SMA(
            self.spread, period=self.p.period)
        self.spread_std = bt.indicators.StandardDeviation(
            self.spread, period=self.p.period)
        # 计算Z-score
        self.zscore = (
            self.spread - self.spread_sma
        ) / self.spread_std

    def next(self):
        if self.zscore[0] > self.p.threshold:
            # Z>2: 价差过大，A被高估 → 卖A买B
            self.sell(data=self.data0, size=self.p.stake)
            self.buy(data=self.data1, size=self.p.stake)
        elif self.zscore[0] < -self.p.threshold:
            # Z<-2: 价差过小，A被低估 → 买A卖B
            self.buy(data=self.data0, size=self.p.stake)
            self.sell(data=self.data1, size=self.p.stake)
        elif abs(self.zscore[0]) < self.p.exit_threshold:
            # Z回归至0附近，平仓获利
            self.close(data=self.data0)
            self.close(data=self.data1)

import backtrader as bt

class TurtleStrategy(bt.Strategy):
    """海龟交易: 突破入场 + ATR仓位管理 + 金字塔加仓"""
    params = (
        ('long_period', 20),    # 入场通道周期
        ('short_period', 10),   # 离场通道周期
        ('atr_period', 14),     # ATR周期
        ('risk_percent', 0.01), # 每笔风险1%
    )

    def __init__(self):
        self.order = None
        self.buyprice = 0
        self.buy_size = 0
        self.buy_count = 0  # 加仓次数
        # 唐奇安通道
        self.H_line = bt.indicators.Highest(
            self.data.high(-1), period=self.p.long_period)
        self.L_line = bt.indicators.Lowest(
            self.data.low(-1), period=self.p.short_period)
        # ATR计算
        self.TR = bt.indicators.Max(
            (self.data.high(0) - self.data.low(0)),
            abs(self.data.close(-1) - self.data.high(0)),
            abs(self.data.close(-1) - self.data.low(0)))
        self.ATR = bt.indicators.SimpleMovingAverage(
            self.TR, period=self.p.atr_period)
        # 突破信号
        self.buy_signal = bt.ind.CrossOver(
            self.data.close(0), self.H_line)

    def next(self):
        if self.order:
            return
        # 入场：突破上轨且空仓
        if self.buy_signal > 0 and self.buy_count == 0:
            # 仓位 = 总资金×1% / ATR
            self.buy_size = self.broker.getvalue() * 0.01 / self.ATR
            self.buy_size = int(self.buy_size / 100) * 100
            self.order = self.buy(size=self.buy_size)
            self.buyprice = self.data.close[0]
            self.buy_count = 1
        # 加仓：每上涨0.5*ATR加仓，最多4次
        elif (self.position.size > 0 and
              self.data.close[0] > self.buyprice + 0.5 * self.ATR[0] and
              self.buy_count < 4):
            self.order = self.buy(size=self.buy_size)
            self.buyprice = self.data.close[0]
            self.buy_count += 1
        # 止损：回落2*ATR
        elif (self.position.size > 0 and
              self.data.close[0] < self.buyprice - 2 * self.ATR[0]):
            self.order = self.close()
            self.buy_count = 0