向量化不总是更快？一次技术指标重写带来的认知刷新

引子：一个”顺手”的重构

事情要从 stock-mcp-server 的技术分析模块说起。

这个模块负责计算 K 线的各种技术指标：MA、MACD、RSI、布林带……大概 9 种常用指标，服务盘中简报、收盘分析、策略回测三个场景。之前是用手写循环实现的，每算一个指标就要重新造一遍轮子。

某天我看到 MyTT 这个库——一套把通达信公式翻译成 numpy/pandas 向量化实现的工具集。代码简洁得让人心动：

# MyTT 风格的 MACD
def MACD(close):
    DIF = EMA(close, 12) - EMA(close, 26)
    DEA = EMA(DIF, 9)
    MACD = (DIF - DEA) * 2
    return DIF, DEA, MACD

对比原来的实现——for 循环里套 if-else，每算一个窗口就手动算一次均值——差距简直不忍直视。

我毫不犹豫地开启了重构。

重构前后对比：从 400 行到 200 行

先看数据。原模块的手写循环版本大概 400 行，定义了 6 个指标函数。MyTT 重写后降到 200 行，新增了 8 个指标（SAR / DMI / ATR / KDJ / CCI / WR / OBV / MFI），代码量却减少了一半。

核心工具函数的对比最能说明问题：

均线 MA

# 手写循环版
def ma_loop(S, n):
    result = []
    for i in range(len(S)):
        if i < n - 1:
            result.append(S[i])
        else:
            result.append(sum(S[i-n+1:i+1]) / n)
    return result

# MyTT 向量化版
def MA(S, n):
    return pd.Series(S).rolling(n).mean().values

4 行 vs 1 行。

MACD

# 手写循环版
def macd_loop(S):
    ema12 = ema_loop(S, 12)    # 自行实现 EMA 循环
    ema26 = ema_loop(S, 26)
    dif = [ema12[i] - ema26[i] for i in range(min(len(ema12), len(ema26)))]
    dea = ema_loop(dif, 9)
    macd = [2 * (dif[i] - dea[i]) for i in range(min(len(dif), len(dea)))]
    return dif, dea, macd

# MyTT 向量化版
def EMA(S, n):
    return pd.Series(S).ewm(span=n, adjust=False).mean().values

def calc_macd(closes):
    dif = EMA(closes, 12) - EMA(closes, 26)
    dea = EMA(dif, 9)
    macd = (dif - dea) * 2
    ...

EMA 从 8 行手写指数加权变成 1 行 ewm 调用。MACD 从 8 行变成 3 行。

MyTT 的”元工具”设计

MyTT 最巧妙的地方是定义了一套 0 级工具函数（约 15 个），所有指标都通过它们组合实现：

# 核心工具函数
def MA(S, N):        return pd.Series(S).rolling(N).mean().values
def EMA(S, N):       return pd.Series(S).ewm(span=N, adjust=False).mean().values
def HHV(S, N):       return pd.Series(S).rolling(N).max().values   # N日最高
def LLV(S, N):       return pd.Series(S).rolling(N).min().values   # N日最低
def CROSS(S1, S2):   return np.concatenate(([False], (S1[:-1] <= S2[:-1]) & (S1[1:] > S2[1:])))
def SUM(S, N):       return pd.Series(S).rolling(N).sum().values
def REF(S, N):       return pd.Series(S).shift(N).values
def STD(S, N):       return pd.Series(S).rolling(N).std(ddof=0).values

有了这些积木，新指标就像拼乐高一样简单。比如 KDJ：

def calc_kdj(closes, highs, lows, N=9, M1=3, M2=3):
    hhv = HHV(highs, N)
    llv = LLV(lows, N)
    rsv = (closes - llv) / (hhv - llv + 1e-10) * 100
    k = EMA(rsv, M1 * 2 - 1)
    d = EMA(k, M2 * 2 - 1)
    j = k * 3 - d * 2
    ...

看一眼就知道它在算什么——和通达信公式几乎一字不差。

性能测试：没想到的结果

重构完代码后，我习惯性地写了一个 benchmark，预期”向量化怎么也得快 10 倍吧”。

结果让我意外。

这是用 120 条 K 线（盘中简报的典型数据量）测的数据：

指标	手写循环	向量化 (pandas)	对比
MA5	33µs	88µs	向量化慢 2.6x
MACD	51µs	408µs	向量化慢 8x
RSI(14)	639µs	3,215µs	向量化慢 5x
布林带	868µs	770µs	接近
全指标(9种)	1.68ms	3.13ms	向量化慢 1.9x