用 Python 实现一个简单的 A 股交易策略回测框架 大家好，我是Bob！之前的文章，对于新手学习python和在自己的代码中接入DeepSeek来说，已经完全够用了。在此基础上，是可以根据自己的思路，做出一些金融小工具的。 最近我忙着做加密币的量化交易，觉得是时候出一篇交易策略回测框架的教程了！学以致用才是传播知识的初衷！
在量化交易的世界里，回测是验证交易策略有效性的重要手段。今天我们将通过一个实际案例，以贵州茅台（600519）为例，构建一个简单但完整的双均线策略回测框架。
文章结尾附上完整源码。
一、回测框架的核心要素 一个完整的回测框架通常包含以下几个关键部分：
二、获取市场数据 首先，我们使用 akshare 库获取贵州茅台的历史数据：
import akshare as ak import pandas as pd # 获取贵州茅台的日线历史行情 stock_df = ak.stock_zh_a_hist(symbol= "600519" , period= "daily" , adjust= "qfq" ) stock_df[ 'date' ] = pd.to_datetime(stock_df[ '日期' ]) stock_df.set_index(
'date' , inplace= True ) stock_df = stock_df[[ '收盘' ]] stock_df.rename(columns={ '收盘' : 'close' }, inplace= True ) # 取最近一年的数据用于分析 stock_df = stock_df.tail( 252 ) 这里我们使用前复权（qfq）数据，确保历史价格的连续性，避免因分红送股造成的价格跳空。
三、双均线策略实现 双均线策略是最经典的技术分析策略之一。当短期均线上穿长期均线时买入（金叉），反之卖出（死叉）。
# 计算均线 short_window = 5 long_window = 20 stock_df[ 'ma_short' ] = stock_df[ 'close' ].rolling(window=short_window).mean() stock_df[ 'ma_long' ] = stock_df[ 'close' ].rolling(window=long_window).mean() # 生成交易信号 stock_df[ 'signal' ] = 0 stock_df.iloc[short_window:, stock_df.columns.get_loc( 'signal' )] = ( stock_df[ 'ma_short' ].iloc[short_window:] > stock_df[ 'ma_long' ].iloc[short_window:] ).astype( int ) # 计算买卖点 stock_df[ 'position' ] = stock_df[ 'signal' ].diff() 这里的关键是：
signal 为1表示持仓，为0表示空仓 position 为1表示买入信号，为-1表示卖出信号
四、收益率计算 回测的核心是计算策略收益并与基准对比：
# 计算收益率 stock_df[ 'returns' ] = stock_df[ 'close' ].pct_change() stock_df[ 'strategy_returns' ] = stock_df[ 'signal' ].shift( 1 ) * stock_df[ 'returns' ] stock_df[ 'cum_returns' ] = ( 1 + stock_df[ 'returns' ]).cumprod() - 1 stock_df[ 'cum_strategy_returns' ] = ( 1 + stock_df[ 'strategy_returns' ]).cumprod() - 1 注意这里使用 shift(1) 是为了避免未来函数，确保我们在当日收盘后才能获得信号。
五、性能评估指标 一个好的回测框架需要提供多维度的性能指标：
# 计算关键指标 total_trades = len (buy_signals) + len (sell_signals) win_trades = sum (stock_df[ 'strategy_returns' ] > 0 ) total_days = sum (~np.isnan(stock_df[
'strategy_returns' ])) win_rate = win_trades / total_days if total_days > 0 else 0 # 最大回撤 strategy_returns = stock_df[ 'cum_strategy_returns' ].dropna() max_drawdown = (strategy_returns.cummax() - strategy_returns). max () # 最终收益 final_return = strategy_returns.iloc[- 1 ] * 100 buy_hold_return = stock_df[ 'cum_returns' ].iloc[- 1 ] * 100 六、策略可视化 可视化是回测框架的重要组成部分，帮助我们直观理解策略表现：
import matplotlib.pyplot as plt import matplotlib.ticker as mticker # 创建图形 fig = plt.figure(figsize=( 16 , 8 )) ax = fig.add_subplot( 111 ) # 绘制价格和均线 ax.plot(stock_df.index, stock_df[ 'close' ], label= 'Maotai Price' ) ax.plot(stock_df.index, stock_df[ 'ma_short' ], label= f' {short_window} -Day MA' ) ax.plot(stock_df.index, stock_df[ 'ma_long' ], label= f'
{long_window} -Day MA' ) # 标记买卖点 buy_signals = stock_df[stock_df[ 'position' ] == 1 ] sell_signals = stock_df[stock_df[ 'position' ] == - 1 ] ax.scatter(buy_signals.index, buy_signals[ 'close' ], marker= '^' , color= 'g' , s= 150 , label= 'Buy Signal' ) ax.scatter(sell_signals.index, sell_signals[ 'close' ], marker= 'v' , color= 'r' , s= 150 , label= 'Sell Signal' ) 七、策略回测结果解读 运行回测后，我们得到了以下结果： 这个结果告诉我们什么？ 1. 策略失效 ：在这个时间段内，双均线策略不仅没有赚钱，反而比简单持有亏损更多 2. 频繁交易 ：短期均线导致频繁交易，增加了交易成本 3. 低胜率 ：大部分交易都是亏损的，表明这不是一个好的策略配置
八、回测框架的扩展思路 基于这个简单框架，我们可以进行多方面扩展：
参数优化 # 遍历不同的均线组合 for short in [ 5 , 10 , 20 ]: for long in [ 20 , 30 , 60 ]: if short < long: # 运行回测并记录结果 pass 风险管理 # 加入止损逻辑 if current_loss > stop_loss_threshold: generate_sell_signal() 多因子策略 # 结合成交量指标 volume_condition = stock_df[ 'volume' ] > stock_df[ 'volume' ].rolling( 20 ).mean() signal = ma_cross_signal & volume_condition 交易成本考虑 # 考虑手续费和滑点 commission_rate = 0.001 # 千分之一 slippage = 0.002 # 千分之二 net_returns = strategy_returns - commission_rate - slippage 九、总结与建议 通过这个简单的回测框架，我们实现了：
完整的回测流程：从数据获取到结果展示 标准化的评估指标：收益率、胜率、回撤等 直观的可视化：清晰展示策略执行情况
但同时也发现了双均线策略的局限性。对于量化交易新手，建议：
从简单策略开始：先理解基础原理，再逐步增加复杂度 重视风险管理：不要只看收益，更要控制回撤 多样本测试：在不同市场环境、不同股票上测试策略 持续优化：根据回测结果不断改进策略参数
回测只是验证策略的第一步，真实市场会更加复杂。保持谨慎，持续学习，才能在量化交易的道路上走得更远。
最近玩了一阵子量化之后，感觉这个东西非常有意思，所以写一篇文章，抛砖引玉额，希望能和量化高手多多交流。后续会建立微信群，集思广益，把思路付诸于代码！
import akshare as ak import pandas as pd import matplotlib.pyplot as plt import matplotlib.ticker as mticker import matplotlib.style as style import numpy as np
# 设置英文字体和样式 plt.rcParams[ 'font.family' ] = 'sans-serif' plt.rcParams[ 'font.sans-serif' ] = [ 'Arial' , 'Helvetica' , 'DejaVu Sans' ] plt.rcParams[ 'axes.unicode_minus' ] = True style.use( 'ggplot' ) # 获取贵州茅台的日线历史行情 stock_df = ak.stock_zh_a_hist(symbol= "600519" , period= "daily" , adjust= "qfq" ) stock_df[ 'date' ] = pd.to_datetime(stock_df[ '日期' ]) stock_df.set_index( 'date' , inplace= True ) stock_df = stock_df[[ '收盘' ]] stock_df.rename(columns={ '收盘' : 'close' }, inplace= True ) # 取最近的数据用于分析 stock_df = stock_df.tail( 252 ) # 约一年的交易日数据 # 计算均线和信号 short_window = 5 long_window = 20 stock_df[ 'ma_short' ] = stock_df[ 'close' ].rolling(window=short_window).mean() stock_df[ 'ma_long' ] = stock_df[ 'close' ].rolling(window=long_window).mean() stock_df[ 'signal' ] = 0 # 使用 .iloc 进行基于位置的索引
'signal' )] = ( stock_df[ 'ma_short' ].iloc[short_window:] > stock_df[ 'ma_long' ].iloc[short_window:] ).astype( int ) stock_df[ 'position' ] = stock_df[ 'signal' ].diff() # 计算收益率 stock_df[ 'returns' ] = stock_df[ 'close' ].pct_change() stock_df[ 'strategy_returns' ] = stock_df[ 'signal' ].shift( 1 ) * stock_df[ 'returns' ] stock_df[ 'cum_returns' ] = ( 1 + stock_df[ 'returns' ]).cumprod() - 1 stock_df[ 'cum_strategy_returns' ] = ( 1 + stock_df[ 'strategy_returns' ]).cumprod() - 1 print (stock_df.tail()) # 创建图形 fig = plt.figure(figsize=( 16 , 8 ), dpi= 100 , facecolor= 'white' ) ax = fig.add_subplot( 111 ) # 颜色方案 price_color = '#2c3e50' # 深蓝色 short_ma_color = '#e74c3c' # 红色 long_ma_color = '#3498db' # 蓝色 buy_color = '#27ae60' # 绿色 sell_color = '#c0392b' # 深红色 background_color = '#f9f9f9' # 浅灰背景 grid_color = '#bdc3c7' # 网格颜色 # 设置背景色 ax.set_facecolor(background_color) # 绘制价格和移动平均线 ax.plot(stock_df.index, stock_df[ 'close' ], linewidth= 2 , color=price_color, label= 'Maotai Price' , alpha= 0.8 ) ax.plot(stock_df.index, stock_df[ 'ma_short' ], linewidth= 2 , color=short_ma_color, label= f' {short_window} -Day MA' , alpha= 0.9 ) ax.plot(stock_df.index, stock_df[ 'ma_long' ], linewidth= 2 , color=long_ma_color, label= f' {long_window} -Day MA' , alpha= 0.9 ) # 买卖信号点 buy_signals = stock_df[stock_df[ 'position' ] == 1 ] sell_signals = stock_df[stock_df[ 'position' ] == - 1 ] # 买入点 ax.scatter(buy_signals.index, buy_signals[ 'close' ], s= 150 , marker= '^' , color=buy_color, label= 'Buy Signal' , zorder= 5 , edgecolors= 'white' , linewidth= 1.5 ) # 卖出点 ax.scatter(sell_signals.index, sell_signals[ 'close' ], s= 150 , marker= 'v' , color=sell_color, label= 'Sell Signal' , zorder= 5 , edgecolors= 'white' , linewidth= 1.5 ) # 添加金叉死叉区域高亮 for i in range ( len (stock_df) - 1 ): if i > 0 : # 确保不会越界 if stock_df[ 'ma_short' ].iloc[i] > stock_df[ 'ma_long' ].iloc[i] and \ stock_df[ 'ma_short' ].iloc[i- 1 ] <= stock_df[ 'ma_long' ].iloc[i- 1 ]: # 均线金叉 ax.axvspan(stock_df.index[i- 1 ], stock_df.index[i+ 1 ], alpha= 0.2 , color=buy_color) elif stock_df[ 'ma_short' ].iloc[i] < stock_df[ 'ma_long' ].iloc[i] and \ stock_df[ 'ma_short' ].iloc[i- 1 ] >= stock_df[ 'ma_long' ].iloc[i- 1 ]: # 均线死叉 ax.axvspan(stock_df.index[i- 1 ], stock_df.index[i+ 1 ], alpha= 0.2 , color=sell_color) # 格式化y轴，使用人民币符号（¥） def rmb_format ( x, pos ): return f'¥ {x: .0 f} ' # 整数显示，适合茅台高价格 ax.yaxis.set_major_formatter(mticker.FuncFormatter(rmb_format)) # 设置网格 ax.grid( True , linestyle= '--' , alpha= 0.6 , color=grid_color) # 设置标题和标签 ax.set_title( "Kweichow Moutai (600519) Moving Average Trading Strategy" , fontsize= 16 , fontweight= 'bold' , pad= 20 ) ax.set_xlabel( 'Date' , fontsize= 12 ) ax.set_ylabel( 'Price (CNY)' , fontsize= 12 ) # 添加图例 leg = ax.legend(loc= 'upper left' , frameon= True , framealpha= 0.9 , fontsize= 10 , ncol= 3 , fancybox= True , shadow= True ) leg.get_frame().set_edgecolor( '#d5d5d5' ) # 添加茅台标识 plt.figtext( 0.02 , 0.02 , "贵州茅台 Kweichow Moutai (600519.SH)" , fontsize= 10 , alpha= 0.7 )
# 添加性能指标文本框 total_trades = len (buy_signals) + len (sell_signals) win_trades = sum (stock_df[ 'strategy_returns' ] > 0 ) total_days = sum (~np.isnan(stock_df[ 'strategy_returns' ])) win_rate = win_trades / total_days if total_days > 0 else 0 strategy_returns = stock_df[ 'cum_strategy_returns' ].dropna() if len (strategy_returns) > 0 : max_drawdown = (strategy_returns.cummax() - strategy_returns). max () final_return = strategy_returns.iloc[- 1 ] * 100 else : max_drawdown = 0 final_return = 0 buy_hold_return = stock_df[ 'cum_returns' ].iloc[- 1 ] * 100 performance_text = ( f"Strategy Performance:\n" f"Total Trades: {total_trades} \n" f"Win Rate: {win_rate: .2 %} \n" f"Max Drawdown: {max_drawdown: .2 %} \n" f"Strategy Return: {final_return: .2 f} %\n" f"Buy & Hold Return:
{buy_hold_return: .2 f} %" ) fig.text( 0.92 , 0.5 , performance_text, fontsize= 10 , va= 'center' , ha= 'right' , bbox= dict (boxstyle= 'round,pad=0.5' , facecolor= 'white' , alpha= 0.8 , edgecolor= '#3498db' )) # 调整布局 plt.tight_layout() plt.subplots_adjust(right= 0.88 ) # 显示图表 plt.show() # 控制台输出性能统计 print ( "\n=== Strategy Performance Analysis ===" ) print ( f"Data Period: {stock_df.index[ 0 ].date()} to {stock_df.index[- 1 ].date()} " ) print ( f"Total Trading Days: { len (stock_df)} " ) print ( f"Total Trades: {total_trades} " ) print ( f"Buy Signals: { len (buy_signals)} " ) print ( f"Sell Signals: { len (sell_signals)} " ) if
0 : print ( f"Win Rate: {win_rate: .2 %} " ) print ( f"Strategy Return: {final_return: .2 f} %" ) print ( f"Buy & Hold Return: {buy_hold_return: .2 f} %" ) print ( f"Max Drawdown: {max_drawdown: .2 %} " ) 
