跳到主要内容

均线策略实战

本章将通过一个完整的均线策略实战案例，演示从数据获取、策略开发、回测分析到实盘准备的全流程。

策略设计​

策略思路​

均线交叉是最经典的趋势跟踪策略之一。其核心逻辑是：

金叉买入：短期均线上穿长期均线，表明短期趋势转强，买入 死叉卖出：短期均线下穿长期均线，表明短期趋势转弱，卖出

这个策略的优点是逻辑简单、易于实现，缺点是在震荡市场中会产生大量假信号。

策略优化​

原始的均线交叉策略存在一些问题，我们可以进行优化：

添加过滤条件：只在趋势明确时交易，避免震荡市场的假信号 动态参数：根据市场波动调整均线周期 风险控制：添加止损止盈，控制单笔风险

完整代码实现​

import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime
import matplotlib.pyplot as plt

class EnhancedMAStrategy(bt.Strategy):
    """增强版均线策略"""
    
    params = (
        ('short_period', 10),      # 短期均线周期
        ('long_period', 30),       # 长期均线周期
        ('stop_loss', 0.05),       # 止损比例
        ('take_profit', 0.15),     # 止盈比例
        ('risk_per_trade', 0.02),  # 单笔风险比例
        ('adx_threshold', 25),     # ADX趋势强度阈值
        ('printlog', True),
    )
    
    def __init__(self):
        # 初始化指标
        self.ma_short = bt.indicators.SMA(self.data.close, period=self.params.short_period)
        self.ma_long = bt.indicators.SMA(self.data.close, period=self.params.long_period)
        self.adx = bt.indicators.ADX(self.data, period=14)
        self.atr = bt.indicators.ATR(self.data, period=14)
        
        # 交叉信号
        self.crossover = bt.indicators.CrossOver(self.ma_short, self.ma_long)
        
        # 订单状态
        self.order = None
        self.entry_price = None
        self.entry_date = None
    
    def log(self, txt, dt=None):
        """日志输出"""
        if self.params.printlog:
            dt = dt or self.datas[0].datetime.date(0)
            print(f'[{dt.isoformat()}] {txt}')
    
    def notify_order(self, order):
        """订单状态通知"""
        if order.status in [order.Submitted, order.Accepted]:
            return
        
        if order.status in [order.Completed]:
            if order.isbuy():
                self.entry_price = order.executed.price
                self.entry_date = self.datas[0].datetime.date(0)
                self.log(f'买入执行: 价格={order.executed.price:.2f}, '
                        f'成本={order.executed.value:.2f}, '
                        f'手续费={order.executed.comm:.2f}')
            else:
                self.log(f'卖出执行: 价格={order.executed.price:.2f}, '
                        f'成本={order.executed.value:.2f}, '
                        f'手续费={order.executed.comm:.2f}')
        
        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log('订单失败')
        
        self.order = None
    
    def notify_trade(self, trade):
        """交易完成通知"""
        if not trade.isclosed:
            return
        
        holding_days = (self.datas[0].datetime.date(0) - self.entry_date).days if self.entry_date else 0
        self.log(f'交易完成: 毛利={trade.pnl:.2f}, 净利={trade.pnlcomm:.2f}, '
                f'持仓天数={holding_days}')
    
    def next(self):
        """策略主逻辑"""
        # 如果有未完成订单，不操作
        if self.order:
            return
        
        # 当前价格和持仓
        current_price = self.data.close[0]
        
        # 如果有持仓，检查止损止盈
        if self.position:
            # 计算盈亏比例
            pnl_pct = (current_price - self.entry_price) / self.entry_price
            
            # 止损
            if pnl_pct <= -self.params.stop_loss:
                self.log(f'触发止损: 盈亏={pnl_pct:.2%}')
                self.order = self.sell()
                return
            
            # 止盈
            if pnl_pct >= self.params.take_profit:
                self.log(f'触发止盈: 盈亏={pnl_pct:.2%}')
                self.order = self.sell()
                return
            
            # 死叉卖出
            if self.crossover < 0:
                self.log(f'死叉卖出信号')
                self.order = self.sell()
        
        else:
            # 没有持仓，检查买入条件
            # 条件1：金叉
            if self.crossover > 0:
                # 条件2：ADX显示有趋势
                if self.adx[0] > self.params.adx_threshold:
                    self.log(f'买入信号: 短期均线={self.ma_short[0]:.2f}, '
                            f'长期均线={self.ma_long[0]:.2f}, ADX={self.adx[0]:.2f}')
                    
                    # 计算仓位
                    self.order = self.buy()
    
    def stop(self):
        """策略结束时调用"""
        self.log(f'策略结束，最终资产: {self.broker.getvalue():.2f}', 
                dt=self.datas[0].datetime.date(0))


