第 3 章 · 04 KDJ 回测代码实战

第 3 章 · 04 KDJ 回测代码实战

本节摘要:这是本章的实战重点。前面三节讲了技术指标的分类、全景与择时方法,本节把这些方法论落到代码——逐行精读 仓库里的 main.py 与 technical.py 两份代码,跑一次完整的「KDJ 金叉死叉信号胜率检验」。technical.py 用 40 日 rolling min/max 算 RSV,再用 ewm(span=2) 双重平滑得到 K、D,通过 shift(1) 对比昨日与今日的 K/D 大小生成金叉死叉信号;main.py 遍历股票 CSV,用 shift(-N) 算未来 1/5/10/20 日涨跌幅,按 signal 列 groupby 分组,用 describe() 看分布并算胜率。读完本节,你彻底掌握 pandas 时序回测的几个核心技巧——shift、groupby、describe、rolling、ewm,并能独立写一个简单但完整的指标回测框架。

内容来源:资料 技术指标回测代码 / main.py + technical.py 逐行精读,知识结构化整理。

⚠️ 学习提示:本节是全教程少有的「逐行精读代码」环节。即使你 pandas 已熟练,也建议跟着读懂每一步——这里展示的是「信号回测的最小骨架」,后续可扩展到任何技术指标。

学习目标

阅读完本节,你应当能够:

1. 逐行说清 technical.py 中 RSV、K、D、signal 的计算逻辑。
2. 逐行说清 main.py 中遍历股票、算未来收益、分组统计的流程。
3. 掌握 pandas 的 shift、groupby、describe、rolling、ewm 五个核心方法。
4. 理解未来函数(shift(-N))与信号延迟(shift(1))的区别与正确用法。
5. 独立修改这两份代码,测试其他技术指标的胜率。

一、整体架构:两份代码各做什么

仓库的技术指标回测代码只有两份,极其简洁:

分工:

• technical.py:定义函数 technical_indicator(table),输入单只股票的 OHLC 表,输出加了 RSV/K/D/signal 列的表。只算指标与信号,不算收益。
• main.py:遍历股票 CSV,对每只股票调用 technical_indicator,再统一算未来 N 日涨跌幅,合并后按 signal 分组统计胜率。只算收益与统计,不算指标。

这种「指标计算」与「回测统计」分离的设计,让换指标只需改 technical.py、换统计口径只需改 main.py,互不干扰。是工程上很好的实践。

二、technical.py 逐行精读

整份 technical.py 只有一个函数,逐行看:

def technical_indicator(table):
​

输入 table 是单只股票的 DataFrame,必须包含 最低价_复权、最高价_复权、收盘价_复权 这几列(中文列名,后复权价格)。后复权是为了让历史价格连续,避免除权造成的跳空。

第一步:算 40 日内的最低价与最高价

table['LOW_N'] = table['最低价_复权'].rolling(40).min()
table['HIGH_N'] = table['最高价_复权'].rolling(40).max()
​

• rolling(40).min():过去 40 天(含当日)的最低价的最小值——即 40 日内最低点。
• rolling(40).max():40 日内最高点。

💡 核心心法:rolling(40) 是 pandas 滚动窗口的核心方法,返回一个 Rolling 对象,再 .min()/.max()/.mean() 等聚合。前 39 天因为窗口不足 40 天,值为 NaN——这是 rolling 的固有行为,后续会自动被丢弃或填充。

注意原代码用的是 span=(3-1) 即 2,但 40 日的窗口是写死的——你可以改成 9 日(传统 KDJ 默认)或 14 日,看哪个版本在你的股票上更有效。

第二步:算 RSV(未成熟随机值)

table['RSV'] = (table['收盘价_复权'] - table['LOW_N']) / (table['HIGH_N'] - table['LOW_N']) * 100
​

这正是第 01 节讲的 KDJ 第一步:

RSV = (收盘价 - N日最低) / (N日最高 - N日最低) × 100
​

含义:当前价格在过去 40 天的高低区间的相对位置。

• RSV = 100:收盘价就是 40 日最高,处于区间顶端。
• RSV = 0:收盘价就是 40 日最低,处于区间底端。
• RSV = 50:收盘价位于区间中点。

RSV 本身就是「相对强弱」的度量,但它太敏感、噪声大,需要平滑得到 K、D。

