技术指标这种东西,听起来高深,拆到底就是几根移动平均线、最高价最低价的范围、涨跌幅度的统计。我在本地跑回测的时候,最大的痛苦不是写策略,而是把行情软件里现成的指标公式,一条一条搬到自己的Python环境里。搬一次骂一次:不同软件的函数名不一样、返回值顺序不一样,甚至同一个MACD,有的按拉齐算法、有的按权重算法,结果能差出一个小数点。后来我想通了一个事:不管界面怎么变,指标公式最后落地的单位无非是REF、MA、EMA、HHV、LLV、SUM、ABS、MAX、MIN、COUNT、CROSS、IF这一二十个基础算子。只要把这套算子用numpy和pandas原样实现一遍,剩下的工作就是像搭积木一样把指标拼出来。于是我用大约70行代码,封装了一套极简指标库,主流行情软件里MACD、KDJ、RSI、BOLL这类常用指标直接套用公式就能算,甚至连它们公式语法里那些生僻的写法,也能通过组合算子还原出来。这里就把这套代码、使用方法和踩坑经验全部摊开来讲,适合量化新手,也适合在多个平台间迁移指标的老手。
1. 为什么“70行”就能覆盖大部分技术指标公式
1.1 不同行情软件里的公式,其实是同一套积木
打开任何一个行情软件,自带的公式系统虽然名字不一样,语法也各有细节,但翻开它们的帮助文档就会发现,绝大多数技术指标都是围绕K线数据(开、高、低、收、量)做时间序列运算。MACD是两条指数均线的差值再平滑一次,KDJ是近N日最高价和最低价区间里的位置百分比,RSI是对涨跌幅做加权平均再算比例,BOLL是均线加上固定倍数的标准差。这些定义在数学上是统一的,差的只是函数名称和参数写法。比如移动平均,A软件叫MA,B软件也叫MA,还有的软件里叫“均线”;最高值,有叫HHV的,有叫HG的;引用前N周期,有叫REF的,也有叫“前前”的。名称乱,但数学定义一致。这意味着只要在Python里实现这些“原子算子”,就能用同一套代码翻译不同平台的公式。这种“一套积木拼所有”的思路,不是我的原创,很多量化框架也这么干,只不过它们往往做得太重,封装了很多复杂对象。我更想用最小的代码量,直击算法本质。
1.2 基础算子的覆盖力:一张表看懂
我统计了一下常见指标公式里出现的算子频率,得到这么一张高频算子清单。移动平均类有MA、EMA、SMA(一种递归均线),统计类有HHV、LLV、SUM、STD、COUNT,逻辑类有ABS、MAX、MIN、IF、CROSS、REF。光凭这几个算子,就能组合出常见指标里90%以上的公式。比如MACD只用了EMA和减法,KDJ只用了HHV、LLV、SMA,RSI只用了SMA和IF,BOLL只用了MA和STD。这也就解释了为什么代码可以精简到70行——很多看似复杂的指标,本质上只是这些原子算子的不同嵌套。我把这些算子用numpy写出来后,每一个都对应着一个简单的数学动作,而不是某个特定的指标。以后想加自己的新指标,只要能用这些算子表达,就能直接跑出来。这是最重要的思维转变:从记忆某指标公式,变成掌握算子语法。
| 算子 | 数学含义 | 典型用途 |
|---|---|---|
| REF(S, n) | 序列S前移n周期 | 取前值、构建交叉条件 |
| MA(S, n) | n周期算术平均 | 均线、BOLL中轨 |
| EMA(S, n) | n周期指数衰减平均 | MACD快慢线 |
| SMA(S, n, m) | 递归加权平滑 | KDJ、RSI的平滑环节 |
| HHV(S, n) | n周期最高值 | KDJ中的用途 |
| LLV(S, n) | n周期最低值 | KDJ中的用途 |
| SUM(S, n) | n周期求和 | 成交量累计、MONEY指标 |
| STD(S, n) | n周期标准差 | BOLL上下轨 |
| COUNT(cond, n) | n周期内条件成立次数 | 连续涨跌天数 |
| CROSS(a, b) | a上穿b | 金叉死叉信号 |
| IF(cond, a, b) | 条件成立取a,否则取b | 分段计算、过滤信号 |
2. 核心数据结构与基础算子设计
2.1 为什么选择numpy数组 + pandas rolling
写指标库之前要先定数据结构。使用原生Python列表当然能算,但循环几千上万次不仅慢,代码还啰嗦。numpy数组是搞数值计算的标配,加上pandas的rolling窗口,移动平均、最高最低值这些操作都是一行搞定。更重要的一点是,技术指标天然喜欢“错位引用”:今天的指标可能需要三五天前的数据,numpy的切片正好可以把序列整体“挪位置”。我的方案是:所有输入输出都是numpy一维float数组,长度等于K线根数;用NaN表示“还没有有效值”的周期,比如前N-1根的MA。这样做的好处是干净,所有算子都对任意一维数组生效,组合起来不会出现维度对不上的情况。缺点也有:不能像DataFrame那样自动对齐索引,但只要保证K线顺序固定,不随手排序,就没有对齐问题。
2.2 基础函数逐个拆解
先列最常用的算子,每个算子的实现思路如下。
REF(S, n)是S序列向前平移n个周期,今天得到的值是n个周期前那个值。我用out[n:] = S[:-n],前面的位置补NaN。这个算子是一切“金叉、死叉、前值判断”的基础。
MA(S, n)是普通移动平均,直接对S做rolling(n).mean()。注意pandas默认就是窗口内均值,能用。
EMA(S, n)是指数移动平均,市面通用算法都要求权重按2/(n+1)衰减,pandas的ewm(span=n, adjust=False)正好对应,不需要自己写递归。
SMA(S, n, m)是很多新手栽跟头的地方。它并不是统计学里的简单平均,而是一种递归平滑:Y = (M * X + (N - M) * Y') / N,其中Y'是上一根K线的Y值。因为存在递归,numpy没法直接向量化,只能用for循环。好在K线量少,几千根算下来在毫秒级,完全没问题。
HHV/LLV是滚动窗口最大值和最小值,我用rolling(n).max()/.min(),窗口前补NaN。
SUM/STD是滚动求和和滚动标准差。这里有个容易忽略的细节:软件公式里的STD默认是除以N(总体标准差),而pandas默认是除以N-1(样本标准差)。如果不把ddof设为0,BOLL上下轨会和软件差不少。
COUNT、ABS、MAX、MIN、IF、CROSS这几类更像逻辑运算。COUNT是对条件序列的滚动求和,ABS是取绝对值,MAX/MIN比较两个序列或标量,IF实现条件选值,CROSS判断上穿。
把这些函数放在一起,就是一套“公式解释器”的核心。
2.3 这里必须提三个容易理解错的运算
首先是EMA。很多人以为EMA可以通过递归公式写,但初始值怎么给是个问题。有的软件首根取第一个收盘价,有的软件首根取0,有的软件前几根不输出。如果直接用pd.Series(S).ewm(span=n, adjust=False).mean(),它的起始点会以第一个有效值为基准,计算出来的序列在初始几个周期可能与某些软件有细微差异,但经过一段稳定期后收敛,实用问题不大。其次是SMA。这个函数在行情软件的公式里太常用了,但和EMA完全不同。EMA的系数固定是2/(N+1),而SMA的系数是M/N。KDJ里的RSV平滑就是SMA(RSV,3,1),RSI里的涨跌平滑也是SMA(UP,14,1)。写错一个参数,结果就完全不对。第三是CROSS。CROSS(A,B)在软件里的含义是“A从下方上穿B”,也就是上一周期A小于等于B,当前周期A大于B。我的实现用两个前移序列比较,特别注意上一周期相等的情况也算“先小于等于再大于”,这样才能覆盖到紧贴穿越的边界。逐根验证下来,它和软件里金叉位置完全一致。
3. 70行核心代码实现
3.1 完整代码清单
下面这段代码就是核心部分。为了便于阅读,我加了注释,去掉注释后约70行。它包含前面说的所有基础算子,以及MACD、KDJ、RSI、BOLL四个常用指标组合。
import numpy as np import pandas as pd # ========= 核心算子 ========= def REF(S, n): out = np.full(len(S), np.nan) out[n:] = S[:-n] return out def MA(S, n): return pd.Series(S).rolling(n).mean().to_numpy() def EMA(S, n): return pd.Series(S).ewm(span=n, adjust=False).mean().to_numpy() def SMA(S, n, m): out = np.full(len(S), np.nan) y = np.nan for i, x in enumerate(S): if np.isnan(x): y = np.nan elif np.isnan(y): y = x else: y = (m * x + (n - m) * y) / n out[i] = y return out def SUM(S, n): return pd.Series(S).rolling(n).sum().to_numpy() def HHV(S, n): return pd.Series(S).rolling(n).max().to_numpy() def LLV(S, n): return pd.Series(S).rolling(n).min().to_numpy() def STD(S, n): return pd.Series(S).rolling(n).std(ddof=0).to_numpy() def MAX(a, b): return np.maximum(a, b) def MIN(a, b): return np.minimum(a, b) def ABS(S): return np.abs(S) def COUNT(cond, n): return pd.Series(np.where(cond, 1.0, 0.0)).rolling(n).sum().to_numpy() def CROSS(a, b): pa = np.concatenate([[np.nan], a[:-1]]) pb = np.concatenate([[np.nan], b[:-1]]) return (a > b) & (pa <= pb) def IF(cond, a, b): return np.where(cond, a, b) # ========= 指标组合 ========= def MACD(C): dif = EMA(C, 12) - EMA(C, 26) dea = EMA(dif, 9) return dif, dea, (dif - dea) * 2 def KDJ(H, L, C): lv = LLV(L, 9) hh = HHV(H, 9) rsv = np.where(hh - lv != 0, (C - lv) / (hh - lv) * 100, 0) K = SMA(rsv, 3, 1) D = SMA(K, 3, 1) return K, D, 3 * K - 2 * D def RSI(C, n=14): diff = np.diff(C, prepend=C[0]) # 首周期涨跌视为0 up = IF(diff > 0, diff, 0.0) down = IF(diff < 0, -diff, 0.0) su = SMA(up, n, 1) sd = SMA(down, n, 1) return su / (su + sd) * 100 def BOLL(C, n=20, p=2): mid = MA(C, n) std = STD(C, n) return mid + p * std, mid, mid - p * std调用的时候,只需要把K线数组准备出来:
# df 需要包含 open, high, low, close, volume 五列 C = df['close'].to_numpy(dtype=float) H = df['high'].to_numpy(dtype=float) L = df['low'].to_numpy(dtype=float) dif, dea, bar = MACD(C) K, D, J = KDJ(H, L, C) rsi = RSI(C, 14) up, mid, low = BOLL(C)我特别说明一下KDJ里的处理:当HHV和LLV相等时,也就是最高和最低价格区间为零(比如一字涨停或长期停牌后恢复的极端情况),公式会出现分母为零。这里我强制除号结果为0,可以避免NaN污染整个序列。实际中也可以填50,但0更接近某些软件的默认输出。
3.2 怎样“翻译”一条软件公式
有了这套算子,翻译公式就很机械了。比如翻看软件里BOLL的公式源码,通常写成:
MID:=MA(CLOSE,N); UPPER:=MID+P*STD(CLOSE,N); LOWER:=MID-P*STD(CLOSE,N);翻译到我们的代码就是:
mid = MA(C, n) upper = mid + p * STD(C, n) lower = mid - p * STD(C, n)再比如某软件里MACD公式:
DIF:EMA(CLOSE,12)-EMA(CLOSE,26); DEA:EMA(DIF,9); MACD:(DIF-DEA)*2;对应代码就是:
dif = EMA(C, 12) - EMA(C, 26) dea = EMA(dif, 9) macd = (dif - dea) * 2几乎是一行对一行,唯一需要留意的是一般软件里“CLOSE”指收盘价序列,我们这里用C变量代替。真正碰到那种有“BACKSET”“PEAK”之类高级函数的指标,可能得自己补一些算子,但常用指标基本不用。
3.3 真实K线校验的完整流程
代码写出来,下一步必须验证和软件计算是否一致。我的做法是:先导出一段指数日线数据,包含日期、开高低收量,然后分别用行情软件和Python代码计算同一指标。校验时要注意三点:第一,数据必须完全一致,包括复权设置;第二,要逐周期对比,不能只比最后一个值;第三,指标输出的对齐顺序要和K线时间顺序一致,不要把第0根当最新值。我把DIF、DEA、K、D、J这些序列都打印出来,逐根比对,一般到第20根之后误差就在1e-6以内。前几根如果不一致,多半是初始值规则不同,可接受。比如EMA在软件里可能前几根显示空值,而pandas会从第1根就给出值;KDJ的起始几根也可能因SMA初值不同有偏移。这在实际使用中不会影响买卖点,因为交易系统都会过滤“信号太新”的情况。
4. 实操中容易踩的坑与排查记录
4.1 MA、EMA、SMA三兄弟分不清
很多人上来就用MA代替EMA,计算MACD时把DIF写成MA(CLOSE,12)-MA(CLOSE,26)。这完全错误,因为MA是等权平均,EMA是衰减加权平均,对最近价格的敏感度完全不同。MACD公式里明确要求EMA,而KDJ里用的是SMA,不是MA、也不是EMA。这三个函数对应三种平滑风格:MA反映整体平均,反应最慢;EMA强调近期,反应中等;SMA的灵敏度取决于M和N的比值,M/N越大越贴近当前值。把它们混用的结果就是指标曲线形态相似但数值偏离,金叉死叉位置可能差好几根K线。写代码时,函数命名必须保留这三个名字,甚至注释里标注“此SMA非统计SMA”,防止自己回头也看混。
4.2 NaN传播导致全序列失效
技术指标的起始位置经常是NaN,比如MA(20)的前19个值都无效。如果后续计算里没有处理NaN,减法、除法会让NaN一路传染,导致最终指标大面积无值。最典型的场景是KDJ分母为0时我提到过的处理;另一个典型是RSI计算中diff的第一根,如果用np.diff默认会少一个元素,如果处理不对,后面所有值都会错位。我建议做指标前统一把开高低收转成np.float64,并检查数据中是否存在缺失值;如果数据源本身有停牌导致的空值,先填充或剔除,再喂给算子。否则一个NaN就能毁掉整根曲线,而且肉眼很难发现。
4.3 复权、涨跌停与数据对齐问题
千万不要用未复权价格直接算指标然后拿去和软件对比。很多行情软件默认前复权显示K线,而导出的原始CSV可能是不复权。前后复权下的最高最低价不同,HHV、LLV的结果自然不同。另外,如果是从多个数据源拼接K线,要确认每根K线的时间戳是自然日还是交易日,是否包含集合竞价数据。我在一次迁移中发现,某平台1分钟数据第一根包含了竞价价格,而行情软件把这根排除掉,导致开盘第一根的均线指标对不上。解决办法很简单:统一数据清洗标准,开盘价按集合竞价价格保留,最高最低价按连续竞价时段取值,所有对比数据只使用同一来源。
4.4 CROSS边界条件的处理
CROSS看起来简单,但边界条件坑很大。如果昨天A=B,今天A>B,这算不算上穿?软件公式里通常算。如果昨天A和B都是NaN,今天A与B有比较值,应该输出False,因为信号没有历史。还有,两个序列在某一周期同时跳空,导致直接用切片比较时把昨天的NaN和今天的数字比较,结果会是False,但有时我们希望把窗口缺失当作“没有状态”。我在CROSS实现里用np.concatenate把序列向右移一位,并把第一位置为NaN,这样NaN参与比较永远返回False,逻辑清晰。另外,如果只用(a > b) & (prev_a <= prev_b),当prev_a是NaN时,Python里比较结果是False,整体False,正合我意。这个方法处理了大多数情况。
4.5 性能与精度的矛盾
一般日线数据几千根,用for循环写SMA完全没问题,但如果你把1分钟数据全部加载进来,一年可能有十几万根,此时SMA的Python循环会明显变慢。我建议在两个方向优化:一是把高频数据先降采样成日线再算部分指标;二是如果实在要按分钟算,把SMA改成numba jit或者用scipy.signal.lfilter实现递归滤波。lfilter把递归表达式转换成差分方程,能极大提速。比如SMA的递推式Y = (m*X + (n-m)*Y')/n,转换成lfilter的分子分母系数可以一行搞定。不过对普通场景,for循环已经够用,不要提前优化,保持代码可读性更重要。
5. 从指标到策略:还能继续往哪走
5.1 把指标结果直接喂给回测引擎
这套函数输出的是numpy一维数组,与K线一一对应。做策略回测时,可以直接生成买卖信号数组。比如MACD金叉:DIF上穿DEA,对应的CROSS(dif, dea)就是信号;再配合其他过滤条件,比如DIF大于0才做多,用IF和逻辑与组合就行。我通常会把指标算完后直接放进回测框架的DataFrame里,然后逐根遍历或批量计算仓位。因为这只是一个算术层,不依赖任何GUI,所以跑批量回测很方便。
5.2 参数批量搜索的妙用
行情软件里改一个参数需要点鼠标,还得程序刷新。在这套代码里,参数就是函数入参,直接循环遍历就行。比如想测RSI的周期参数从6到24哪个效果最好,只需要写个循环,每次算一遍RSI并跑一个简单回测。一次能测几百组参数,这在行情软件里几乎做不到。我自己做参数扫描时,会把指标计算和回测分开,先用这套代码批量生成所有参数组合下的指标序列,存成中间文件,再统一做回测。这样避免每次回测都重复算指标,能节省不少时间。
5.3 扩展更多算子,实现公式翻译自由
这套70行代码不是终点,而是一个底层基础。行情软件的公式库里还有BACKSET(未来数据)、PEAK、TROUGH、ZIG等。这些算子大多可以用原始数据配合循环实现。比如ZIG通常会对未来数据作“之字转向”,本质上是寻找局部极值点,这需要先向后看一段数据。由于这类指标容易用到未来信息,做回测时要格外注意,只有用历史数据可复现的版本才能接入策略,否则会产生前视偏差。我建议先把基础算子练熟,再按需补充高级算子;不要一上来就想覆盖全部公式,那样代码量会失控,也不利于维护。反过来,有了这套基础后,你就拥有了一台“极简公式翻译机”,大部分常用指标迁移都能快速完成。
我个人最早写这套东西的时候,花了一个晚上就在本地把常用的三十多个指标全部翻译了一遍,但真正用起来,最值钱的不是“能跑”,而是“对得上”。记得有次我拿同一根K线,用Python算的MACD柱子和行情软件显示的柱子差了0.03,排查半天发现是行情软件把MACD柱子显示成DIF-DEA,没有乘2,而公式源码里明确乘了2。这种细节,只有逐根对比才能发现。所以我把代码固定下来之后,每次拿到新数据源,都会先跑一遍指标一致性检查,把常用指标的最近20个值打印出来和行情软件对照,确认无误后再接入策略。
最后分享一个小技巧:把基础算子全放进一个模块,新增指标时只写公式层,例如定义MACD(C, fast=12, slow=26, sig=9),这样即便以后换到任何平台,你都能用同一套Python函数快速验证新想法。这也是我觉得这70行代码的最大价值:不是代码本身值钱,而是它把分散在各软件里的“黑盒指标”变成了你手里明明白白的算术公式。