def run_backtest(ticker='AAPL', start_date='2020-01-01', end_date='2023-12-31',
                 initial_cash=100000, commission=0.001):
    """
    运行回测
    
    参数:
        ticker: 股票代码
        start_date: 开始日期
        end_date: 结束日期
        initial_cash: 初始资金
        commission: 手续费率
    
    返回:
        回测结果
    """
    # 创建回测引擎
    cerebro = bt.Cerebro()
    
    # 添加策略
    cerebro.addstrategy(EnhancedMAStrategy)
    
    # 获取数据
    print(f'正在获取 {ticker} 数据...')
    data = yf.download(ticker, start=start_date, end=end_date, progress=False)
    
    if len(data) == 0:
        print('数据获取失败')
        return None
    
    # 转换数据格式
    data_feed = bt.feeds.PandasData(
        dataname=data,
        datetime=None,
        open='Open',
        high='High',
        low='Low',
        close='Close',
        volume='Volume',
        openinterest=-1
    )
    
    cerebro.adddata(data_feed)
    
    # 设置初始资金
    cerebro.broker.setcash(initial_cash)
    
    # 设置手续费
    cerebro.broker.setcommission(commission=commission)
    
    # 添加分析器
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe', riskfreerate=0.02)
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
    cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
    cerebro.addanalyzer(bt.analyzers.SQN, _name='sqn')
    
    # 运行回测
    print(f'初始资金: {initial_cash:,.2f}')
    results = cerebro.run()
    strat = results[0]
    
    # 输出结果
    print('\n' + '='*50)
    print('回测结果')
    print('='*50)
    print(f'最终资金: {cerebro.broker.getvalue():,.2f}')
    
    # 收益分析
    returns = strat.analyzers.returns.get_analysis()
    print(f'\n收益分析:')
    print(f'  总收益率: {returns.get("rtot", 0)*100:.2f}%')
    print(f'  年化收益率: {returns.get("rnorm100", 0):.2f}%')
    
    # 风险分析
    drawdown = strat.analyzers.drawdown.get_analysis()
    print(f'\n风险分析:')
    print(f'  最大回撤: {drawdown.get("max", {}).get("drawdown", 0):.2f}%')
    print(f'  最大回撤持续: {drawdown.get("max", {}).get("len", 0)} 天')
    
    # 夏普比率
    sharpe = strat.analyzers.sharpe.get_analysis()
    print(f'\n风险调整收益:')
    print(f'  夏普比率: {sharpe.get("sharperatio", "N/A")}')
    
    # 交易分析
    trades = strat.analyzers.trades.get_analysis()
    print(f'\n交易分析:')
    print(f'  总交易次数: {trades.get("total", {}).get("total", 0)}')
    print(f'  盈利交易: {trades.get("won", {}).get("total", 0)}')
    print(f'  亏损交易: {trades.get("lost", {}).get("total", 0)}')
    
    if trades.get("total", {}).get("total", 0) > 0:
        win_rate = trades.get("won", {}).get("total", 0) / trades["total"]["total"] * 100
        print(f'  胜率: {win_rate:.2f}%')
        
        avg_win = trades.get("won", {}).get("pnl", {}).get("average", 0)
        avg_loss = trades.get("lost", {}).get("pnl", {}).get("average", 0)
        print(f'  平均盈利: {avg_win:.2f}')
        print(f'  平均亏损: {avg_loss:.2f}')
        
        if avg_loss != 0:
            print(f'  盈亏比: {abs(avg_win/avg_loss):.2f}')
    
    # SQN（系统质量数）
    sqn = strat.analyzers.sqn.get_analysis()
    print(f'\n系统质量数: {sqn.get("sqn", "N/A")}')
    
    return results, cerebro