第三步:双重 EMA 平滑得到 K 与 D

table['K'] = table['RSV'].ewm(span=(3-1), adjust=False).mean()
table['D'] = table['K'].ewm(span=(3-1), adjust=False).mean()
​

• ewm(span=2, adjust=False).mean():指数加权移动平均(EMA),span=2 对应平滑系数 α = 2/(2+1) = 0.667,给近期数据极高权重,响应非常快。
• K = EMA(RSV),是 RSV 的一次平滑。
• D = EMA(K),是 K 的再次平滑(即 RSV 的二次平滑)。

adjust=False 是关键参数——它使用递归公式 y_t = (1-α)·y_(t-1) + α·x_t,与多数技术分析软件(K线软件)的 EMA 计算一致;若 adjust=True,会用除以权重和的方式修正,前几天的值会显著不同。技术指标回测务必与目标软件对齐,A 股常用 adjust=False。

💡 关键观察:span=2 意味着平滑非常轻——K 几乎跟着 RSV 走,D 也几乎跟着 K 走。这是「快速 KDJ」的设置,适合短线;长线 KDJ 常用 span=3 或更大。

第四步:生成金叉死叉信号

table.loc[(table['K'].shift(1) <= table['D'].shift(1)) & (table['K'] > table['D']), 'signal'] = 1
table.loc[(table['K'].shift(1) >= table['D'].shift(1)) & (table['K'] < table['D']), 'signal'] = 0
​

这两行是整个 KDJ 择时的关键。逐字解读:

• table['K'].shift(1):昨日的 K 值(把列向下移动一格)。
• table['K'].shift(1) <= table['D'].shift(1):昨日 K ≤ 昨日 D(即昨日快线还没穿过慢线)。
• table['K'] > table['D']:今日 K > 今日 D(今日快线已经在慢线之上)。
• 两个条件同时满足 → 今日发生金叉(K 从下方穿到上方),signal = 1(看涨)。

死叉对称:

• 昨日 K ≥ 昨日 D,今日 K < 今日 D → 今日发生死叉,signal = 0(看跌)。

💡 核心心法:shift(1) 在这里是关键防未来函数的设计——它确保我们用「昨日的 K/D」与「今日的 K/D」对比,信号是基于今日收盘(已知)和昨日(已知)算出的,没有用到未来信息。如果错误地用 shift(0) 与「未来某天」对比,就会引入未来函数。

signal 的两种状态

注意代码只在金叉/死叉日给 signal 赋值,其他日子的 signal 是 NaN。这意味着:

• signal == 1:金叉当日(看涨信号触发)。
• signal == 0:死叉当日(看跌信号触发)。
• signal 为 NaN:其他日子(没有信号触发)。

后续 main.py 的 groupby 会自动忽略 NaN,只统计有信号的行。这种设计很巧妙——它区分了「信号触发日」与「无信号日」,统计胜率时只看触发日。

technical.py 到此结束,函数返回带 RSV/K/D/signal 列的表。

三、main.py 逐行精读

main.py 的整体流程:遍历股票 → 算指标 → 算未来收益 → 合并 → 分组统计。

设定测试参数

day_list = [1, 5, 10, 20]
start_time = '20070101'
end_time = '20220930'
​

day_list 是要测试的未来收益周期——买入后 1 天、5 天、10 天、20 天的涨跌幅。这覆盖了短线(1 日)到中线(20 日约一个月)的多种持有期,看信号在不同周期下的胜率。

遍历所有股票 CSV

file_path = os.path.abspath(os.path.dirname(__file__)) + '/股票数据/'
file_list = os.listdir(file_path)
file_list = [f for f in file_list if '.csv' in f]
​

获取「股票数据」文件夹下所有 CSV 文件——每只股票一个 CSV。注意编码:后续 pd.read_csv 用 encoding='gbk',因为 A 股数据 CSV 常是 GBK 中文编码。

tables = []
for f in file_list:
    print(f)
    table = pd.read_csv(os.path.join(file_path, f), encoding='gbk', parse_dates=['交易日期'])
    table = technical_indicator(table)
​

逐个读取 CSV,parse_dates=['交易日期'] 让 pandas 自动把日期列解析为时间类型(后续时间筛选需要)。然后调用 technical_indicator 给这只股票算 KDJ 指标与信号。

