通达信Python量化回测框架

admin 2025-09-08 阅读:34 评论:0

最近一个月，我一直在研究如何利用通达信的股票数据进行量化回测。目前有一点进展，基本完成了“数据获取>>数据处理>>量化回测”的Python代码框架。在此作为案例分享出来，希望能减少大家量化回测中的一些弯路。...

最近一个月，我一直在研究如何利用通达信的股票数据进行量化回测。

目前有一点进展，基本完成了“数据获取>>数据处理>>量化回测”的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)} 次调仓操作")

回测结果的确比反转策略要好些。