def optimize_parameters(ticker='AAPL', start_date='2020-01-01', end_date='2023-12-31'):
    """
    参数优化
    """
    cerebro = bt.Cerebro()
    
    # 参数优化范围
    cerebro.optstrategy(
        EnhancedMAStrategy,
        short_period=range(5, 15),
        long_period=range(20, 50, 5),
        adx_threshold=[20, 25, 30]
    )
    
    # 获取数据
    data = yf.download(ticker, start=start_date, end=end_date, progress=False)
    data_feed = bt.feeds.PandasData(dataname=data)
    cerebro.adddata(data_feed)
    
    # 设置
    cerebro.broker.setcash(100000)
    cerebro.broker.setcommission(commission=0.001)
    
    # 添加分析器
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
    
    # 运行优化
    print('正在进行参数优化...')
    results = cerebro.run(maxcpu=1)
    
    # 整理结果
    optimization_results = []
    for result in results:
        strat = result[0]
        sharpe = strat.analyzers.sharpe.get_analysis().get('sharperatio', 0)
        params = strat.params
        optimization_results.append({
            'short_period': params.short_period,
            'long_period': params.long_period,
            'adx_threshold': params.adx_threshold,
            'sharpe': sharpe if sharpe else 0
        })
    
    # 排序
    df_results = pd.DataFrame(optimization_results)
    df_results = df_results.sort_values('sharpe', ascending=False)
    
    print('\n最优参数组合（按夏普比率排序）:')
    print(df_results.head(10).to_string(index=False))
    
    return df_results


def plot_results(cerebro):
    """绘制回测结果"""
    cerebro.plot(style='candlestick', barup='red', bardown='green')


if __name__ == '__main__':
    # 运行回测
    results, cerebro = run_backtest('AAPL', '2020-01-01', '2023-12-31')
    
    # 参数优化（可选）
    # optimize_parameters('AAPL', '2020-01-01', '2023-12-31')
    
    # 绘制结果
    # plot_results(cerebro)

策略分析​

回测结果解读​

运行上述代码后，我们会得到类似以下的回测结果：

回测结果
==================================================
最终资金: 125,678.90

收益分析:
  总收益率: 25.68%
  年化收益率: 7.89%

风险分析:
  最大回撤: 12.35%
  最大回撤持续: 45 天

风险调整收益:
  夏普比率: 0.85

交易分析:
  总交易次数: 48
  盈利交易: 28
  亏损交易: 20
  胜率: 58.33%
  平均盈利: 1,234.56
  平均亏损: -876.32
  盈亏比: 1.41

策略评估​

收益表现：年化收益率7.89%，相比买入持有策略如何？需要与基准对比。

风险水平：最大回撤12.35%处于可接受范围，但需要评估回撤持续时间。

风险调整收益：夏普比率0.85，处于良好水平，但还有提升空间。

交易质量：胜率58.33%，盈亏比1.41，说明策略能够捕捉到较好的趋势行情。

改进方向​

  1. 添加市场环境过滤：在震荡市场中减少交易
  2. 动态调整参数：根据市场波动调整均线周期
  3. 改进止损机制：使用移动止损锁定利润
  4. 多时间框架确认：使用更长周期的趋势作为过滤条件

样本外验证​

def out_of_sample_test(ticker='AAPL'):
    """样本外测试"""
    
    # 训练期：2020-2022
    print('训练期回测 (2020-2022):')
    train_results, _ = run_backtest(ticker, '2020-01-01', '2022-12-31')
    
    # 测试期：2023
    print('\n测试期回测 (2023):')
    test_results, _ = run_backtest(ticker, '2023-01-01', '2023-12-31')
    
    # 对比分析
    print('\n样本内外对比:')
    # 这里可以添加更详细的对比分析

实盘准备​

实盘注意事项​

滑点设置：回测中的滑点设置需要反映真实交易情况。

# 设置滑点
cerebro.broker.set_slippage_perc(perc=0.0001)  # 万分之一滑点

交易时间：确保只在交易时段下单。

资金管理：实盘中需要更严格的资金管理。

异常处理：处理网络中断、数据异常等情况。

实盘代码框架​

class LiveTrading:
    """实盘交易框架"""
    
    def __init__(self, broker_api, strategy_params):
        self.broker = broker_api
        self.params = strategy_params
        self.position = 0
    
    def on_bar(self, bar):
        """每根K线调用"""
        # 计算指标
        # 生成信号
        # 执行交易
        pass
    
    def on_tick(self, tick):
        """每个Tick调用"""
        pass
    
    def on_order(self, order):
        """订单状态更新"""
        pass
    
    def run(self):
        """运行实盘"""
        # 连接券商
        # 订阅行情
        # 开始交易循环
        pass

小结​

本章通过一个完整的均线策略实战案例，演示了量化策略开发的全流程：

  1. 策略设计：明确策略逻辑和优化方向
  2. 代码实现：使用Backtrader框架实现策略
  3. 回测分析：评估策略的收益和风险特征
  4. 参数优化：寻找最优参数组合
  5. 样本外验证：验证策略的泛化能力
  6. 实盘准备：考虑实盘交易的特殊情况

量化策略开发是一个迭代优化的过程，需要不断测试、分析、改进。记住，回测表现好不代表实盘一定盈利，始终保持谨慎的态度。