算未来 N 日涨跌幅(shift 负数的妙用)

这是整个 main.py 最关键、最巧妙的部分:

for day in day_list:
    table['%s日后涨跌幅' % day] = table['收盘价_复权'].shift(0 - day) / table['收盘价_复权'] - 1
    table['%s日后是否上涨' % day] = table['%s日后涨跌幅' % day] > 0
    table['%s日后是否上涨' % day].fillna(value=False, inplace=True)
​

逐句拆解:

• table['收盘价_复权'].shift(0 - day):即 shift(-day),把收盘价列向上移动 day 格——也就是说,当前行的「未来 day 日收盘价」被搬到当前行。
• 未来价格 / 当日价格 - 1:即未来 day 日的涨跌幅。

shift(-N) 的效果(以 N=2 为例):
原始收盘价:  [10, 11, 12, 13, 14]
shift(-2) 后: [12, 13, 14, NaN, NaN]

第 0 行的未来 2 日涨跌幅 = 12/10 - 1 = 20%
第 1 行的未来 2 日涨跌幅 = 13/11 - 1 ≈ 18.2%
...
最后 2 行无未来数据,为 NaN
​

⚠️ 学习提示:shift(-N) 看似简单,却是时序回测里最容易出错的点——它是「把未来信息搬到当前行」,因此只能在已经发生的历史数据上用来评估信号,绝不能在实盘的「当前时点」使用——你不可能知道未来 N 天的价格。这是回测专用的算子,核心用途是给历史信号贴标签。

是否上涨 是从涨跌幅派生的二值列:True 表示涨,False 表示跌。fillna(False) 把最后 N 行的 NaN 填成 False(因为没有未来数据,默认算作不涨)。

时间筛选

table = table[table['交易日期'] >= pd.to_datetime(start_time)]
table = table[table['交易日期'] <= pd.to_datetime(end_time)]
​

只保留指定时间段内的数据。pd.to_datetime 把字符串转成时间戳,与 parse_dates 解析的日期列做布尔比较。

合并所有股票

tables.append(table)
all_table = pd.concat(tables, ignore_index=True)
​

把所有股票的 DataFrame 合并成一个大表,ignore_index=True 重新生成行索引。后续统计在这个大表上进行——这是「全市场所有股票所有时间」的样本池。

分组统计:核心的 groupby + describe

for signal, group in all_table.groupby('signal'):
    if signal == 1:
        print('\n', '=' * 10, '看涨信号', '=' * 10)
    elif signal == 0:
        print('\n', '=' * 10, '看跌信号', '=' * 10)
    print(group[[str(i) + '日后涨跌幅' for i in day_list]].describe())
​

逐句拆解:

• all_table.groupby('signal'):按 signal 列分组,返回 (signal 值, 对应的子 DataFrame) 对。NaN 自动被忽略——只有 signal 为 1 或 0 的行进入统计。
• group[[...]]:从子表里选「1日后涨跌幅」「5日后涨跌幅」等列。
• .describe():pandas 内置方法,一次性输出每列的 count/mean/std/min/25%/50%/75%/max——即样本数、均值、标准差、最小最大值与四分位数。

算胜率

for i in day_list:
    if signal == 1:
        print(str(i) + '天后涨跌幅大于0概率', '\t', 
              float(group[group[str(i) + '日后涨跌幅'] > 0].shape[0]) / group.shape[0])
    elif signal == 0:
        print(str(i) + '天后涨跌幅小于0概率', '\t', 
              float(group[group[str(i) + '日后涨跌幅'] < 0].shape[0]) / group.shape[0])
​

逐句拆解:

• group[... > 0]:从子表里筛出「未来 i 日涨跌幅 > 0」的行(看涨信号后真的涨了)。
• .shape[0]:符合条件的行数。
• group.shape[0]:该 signal 子表的总行数。
• 两者相除:胜率——看涨信号中,未来 i 日确实上涨的比例。

对看跌信号对称——统计未来 i 日确实下跌的比例。

💡 核心心法:整个胜率计算的核心是 group[条件].shape[0] / group.shape[0]——「满足条件的行数除以总行数」。这是 pandas 里算胜率/概率的标准套路,务必熟练。

四、describe 输出怎么读

