双均线交易策略实现与优化

# 导入必要库
import backtrader as bt
import akshare as ak
import pandas as pd
import matplotlib.pyplot as plt

# 创建双均线策略
class DualMovingAverageStrategy(bt.Strategy):
    params = (
        ('short_period', 5),    # 短期均线周期
        ('long_period', 20),    # 长期均线周期
    )

    def __init__(self):
        # 初始化指标计算
        self.short_ma = bt.indicators.SimpleMovingAverage(
            self.data.close, period=self.p.short_period)
        self.long_ma = bt.indicators.SimpleMovingAverage(
            self.data.close, period=self.p.long_period)
        self.crossover = bt.indicators.CrossOver(self.short_ma, self.long_ma)
        
    def next(self):
        if not self.position:  # 没有仓位
            if self.crossover > 0:  # 短线上穿长线，买入
                self.buy()
        elif self.crossover < 0:  # 短线下穿长线，卖出
            self.close()

# 获取股票数据
def get_stock_data(stock_code, start_date, end_date):
    df = ak.stock_zh_a_hist(symbol=stock_code, period="daily", 
                           start_date=start_date, end_date=end_date,
                           adjust="qfq")  # 前复权
    df['date'] = pd.to_datetime(df['日期'])
    df.set_index('date', inplace=True)
    df = df[['开盘', '最高', '最低', '收盘', '成交量']]
    df.columns = ['open', 'high', 'low', 'close', 'volume']
    return df

# 回测函数
def run_backtest(short_period=5, long_period=20, stock_code='000001', 
                start_date='2020-01-01', end_date='2022-12-31'):
    # 初始化cerebro
    cerebro = bt.Cerebro()
    cerebro.addstrategy(DualMovingAverageStrategy, 
                       short_period=short_period, 
                       long_period=long_period)
    
    # 加载数据
    data = get_stock_data(stock_code, start_date, end_date)
    feed = bt.feeds.PandasData(dataname=data)
    cerebro.adddata(feed)
    
    # 设置初始资金
    cerebro.broker.setcash(10000.0)
    cerebro.broker.setcommission(commission=0.001)  # 佣金0.1%
    
    # 添加分析器
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe', timeframe=bt.TimeFrame.Days)
    
    # 运行回测
    print('初始资金: %.2f' % cerebro.broker.getvalue())
    results = cerebro.run()
    strat = results[0]
    
    # 打印结果
    print('最终资金: %.2f' % cerebro.broker.getvalue())
    print('收益率: %.2f%%' % (strat.analyzers.returns.get_analysis()['rtot'] * 100))
    print('最大回撤: %.2f%%' % strat.analyzers.drawdown.get_analysis()['max']['drawdown'])
    print('夏普比率: %.2f' % strat.analyzers.sharpe.get_analysis()['sharperatio'])
    
    # 绘图
    cerebro.plot(style='candlestick')

# 参数优化
def optimize_parameters():
    cerebro = bt.Cerebro(optreturn=False)
    
    # 获取数据
    data = get_stock_data('000001', '2020-01-01', '2022-12-31')
    feed = bt.feeds.PandasData(dataname=data)
    cerebro.adddata(feed)
    
    # 优化范围
    strats = cerebro.optstrategy(
        DualMovingAverageStrategy,
        short_period=range(3, 11),    # 短期均线3-10
        long_period=range(15, 51, 5)  # 长期均线15-50，步长5
    )
    
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
    
    # 运行优化
    opt_results = cerebro.run()
    
    # 找出最佳参数组合
    best_sharpe = -1
    best_params = {}
    
    for run in opt_results:
        sharpe = run.analyzers.sharpe.get_analysis()['sharperatio']
        if sharpe > best_sharpe:
            best_sharpe = sharpe
            best_params = {
                'short_period': run.p.short_period,
                'long_period': run.p.long_period,
                'sharpe': sharpe,
                'return': run.analyzers.returns.get_analysis()['rtot']
            }
    
    print("\n最佳参数组合:")
    print(f"短期均线: {best_params['short_period']}天")
    print(f"长期均线: {best_params['long_period']}天")
    print(f"夏普比率: {best_params['sharpe']:.2f}")
    print(f"收益率: {best_params['return']*100:.2f}%")

# 执行回测
if __name__ == '__main__':
    run_backtest(short_period=5, long_period=20)
    # 如果需要参数优化，取消下面一行的注释
    # optimize_parameters()