做量化的人大概都遇到过这种场面行情终端里跑着一条自己很满意的 EJK 指标曲线想把它的历史信号拉出来做统计结果发现终端只能看图、不能批量导出想改一个参数就得在公式编辑器里点半天。于是就有了这篇文章要讲的事——用 Python 把 EJK 指标完整复现出来。这篇内容面向三类人一是手里有自定义指标、想批量验证的量化爱好者二是刚学 pandas、想找个真实练手项目的编程新手三是做投研数据处理、需要把终端指标搬进自己流水线的工程师。我会从公式拆解、数据准备、代码实现一直讲到结果对拍和参数调优代码是完整可运行的你复制过去换一份自己的行情 CSV 就能跑。1. 先把EJK拆开看三条线各自在算什么1.1 为什么终端里的黑箱公式必须自己重写一遍行情终端提供的自定义指标公式本质上是一套轻量脚本语言它能算、能画但很难做批量化的事。你想统计过去五年里 EJK 出现过多少次共振信号、每次信号之后第 3 天和第 5 天的收盘价分布是什么样靠手工翻图根本做不完。更麻烦的是参数敏感度测试同一套逻辑跑 20 组参数组合在终端里就是 20 次手工改公式、20 次肉眼比对效率低到没有实际意义。自己用 Python 重写一遍收益不只是“能批处理”。第一计算过程变成白盒每一个中间变量你都能打印出来看出问题能定位到具体是哪一行。第二可以和多套数据源交叉验证行情终端的历史数据偶有缺口自己手里有一份清洗过的一致性数据更稳。第三能无缝接进回测框架、报表系统或者告警服务不用再靠人工盯盘。但重写这件事有个前提必须说清楚终端公式里的很多写法是“约定俗成”的比如某些函数的初值处理、分母为零时的兜底逻辑官方文档往往只有一句话。这类细节恰恰决定了你的 Python 结果和终端曲线能不能对上。所以复现的核心难点不在数学而在对齐实现细节。这也是后面要花大篇幅讲“对拍”的原因。1.2 E、J、K三条线的常见数学定义与参数含义EJK 这个名字本身就是线索它由三条线组成分别记作 E、J、K。市面上流传的版本不止一种但结构上高度相似一条趋势线 一对摆动指标。我按最常见、也最容易验证的一种构造来讲其它变体你照同样的思路替换即可。E 线本质是一条指数移动平均用来判断中期方向。计算公式是标准的 EMA 递推EMA(t) α × Close(t) (1 − α) × EMA(t−1)其中 α 2 / (N 1)N 取 12 是流传较广的默认值也有用 9 或 21 的。E 线的作用是给一个“方向过滤器”——只有当 E 线本身向上时才考虑做多方向的信号避免在下跌趋势里被反复诱多。K 线与 J 线这一对来自 KDJ 体系。先算原始随机值 RSVRSV(t) (Close(t) − LLV(Low, n)) / (HHV(High, n) − LLV(Low, n)) × 100n 通常取 9。然后 K 和 D 都做一次中国式平滑注意不是简单移动平均K(t) (2 × K(t−1) RSV(t)) / 3D(t) (2 × D(t−1) K(t)) / 3J 线是 K 和 D 的线性组合J(t) 3 × K(t) − 2 × D(t)J 的特点是没有上下界波动比 K、D 剧烈得多所以常被当作“情绪放大器”用。EJK 里的 K 用的就是上面这个 K而 J 用 J这样三条线共享同一套底层数据不会出现时间错位。信号怎么出最典型的一种是共振——E 线向上方向确认 J 线上穿 K 线短期动能确认。两个条件同时满足才记为一次信号。这个设计的好处是能过滤掉大量单边下跌中的假金叉代价是信号数量会显著减少回看历史时经常几个月才出一次。1.3 复现前必须想清楚的三个取舍在动手写代码之前有三个决定会直接影响最终结果选错了后面全白干。取舍一复权方式。用不复权数据算指标遇到除权除息那天就会出现价格跳空RSV 会被瞬间打到极端值K 线和 J 线跟着一起失真。我的建议是统一用前复权数据并且把复权因子固定下来——不要每次跑都重新拉一遍最新复权价否则历史信号会随新数据发布而变化做出来的统计没有任何可比性。取舍二K 线周期。日线、60 分钟、周线算出来的 EJK 完全是三条不同的曲线。终端的公式本身不绑定周期你在哪个周期图表上加载它就在哪个周期算。Python 这边要在代码里显式声明周期并且把时间戳规范化比如统一到交易日收盘时刻否则跨周期数据混在一起会算出鬼东西。取舍三平滑函数的实现。K 和 D 用的中国式平滑SMA(X, N, M)是递推式不是窗口平均。很多人第一反应是rolling(3).mean()结果第一位数就错了而且误差会沿着时间轴一直累积。这个坑我在第 3 节会用代码详细说明怎么避开。把这三个取舍定下来后面的工作就变成纯粹的工程问题了。2. 数据准备与环境搭建让数字能对得上2.1 数据源的三条硬性要求复现指标最怕的不是算错而是“算对了但和数据源不一致”。选数据源的时候我只看三条。第一条必须有完整的高开低收。EJK 里 RSV 同时用到 High、Low、Close缺任何一个都算不出来。有些免费数据接口只提供收盘价这类数据直接排除别想着用收盘价代替高低价算出来的 RSV 会严重失真。第二条时间轴要连续且可对齐。停牌、节假日、半日市这些情况会导致某些日期缺行。如果直接rolling(9)窗口滑动缺行会让窗口跨过更长的真实时间跨度指标就变形了。稳妥做法是先按交易日历重采样补齐停牌日用前一日收盘价填充并在数据里打个标记回看信号时把停牌区间排除掉。第三条复权口径要一致。前复权、后复权、不复权三者的价格序列完全不同。我一般选前复权因为终端的默认显示也是前复权对拍最省事。但要注意前复权的历史价格会随着新的除权事件变化所以生产环境里我会把复权后的数据快照存下来做版本管理。数据获取渠道上用开源财经数据接口、券商提供的行情文件、自己维护的数据库都行本文的代码对数据来源没有依赖你只要把它整理成下面这个格式即可列名类型说明datedatetime64交易日统一到当日 00:00:00openfloat开盘价前复权highfloat最高价前复权lowfloat最低价前复权closefloat收盘价前复权volumefloat成交量可为空2.2 用一份统一的CSV把坑提前挡掉我习惯在项目根目录放一个data/文件夹里面只放经过清洗的 CSV命名规则是代码_周期_复权方式.csv比如600000_D_QFQ.csv。这样做的原因是复现类项目最容易乱的地方就是数据版本今天用 A 接口、明天用 B 接口结果对不上还要回头查是哪份数据的问题。清洗脚本我一般写成一个独立的prepare.py它干四件事读原始数据、按交易日排序去重、检查异常值价格为 0 或负、最高价小于最低价、收盘价超出高低区间、输出标准 CSV。异常值检查这一步别省我见过数据源把某个交易日的最高价写成 0导致 HHV 整个窗口塌掉指标曲线出现一个深坑排查了半天才定位到是脏数据。import pandas as pd def clean_ohlc(raw: pd.DataFrame) - pd.DataFrame: df raw.copy() df[date] pd.to_datetime(df[date]) df df.sort_values(date).drop_duplicates(date) # 基础合法性检查 mask_valid ( (df[[open, high, low, close]] 0).all(axis1) (df[high] df[low]) (df[close] df[high]) (df[close] df[low]) ) bad df.loc[~mask_valid, date].tolist() if bad: print(f发现 {len(bad)} 条异常记录日期{bad[:5]}) df df[mask_valid].reset_index(dropTrue) return df[[date, open, high, low, close, volume]]这段代码不长但它是后面所有计算的地基。跑完打印一下行数和首尾日期跟自己预期的区间核对一遍三十秒的事能省掉后面几小时的困惑。2.3 环境与依赖依赖很少pandas 加 numpy 足够画图用 matplotlib不装也能跑。pip install pandas numpy matplotlib我用的版本是 pandas 2.x、numpy 1.26更低版本也能跑注意 pandas 1.x 里ewm的参数名一致没有兼容问题。如果你在 Windows 上跑注意 CSV 的编码用encodingutf-8-sig读比较稳。3. 核心代码实现从EMA到J线再到信号确认3.1 E线的指数平滑实现与初值陷阱E 线的公式简单但初值处理有讲究。pandas 的ewm有两个关键参数span和adjust。spanN就是按 α 2/(N1) 计算衰减系数这个没问题。adjust才是坑所在。当adjustTrue默认值时pandas 会用加权平均的方式处理前期数据前几个值是无偏修正后的结果当adjustFalse时用的是纯递推第一个值直接取序列的第一个观测值。行情终端里的 EMA 基本都是纯递推所以必须显式写adjustFalse否则前面几十根 K 线和终端对不上。import pandas as pd import numpy as np def ema(series: pd.Series, n: int) - pd.Series: 标准指数移动平均等价于终端的 EMA(X, N) return series.ewm(spann, adjustFalse).mean()还有一个隐含假设序列开头没有 NaN。如果你喂进去的数据前面有缺失值ewm会跳过 NaN 并从第一个有效值开始递推这跟终端的处理可能不同——终端通常是取第一根 K 线的收盘价作为 EMA 初值。所以我的做法是先用ffill()把 NaN 填掉再算 EMA保证起点对齐。3.2 J线的构造与中国式SMA的正确写法这是整个复现里最容易错的地方值得单独拎出来讲。行情终端里常见一个函数SMA(X, N, M)它的定义是Y(t) [M × X(t) (N − M) × Y(t−1)] / N注意它跟简单移动平均完全不是一回事。以SMA(RSV, 3, 1)为例展开后是Y(t) (1 × RSV(t) 2 × Y(t−1)) / 3也就是对当前值和上一期结果做加权递推。用rolling(3).mean()算出来的是最近三期的算术平均两者数值差异很大而且在趋势段会持续偏离。好在 pandas 的ewm支持直接指定 α。把上面的公式改写成递推形式Y(t) α × X(t) (1 − α) × Y(t−1)其中 α M / N所以SMA(X, N, M)等价于X.ewm(alphaM/N, adjustFalse).mean()。这个对应关系记住以后遇到任何终端里的 SMA 都能一行搞定。def sma_cn(series: pd.Series, n: int, m: int 1) - pd.Series: 中国式SMA(X, N, M)递推加权非窗口平均 alpha m / n return series.ewm(alphaalpha, adjustFalse).mean()RSV 的分母还有个边界情况当窗口内最高价等于最低价比如连续一字板或者长期停牌后复牌分母为 0除法会得到 inf 或 NaN。终端的处理通常是取前值或者取 50我选 50因为 50 代表中性位置不会给 K 线造成突变。这个选择要写进注释方便以后别人接手时知道你为什么这么写。def kdj(close: pd.Series, high: pd.Series, low: pd.Series, n: int 9, m1: int 3, m2: int 3): llv low.rolling(n, min_periods1).min() hhv high.rolling(n, min_periods1).max() rng (hhv - llv).to_numpy() # 分母为0时取50避免 inf / NaN 污染整条曲线 rsv np.where(rng 0, (close - llv) / np.where(rng 0, rng, 1) * 100, 50.0) rsv pd.Series(rsv, indexclose.index) k sma_cn(rsv, m1, 1) d sma_cn(k, m2, 1) j 3 * k - 2 * d return k, d, jmin_periods1这个参数也有讲究。默认情况下rolling(9)前 8 个值是 NaN指标要到第 9 根 K 线才有数。但终端通常是从第一根 K 线就开始算窗口不满时用已有数据。设成 1 就能对齐这个行为代价是前几根的信号参考价值不大——所以我在实际使用中会把前 30 根 K 线的信号直接丢弃避免用不完整窗口算出来的值做判断。3.3 EJK三线合成与信号判定有了 E、K、J组装就是几行代码的事。信号判定我用两个布尔序列相与的方式写好处是中间结果全部保留方便事后逐条核对。def ejk(df: pd.DataFrame, n_e: int 12, n_k: int 9, m1: int 3, m2: int 3) - pd.DataFrame: out df.copy() out[E] ema(out[close], n_e) k, d, j kdj(out[close], out[high], out[low], n_k, m1, m2) out[K], out[D], out[J] k, d, j out[E_up] out[E] out[E].shift(1) out[cross_up] (out[J] out[K]) (out[J].shift(1) out[K].shift(1)) out[signal] out[E_up] out[cross_up] return outcross_up这一行是标准的“上穿”写法当期 J 在 K 之上且上一期 J 不在 K 之上。用shift(1)做比较是唯一稳妥的方式千万别用diff()的符号判断J 和 K 的差值符号在震荡区会来回跳容易误判。另外加一个可选的反向信号E 线向下 J 下穿 K会让统计更对称但要注意这不是“做空信号”只是指标的另一侧表现具体怎么用取决于你自己的研究框架。3.4 完整可运行脚本把上面的片段拼起来加一份模拟数据生成器可以直接跑通。真实使用时把make_demo_data()换成你的 CSV 读取即可。import pandas as pd import numpy as np def ema(s, n): return s.ewm(spann, adjustFalse).mean() def sma_cn(s, n, m1): return s.ewm(alpham / n, adjustFalse).mean() def kdj(close, high, low, n9, m13, m23): llv low.rolling(n, min_periods1).min() hhv high.rolling(n, min_periods1).max() rng (hhv - llv).to_numpy() safe np.where(rng 0, rng, 1.0) rsv np.where(rng 0, (close.to_numpy() - llv.to_numpy()) / safe * 100, 50.0) rsv pd.Series(rsv, indexclose.index) k sma_cn(rsv, m1, 1) d sma_cn(k, m2, 1) j 3 * k - 2 * d return k, d, j def ejk(df, n_e12, n_k9, m13, m23): out df.copy() out[E] ema(out[close], n_e) k, d, j kdj(out[close], out[high], out[low], n_k, m1, m2) out[K], out[D], out[J] k, d, j out[E_up] out[E] out[E].shift(1) out[cross_up] (out[J] out[K]) (out[J].shift(1) out[K].shift(1)) out[signal] out[E_up] out[cross_up] return out def make_demo_data(n500, seed42): rng np.random.default_rng(seed) ret rng.normal(0.0004, 0.018, n) close 10 * np.exp(np.cumsum(ret)) noise_h np.abs(rng.normal(0, 0.008, n)) noise_l np.abs(rng.normal(0, 0.008, n)) high close * (1 noise_h) low close * (1 - noise_l) open_ close * (1 rng.normal(0, 0.004, n)) dates pd.bdate_range(2022-01-03, periodsn) return pd.DataFrame({ date: dates, open: open_, high: high, low: low, close: close, volume: rng.integers(1e5, 1e6, n) }) if __name__ __main__: raw make_demo_data() res ejk(raw) res res.iloc[30:].reset_index(dropTrue) # 丢掉窗口不满的前30根 print(f样本 {len(res)} 根K线信号 {int(res[signal].sum())} 次) print(res.loc[res[signal], [date, close, E, K, J]].head(10).to_string(indexFalse)) res.to_csv(ejk_output.csv, indexFalse, encodingutf-8-sig)跑完会输出信号计数和前十条信号明细ejk_output.csv里保留了全部中间变量。后续做统计时直接读这个文件就行不用再重算。4. 对拍验证怎么证明你的结果和终端一致4.1 误差来源清单复现完之后第一件事不是回测是对拍。我整理了一份常见误差来源按排查优先级排序误差现象可能原因排查方法整体曲线形状一致但数值有小幅偏移EMA 初值或adjust参数不同打印前 5 根 K 线的 E 值与终端逐根比对K 线在震荡区差异明显用了rolling().mean()而非递推 SMA检查sma_cn是否用了ewm(alpha...)出现孤立尖刺数据源有除权跳空未复权检查信号日前后是否有除权事件前 20 根完全对不上后面正常窗口不满时的处理方式不同把min_periods设为 9 再试一次信号数量多出一截上穿判定用了diff()而非shift逐条打印 cross_up 的中间值某一段区间所有值都是 50RSV 分母为 0 的兜底逻辑触发检查该区间是否连续一字板或长期停牌这张表我用了很久绝大多数对不上的情况都能在里面找到对应项。真正麻烦的是“看起来对、但差一点点”这种往往是复权口径不一致导致的肉眼看不出来得逐日做数值差分才能发现。4.2 逐段对拍的具体做法对拍不要一上来就比整条曲线那样差异太分散看不出规律。我的做法是抽样 分段。先抽 3 个区间一段单边上涨、一段横盘震荡、一段急跌。每段取 20 根 K 线把终端的 E、K、J 数值手工抄出来或者从终端导出到表格和自己的 CSV 逐行对齐。重点看三列E 值、K 值、J 值。E 值能对上说明 EMA 实现没问题K 值能对上说明 RSV 和递推 SMA 都对J 值是 K、D 的线性组合前两个对了它自然对得上。然后算绝对误差和相对误差。经验值是E 线偏差应该小于 0.1%K 线偏差小于 0.5J 线偏差小于 1.0。如果某一根 K 线突然偏差很大往前翻一到两周的数据基本都能找到原因除权、停牌补数据、数据源修正历史。最后做一个交叉验证用两套不同来源的数据各自算一遍看信号日期的重合度。重合度高说明指标逻辑本身稳定不依赖某个数据源的特性重合度低就要警惕了大概率是数据清洗环节出了问题而不是代码。4.3 几个高频问题与解决思路问题一终端里 K 线一直有值我的代码前 8 根是 NaN。这是rolling默认min_periodswindow造成的改成min_periods1即可。但要注意前几根的信号不具备参考价值统计时应该剔除。问题二信号数量比终端多很多。八成是上穿判定写成了J K而不带上期条件。J 和 K 在粘合区会反复穿越只判断当期大小关系会把一次金叉算成好几次。问题三某只票的指标曲线在某天变成一条水平线。大概率是那天之后连续停牌数据被前值填充RSV 分母恒为 0走了兜底逻辑。这类区间在做信号统计时必须剔除否则会严重污染结果。问题四换了一份数据历史信号全变了。前复权数据会随新的除权事件回溯调整。解决办法是把复权后的快照固定下来或者统一用后复权数据做研究后复权不会因新事件变化。问题五pandas 报SettingWithCopyWarning。这是我早期写代码常犯的根源是链式索引。养成习惯进函数第一行就out df.copy()后面所有赋值都在副本上做。5. 参数调优与实战注意事项5.1 参数敏感度实测的方法EJK 的可调参数有三个E 线的周期n_e、KD 的窗口n_k、以及平滑系数m1、m2。参数一变信号数量可能差好几倍所以不做敏感度测试就下结论是很危险的。我的做法是网格扫描 信号密度对比。以n_e取 9/12/21、n_k取 9/14/21 做九组组合每组统计三个数信号总数、信号平均间隔天数、信号在上涨段和下跌段的分布比例。如果某组参数在下跌段也有大量信号说明方向过滤没起作用可以直接排除。这里要提醒一句参数优化的目标是找到稳健区间不是找历史表现最好的那组。我一般会看参数微调 ±20% 时信号数量变化是否平缓变化剧烈的组合直接放弃因为这种参数在未来大概率不稳定。5.2 这个指标能做什么、不能做什么必须说清楚边界。EJK 本质是对历史价格序列的平滑与变换它反映的是过去不含任何前瞻信息。它能做的是把杂乱的 K 线压缩成可比较的形态、提供一个客观的信号触发条件、方便做大规模历史统计。它做不到的是预测未来、识别基本面变化、判断消息面影响。任何声称某个指标能稳定盈利的说法都值得怀疑。我见过太多人把指标信号当成交易指令直接执行结果在震荡市里被反复打脸。指标的正确用法是筛选 佐证先用它缩小观察范围再结合成交量、板块结构、自身的研究框架做二次判断。还有一点信号统计里如果出现“事后 N 日涨幅”这类指标一定要明确它是描述性统计而非策略回测。用未来数据算出来的收益分布天然带有幸存者偏差尤其是样本量只有几十次的时候统计意义非常弱。5.3 我踩过的几个坑坑一用rolling().mean()冒充 SMA。这是最早踩的也是最致命的。当时对着终端的 K 线盯了一下午一直以为是自己数据有问题最后把两个函数在小样本上逐行打印才看出来。教训是任何从终端搬过来的函数先用 5 个数字手算一遍验证再上真实数据。坑二前复权数据每次重拉。有段时间发现同一套代码跑两次结果不一样查了很久才意识到数据源每天都会用最新价格重新算前复权因子。后来改成快照存档问题消失。坑三忘了处理停牌。一只票停牌三个月后复牌RSV 窗口里全是相同价格分母为 0K 线直接变成一条直线后面的信号全废。现在的清洗脚本里专门加了一条规则连续相同收盘价超过 5 天标记为疑似停牌统计时排除。坑四信号统计没剔除窗口不满期。前 30 根 K 线的窗口不完整算出来的指标值参考价值很低但信号照样会触发。我现在的做法是统一加一个warmup30的裁剪所有统计从第 31 根开始。坑五在日线上写死参数然后套到分钟线。9 这个数字在日线上是两周左右的窗口放到 5 分钟线上只有 45 分钟含义完全不同。换周期必须重新调参这一步不能偷懒。最后一个实操心得把整个流程封装成一个函数输入原始 DataFrame 和参数字典输出带全部中间变量的结果表。这样做的好处是换标的、换周期、换参数都只改一行调用而且每次输出的结构完全一致方便批量跑和横向对比。我现在的写法是run_ejk(df, params)返回一个 dataclass里面包含结果表、信号列表、参数快照和运行时戳归档起来一目了然。这套结构跑了几十个标的之后再回头查任何一个历史结果都能快速定位比散落一地的脚本强太多。