.describe() 输出的统计量,每个都对应一个关键问题:

实战中最关注 mean(平均收益)与胜率:

• 看涨信号 mean 显著 > 0 且胜率 > 50%:信号有正期望值,可能可用。
• 看涨信号 mean ≈ 0 或胜率 ≈ 50%:信号与随机没区别,无效。
• 看跌信号 mean 显著 < 0 且「下跌概率」> 50%:看跌信号有效。

五、shift 正负号的区别(易错点)

整个代码里 shift 出现了三次,有正有负,务必区分:

⚠️ 学习提示:这是 pandas 时序回测里最容易出错的点。shift(1) 是「安全的」(只用了过去),shift(-N) 是「危险的」(用了未来)——前者用于实盘信号生成,后者只能用于历史回测评估。混淆两者会引入未来函数,得到虚高的回测收益。

六、运行这套代码的预期结果

把这套代码在 A 股全市场(几千只股票、十几年数据)上跑,你会观察到几个典型现象:

1. 样本数巨大:每只股票十多年,几千只股票,金叉/死叉信号触发次数可能上百万次。
2. 平均涨跌幅很小:看涨信号未来 1 日的 mean 可能只有 0.05%~0.2%(年化几个百分点),与第 02 节广发研究的结论一致——简单技术信号的 alpha 很薄。
3. 胜率略高于 50%:看涨信号 1 日后上涨概率可能约 51%~53%,看似有信号但扣交易成本(每次 0.1%~0.3%)后边际收益微薄。
4. 多日持有胜率衰减:1 日胜率 > 5 日胜率 > 10 日胜率,说明 KDJ 是短线信号,持有越久信号衰减越严重。

这些结果正是「简单技术指标单独难以稳定盈利」的实证——印证了第 02 节广发 125 指标研究的结论。

💡 关键观察:看到这个结果不要失望。这正是为什么后续章节要引入多因子模型(第 2 章)、机器学习(第 4-6 章)——它们能从海量技术指标中学习复杂组合规律,远比单一指标的简单规则有效。本节的代码骨架,正是这些复杂方法的基础。

七、如何修改这套代码测试其他指标

这套代码的优雅之处在于只改 technical.py 就能换指标,main.py 几乎不用动。例如:

• 测 MACD:把 technical_indicator 里的 RSV/K/D 计算改成 DIF/DEA 计算,signal 用 DIF 与 DEA 的金叉死叉。
• 测 RSI:把列改成 RSI 计算,signal 用 RSI 上穿 30(超买卖回升)或下穿 70。
• 测均线:算 MA5 与 MA20,signal 用 MA5 上穿/下穿 MA20。

每次只改 indicator 函数,主框架(遍历股票、算未来收益、分组统计)完全复用。这是工程上关注点分离(separation of concerns)的典范。

本节要点回顾

1. 架构分工:technical.py 只算指标与信号,main.py 只算收益与统计,两者分离、互不干扰。
2. technical.py 流程:rolling(40).min/max 算高低点 → RSV → ewm(span=2) 双重平滑得 K/D → shift(1) 对比今日与昨日 K/D 生成金叉死叉 signal。
3. main.py 流程:遍历 CSV → 调 technical_indicator → shift(-N) 算未来收益 → 合并 → groupby('signal') 分组 → describe() + 胜率统计。
4. shift 正负:shift(1) 向下移=昨日值(安全,生成信号);shift(-N) 向上移=未来值(危险,只能回测贴标签)。混淆会引入未来函数。
5. groupby 胜率套路:group[条件].shape[0] / group.shape[0],满足条件行数除以总行数,是 pandas 算概率/胜率的标准方法。
6. describe 输出:count 样本数、mean 预期收益、std 波动、四分位数分布;关注 mean 与胜率,样本太少无意义。
7. 预期结果:KDJ 金叉死叉的胜率略高于 50%、收益很薄,印证「简单技术指标单独难以稳定盈利」;这是后续引入多因子与机器学习的动机。
8. 可扩展性:只改 technical_indicator 函数就能测 MACD/RSI/MA 等其他指标,主框架完全复用——这是工程上「关注点分离」的典范。

下一章,我们离开手工指标组合,进入传统机器学习选股——从广义线性模型开始,让模型自动从因子中学习选股规律。