最近一个月，我一直在研究如何利用通达信的股票数据进行量化回测。
目前有一点进展，基本完成了“数据获取>>数据处理>>量化回测”的Python代码框架。
在此作为案例分享出来，希望能减少大家量化回测中的一些弯路。
一、量化策略
场内ETF基金，是否存在动量效应或者反转效应？
如果存在，我们就可以通过交易场内ETF基金达到间接投资股票的目的。
目前，场ETF基金（股票+债券+现金+商品等）高达1000多只。
为减少回测工作量和难度，我选择对板块行业指数进行回测。
二、板块行业指数的获取
按照通达信普通行业分类方法，通达信有56个行业。
有关通达信的行业分类，详见昨天的文章：
1、将行业板块指数加入自选股
在【板块指数】标签，点击选择【行业板块】栏目，全选所有行业板块指数，加入自选股板块。具体操作步骤，详见下图。
图1 行业板块指数界面
2、将数据导出为csv文件
键盘输入快键键【34】，调出【数据导出】功能。
图2 键盘精灵--数据导出
点选右上的【高级导出】命令，如图3所示。
图3 高级导出
重要：按照图4对高级导出进行设置！
图4 高级导出设置（重要！）
注意：由于板块指数不存在复权的问题，所以我图4中设置复权方式为不【复权】。如果是导出股票数据，必须选择前复权或后复权，否则行情数据将不准确。
然后【添加品种】，将自选股板块中的所有行业板块指数都添加进来。
图5 添加品种
点击【开始导出】，很快就导出成功了。
图6 导出成功界面
打开导出目录，可看到文件目录下存在56个如下的csv文件：
图7 导出的csv文件
接下来，我们要将这56个csv文件合并到一个文件中。
3、合并数据
在图7中，打开任一个csv文件。我们发现，数据内容格式如下：
图8
对图8中的数据，我简单做下说明。
a.在图4中，我设置了不让通达信【生成导出头部】，因此图8中的数据没有标题行。
b.A~G的数据内容分别为：日期、开盘价、最高价、最低价、收盘价、成交量、成交额。后面，我将通过Python代码给合并的数据加上这些标题行。
c.末行“数据来源:通达信”，属于无效数据，在文件合并时需要忽略掉。
运行以下代码，56个csv文件将合并为一个csv文件：merged_output.csv。
import os
import pandas as pd
from glob import glob
import re
# 请设置成你的真实的文件路径
csv_dir = r"D:\通达信\T0002\export"
output_csv = "merged_output.csv"
# 定义列名（必须与数据列数匹配）
column_names = ["日期", "开盘价", "最高价", "最低价", "收盘价", "成交量", "成交额", "代码"]
# 获取目录下所有CSV文件路径
csv_files = glob(os.path.join(csv_dir, "*.csv"))
# 存储所有DataFrame的列表
dfs = []
for file in csv_files:
# 从文件名提取6位数字代码（如SH#880497.csv → 880497）
match = re.search(r'(\d{6})\.csv$', os.path.basename(file))
if not match:
print(f"警告：跳过不符合命名规则的文件 {file}")
continue
stock_code = match.group(1)
# 读取CSV（无表头，跳过最后一行）
try:
df = pd.read_csv(
file,
encoding='gbk',
header=None,
skipfooter=1,
engine='python'
)
# 检查列数是否匹配（原始数据7列 + 代码列 = 8列）
if len(df.columns) != len(column_names) - 1:
print(f"警告：文件 {file} 有 {len(df.columns)} 列，与预期 {len(column_names)-1} 列不匹配，已跳过")
continue
# 添加代码列
df['代码'] = stock_code
dfs.append(df)
except Exception as e:
print(f"处理文件 {file} 时出错: {str(e)}")
continue
# 合并所有DataFrame
if dfs:
merged_df = pd.concat(dfs, ignore_index=True)
# 添加列名
merged_df.columns = column_names
# 保存结果（包含表头）
merged_df.to_csv(output_csv, index=False, encoding='gbk')
print(f"合并完成！共处理 {len(dfs)} 个文件，结果保存到 {output_csv}")
print("\n合并文件结构示例：")
print(merged_df.head())
else:
print("没有找到符合条件的CSV文件")
三、反转策略回测
反转策略交易规则如下：
从2015年10月8日起，每次买入5日涨幅最小的5个行业板块，5日后卖出，然后重新买入新的5日涨跌幅最小的5个行业板块。
首先，需要获得每个行业指数每天的5日涨跌幅数据。
运行以下Python代码，将数据保存为一个csv文件：周涨幅20152025.csv。
import pandas as pd
# 读取CSV文件（指定GBK编码）
df = pd.read_csv('merged_output.csv',
dtype={'代码': str},
encoding='gbk')
# 转换日期列并排序
df['日期'] = pd.to_datetime(df['日期'])
df = df.sort_values(by=['代码', '日期']) # 按代码和日期排序
# 筛选2015年10月8日及之后的数据
start_date = pd.to_datetime('2015-10-08')
df = df[df['日期'] >= start_date].copy()
# 计算每个代码的5日涨幅（使用groupby确保按股票单独计算）
def calculate_5day_change(group):
group['5日涨幅(%)'] = (group['收盘价'] - group['收盘价'].shift(5)) / group['收盘价'].shift(5) * 100
return group
df = df.groupby('代码', group_keys=False).apply(calculate_5day_change)
df['5日涨幅(%)'] = df['5日涨幅(%)'].round(2)
# 保存结果
df.to_csv('周涨幅20152025.csv', index=False, encoding='gbk')
print("计算完成！结果说明：")
print(f"- 数据时间范围: {df['日期'].min().date()} 至 {df['日期'].max().date()}")
print(f"- 包含股票数量: {df['代码'].nunique()}只")
print("\n示例数据（展示每个股票前5天由于缺乏历史数据产生的空值）：")
sample_codes = df['代码'].unique()[:3] # 展示前3个股票
print(df[df['代码'].isin(sample_codes)].head(15))
然后，运行以下代码，对反转策略进行量化回测。
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib
# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
# 读取CSV文件
try:
df = pd.read_csv('周涨幅20152025.csv',
dtype={'代码': str},
encoding='gbk',
parse_dates=['日期'])
except UnicodeDecodeError:
try:
df = pd.read_csv('周涨幅20152025.csv',
dtype={'代码': str},
encoding='utf-8-sig',
parse_dates=['日期'])
except Exception as e:
raise Exception(f"文件读取失败: {str(e)}")
# 数据清洗
start_date = pd.to_datetime('2015-10-08')
df = df[(df['日期'] >= start_date) & (df['5日涨幅(%)'].notna())].copy()
# 获取所有唯一交易日并按顺序排序
trade_dates = df['日期'].unique()
trade_dates.sort()
trade_dates = pd.to_datetime(trade_dates)
# 生成交易信号（每5个交易日调仓）
signals = []
rebalance_records = []
for i in range(0, len(trade_dates), 5): # 每5个交易日调仓一次
current_date = trade_dates[i]
current_data = df[df['日期'] == current_date]
# 获取当日涨幅最小的5只股票
top5 = current_data.nsmallest(5, '5日涨幅(%)')
# 记录调仓明细
rebalance_records.append({
'调仓日期': current_date,
'调仓股票数': len(top5),
'平均5日涨幅': top5['5日涨幅(%)'].mean(),
'股票代码列表': ','.join(top5['代码'])
})
# 计算卖出日期（5个交易日后的日期）
sell_date_idx = min(i + 5, len(trade_dates) - 1) # 修正了括号问题
sell_date = trade_dates[sell_date_idx]
# 生成交易信号
for _, row in top5.iterrows():
signals.append({
'买入日期': current_date,
'卖出日期': sell_date, # 精确按5个交易日计算
'代码': row['代码'],
'买入价格': row['收盘价'],
'买入5日涨幅': row['5日涨幅(%)']
})
# 转换为DataFrame
signals_df = pd.DataFrame(signals)
rebalance_df = pd.DataFrame(rebalance_records)
# 合并卖出价格
df_sell = df[['日期', '代码', '收盘价']].rename(columns={'日期': '卖出日期', '收盘价': '卖出价格'})
result = pd.merge_asof(
signals_df.sort_values('卖出日期'),
df_sell.sort_values('卖出日期'),
by='代码',
on='卖出日期',
direction='backward'
)
# 计算收益率和持有天数
result['收益率'] = (result['卖出价格'] - result['买入价格']) / result['买入价格']
result['持有天数'] = (result['卖出日期'] - result['买入日期']).dt.days
# 计算组合累计收益
portfolio = result.groupby('卖出日期').agg(
平均收益率=('收益率', 'mean'),
交易数量=('代码', 'count')
).reset_index()
portfolio['累计收益'] = (1 + portfolio['平均收益率']).cumprod() - 1
# 绘制收益曲线
plt.figure(figsize=(12, 6))
plt.plot(portfolio['卖出日期'], portfolio['累计收益']*100, label='策略累计收益', linewidth=2)
plt.axhline(0, color='gray', linestyle='--')
plt.title('5日调仓策略收益曲线（2015/10/08起）\n'
f"最终收益: {portfolio['累计收益'].iloc[-1]*100:.1f}% | "
f"最大回撤: {portfolio['累计收益'].min()*100:.1f}%",
fontsize=14)
plt.xlabel('日期', fontsize=12)
plt.ylabel('累计收益率(%)', fontsize=12)
plt.legend(fontsize=12)
plt.grid(True)
plt.savefig('5day_strategy_performance.png', dpi=300, bbox_inches='tight')
plt.show()
# 保存结果with pd.ExcelWriter('周涨幅反转策略回测明细.xlsx') as writer:
result.to_excel(writer, sheet_name='交易明细', index=False)
portfolio.to_excel(writer, sheet_name='组合表现', index=False)
rebalance_df.to_excel(writer, sheet_name='调仓记录', index=False)
df.to_excel(writer, sheet_name='原始数据', index=False)
result.to_csv('5day_strategy_trades.csv', index=False, encoding='gbk')
portfolio.to_csv('5day_strategy_performance.csv', index=False, encoding='gbk')
rebalance_df.to_csv('5day_rebalance_records.csv', index=False, encoding='gbk')
print("回测完成！结果已保存")
print(f"总交易次数: {len(result)}")
print(f"平均持有天数: {result['持有天数'].mean():.1f}天")
print(f"平均单次收益率: {result['收益率'].mean()*100:.2f}%")
print(f"盈利交易占比: {(result['收益率'] > 0).mean()*100:.1f}%")
print(f"共发生 {len(rebalance_df)} 次调仓操作")
结果是悲伤的。
最终收益率为亏损10.5%，最大回撤为49.5%。
图9 反转策略回测
没关系，再回测下动量策略，说不定结果是相反的。
四、动量策略回测
动量策略与反转策略相反，每次买入5日涨幅最大的5只行业板块指数。
运行以下Python代码：
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib
# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
# 读取CSV文件
try:
df = pd.read_csv('周涨幅20152025.csv',
dtype={'代码': str},
encoding='gbk',
parse_dates=['日期'])
except UnicodeDecodeError:
try:
df = pd.read_csv('weekdata.csv',
dtype={'代码': str},
encoding='utf-8-sig',
parse_dates=['日期'])
except Exception as e:
raise Exception(f"文件读取失败: {str(e)}")
# 数据清洗
start_date = pd.to_datetime('2015-10-08')
df = df[(df['日期'] >= start_date) & (df['5日涨幅(%)'].notna())].copy()
# 获取所有唯一交易日并按顺序排序
trade_dates = df['日期'].unique()
trade_dates.sort()
trade_dates = pd.to_datetime(trade_dates)
# 生成交易信号（每5个交易日调仓）
signals = []
rebalance_records = []
for i in range(0, len(trade_dates), 5): # 每5个交易日调仓一次
current_date = trade_dates[i]
current_data = df[df['日期'] == current_date]
# 修改点：获取当日涨幅最大的5只股票（原为nsmallest）
top5 = current_data.nlargest(5, '5日涨幅(%)') # 改为nlargest
# 记录调仓明细
rebalance_records.append({
'调仓日期': current_date,
'调仓股票数': len(top5),
'平均5日涨幅': top5['5日涨幅(%)'].mean(),
'股票代码列表': ','.join(top5['代码'])
})
# 计算卖出日期（5个交易日后的日期）
sell_date_idx = min(i + 5, len(trade_dates) - 1)
sell_date = trade_dates[sell_date_idx]
# 生成交易信号
for _, row in top5.iterrows():
signals.append({
'买入日期': current_date,
'卖出日期': sell_date,
'代码': row['代码'],
'买入价格': row['收盘价'],
'买入5日涨幅': row['5日涨幅(%)']
})
# 转换为DataFrame
signals_df = pd.DataFrame(signals)
rebalance_df = pd.DataFrame(rebalance_records)
# 合并卖出价格
df_sell = df[['日期', '代码', '收盘价']].rename(columns={'日期': '卖出日期', '收盘价': '卖出价格'})
result = pd.merge_asof(
signals_df.sort_values('卖出日期'),
df_sell.sort_values('卖出日期'),
by='代码',
on='卖出日期',
direction='backward')
# 计算收益率和持有天数
result['收益率'] = (result['卖出价格'] - result['买入价格']) / result['买入价格']
result['持有天数'] = (result['卖出日期'] - result['买入日期']).dt.days
# 计算组合累计收益
portfolio = result.groupby('卖出日期').agg(
平均收益率=('收益率', 'mean'),
交易数量=('代码', 'count')
).reset_index()
portfolio['累计收益'] = (1 + portfolio['平均收益率']).cumprod() - 1
# 绘制收益曲线
plt.figure(figsize=(12, 6))
plt.plot(portfolio['卖出日期'], portfolio['累计收益']*100, label='策略累计收益', linewidth=2)
plt.axhline(0, color='gray', linestyle='--')
plt.title('5日调仓策略（买入涨幅最大5只）\n'
f"最终收益: {portfolio['累计收益'].iloc[-1]*100:.1f}% | "
f"最大回撤: {portfolio['累计收益'].min()*100:.1f}%",
fontsize=14)
plt.xlabel('日期', fontsize=12)
plt.ylabel('累计收益率(%)', fontsize=12)
plt.legend(fontsize=12)
plt.grid(True)
plt.savefig('5day_strategy_performance_max.png', dpi=300, bbox_inches='tight') # 修改输出文件名
plt.show()
# 保存结果
with pd.ExcelWriter('5day_strategy_results_max.xlsx') as writer: # 修改文件名
result.to_excel(writer, sheet_name='交易明细', index=False)
portfolio.to_excel(writer, sheet_name='组合表现', index=False)
rebalance_df.to_excel(writer, sheet_name='调仓记录', index=False)
df.to_excel(writer, sheet_name='原始数据', index=False)
result.to_csv('5day_strategy_trades_max.csv', index=False, encoding='gbk') # 修改文件名portfolio.to_csv('5day_strategy_performance_max.csv', index=False, encoding='gbk') # 修改文件名rebalance_df.to_csv('5day_rebalance_records_max.csv', index=False, encoding='gbk') # 修改文件名
print("回测完成！结果已保存")
print(f"总交易次数: {len(result)}")
print(f"平均持有天数: {result['持有天数'].mean():.1f}天")
print(f"平均单次收益率: {result['收益率'].mean()*100:.2f}%")
print(f"盈利交易占比: {(result['收益率'] > 0).mean()*100:.1f}%")
print(f"共发生 {len(rebalance_df)} 次调仓操作")
回测结果的确比反转策略要好些。
图10 动量策略回测
动量策略，最终收益率为4.5%，最大回撤为26.4%。
动量策略，依然不是一个好的盈利策略。
五、为什么两种策略都赚不到钱？
仔细观察图10动量策略的收益率曲线，发现曲线走势与大盘走势基本是相同的。
上述的买卖规则，我们只买了5个行业指数，但实际上我们仍相当于在买卖大盘指数，最终的收益大小必然和大盘涨跌幅基本相当。
在前述回测中，如果每次只买1个行业指数，结果会如何呢？
反转策略和动量策略的结果大相径庭！
图11 反转策略（只买一只）
图12 动量策略（只买一只）
