如何编写指标代码

小象实战讲义 · 进阶量化交易实战：迭代式的量化策略研发 本讲义仅用于编程与量化方法教学，不构成任何投资建议。

在量化策略研发中，指标计算是特征工程与择时信号生成的基础。将数学公式转化为高效、无未来函数的向量化代码是量化工程师的基本功。本节深入剖析均线、动量及震荡类指标的高性能编写方法。

💡 核心导读

• 指标计算的向量化与事件驱动实现：
  • 向量化计算（Pandas/Numpy）：充分利用 SIMD 指令集进行批处理，提升历史回测效率；
  • 递推式计算（EMA/SMA 状态机）：在线上实时推送 Tick 时保持 $O(1)$ 时间复杂度；
• 无未来函数严格约束：
  • 指标必须仅基于历史与当前 Bar（或前一 Bar 收盘价 close.shift(1)）计算，严禁引入未来数据。

┌─────────────────────────────────────────────────────────────┐
│                    技术指标计算架构与数据流拓扑             │
└──────────────────────────────┬──────────────────────────────┘
                               │
       ┌───────────────────────┴───────────────────────┐
       ▼                                               ▼
┌─────────────────────────────┐                 ┌─────────────────────────────┐
│ 批量向量化计算 (Vectorized) │                 │状态机递推计算 (Online State)│
├─────────────────────────────┤                 ├─────────────────────────────┤
│ • rolling(N).mean() / std() │                 │ • prev_ema * alpha + price  │
│ • 适用于离线历史全量回测    │                 │ • 适用于实盘毫秒级事件驱动  │
└─────────────────────────────┘                 └─────────────────────────────┘

一、经典技术指标的数学原理与向量化实现

均线指标（SMA/EMA/WMA）反映不同时间尺度的价格重心。简单移动平均线（SMA）对所有历史数据等权重加权，其公式为：

$$SMA_t(N) = \frac{1}{N} \sum_{i=0}^{N-1} P_{t-i}$$

指数移动平均线（EMA）则赋予近期价格更大的权重，衰减系数为 $\alpha = \frac{2}{N+1}$：

$$EMA_t = \alpha P_t + (1 - \alpha) EMA_{t-1}$$

在编写指标时，我们必须注意滚动窗口的边界效应与对齐方式。

import pandas as pd
import numpy as np

def calculate_technical_indicators(df: pd.DataFrame) -> pd.DataFrame:
    """
    计算经典均线与动量指标（向量化无未来函数实现，合成数据演示，历史模式不代表未来）
    """
    res = df.copy()
    # 1. 简单移动平均线
    res['SMA_10'] = res['close'].rolling(window=10).mean()
    res['SMA_30'] = res['close'].rolling(window=30).mean()
    
    # 2. 指数移动平均线
    res['EMA_12'] = res['close'].ewm(span=12, adjust=False).mean()
    res['EMA_26'] = res['close'].ewm(span=26, adjust=False).mean()
    
    # 3. MACD 差离值与信号线
    res['DIF'] = res['EMA_12'] - res['EMA_26']
    res['DEA'] = res['DIF'].ewm(span=9, adjust=False).mean()
    res['MACD_HIST'] = (res['DIF'] - res['DEA']) * 2.0
    
    return res

# 合成价格序列（固定种子保证可复现，不构成收益承诺）
rng = np.random.default_rng(42)
prices = 100.0 + np.cumsum(rng.normal(0.05, 1.0, 50))
df_demo = pd.DataFrame({'close': prices}, index=pd.date_range('2024-01-01', periods=50, freq='D'))

# 计算指标并展示结果
df_with_indicators = calculate_technical_indicators(df_demo)
print("===== 技术指标计算演示（向量化实现） =====")
print(f"数据形状: {df_with_indicators.shape}")
print("\n最后 5 行数据（含指标）:")
print(df_with_indicators[['close', 'SMA_10', 'EMA_12', 'DIF', 'MACD_HIST']].tail())
print("\n指标说明: SMA_10 (10日简单均线), EMA_12 (12日指数均线), DIF (MACD快线), MACD_HIST (MACD柱)")

===== 技术指标计算演示（向量化实现） =====
数据形状: (50, 8)

最后 5 行数据（含指标）:
                 close      SMA_10      EMA_12       DIF  MACD_HIST
2024-02-15  105.019035  104.225514  104.086495  1.059774   0.091423
2024-02-16  105.940463  104.380096  104.371721  1.129167   0.184168
2024-02-17  106.214059  104.641053  104.655157  1.192492   0.248654
2024-02-18  106.942972  105.052350  105.007129  1.286663   0.349597
2024-02-19  107.060552  105.405345  105.323040  1.355161   0.389273

指标说明: SMA_10 (10日简单均线), EMA_12 (12日指数均线), DIF (MACD快线), MACD_HIST (MACD柱)

二、实盘递推状态机实现与性能优化

在实盘 Tick 驱动场景下，重复计算全量历史序列会造成严重的 CPU 算力浪费和延迟瓶颈。通过维护内部状态（如累计和、前值 EMA），可以实现 $O(1)$ 常数时间复杂度的极速更新。

$$\Delta EMA_t = \alpha \times (P_t - EMA_{t-1})$$

class StreamingEMA:
    """
    实盘高频事件驱动 EMA 递推计算引擎（状态机实现，避免未来函数）
    """
    def __init__(self, period: int):
        self.period = period
        self.alpha = 2.0 / (period + 1.0)
        self.current_ema = None
        
    def update(self, price: float) -> float:
        if self.current_ema is None:
            self.current_ema = price
        else:
            self.current_ema = self.alpha * price + (1.0 - self.alpha) * self.current_ema
        return self.current_ema

# 演示 StreamingEMA 的逐点更新过程
price_stream = [100.0, 101.5, 99.8, 102.2, 100.5]
ema_calculator = StreamingEMA(period=3)
print("===== 流式 EMA 计算演示（逐点递推） =====")
print(f"周期 (period): {ema_calculator.period}")
print(f"平滑系数 (alpha): {ema_calculator.alpha:.4f}")
print("\n价格流与 EMA 更新过程:")
for i, price in enumerate(price_stream):
    ema_val = ema_calculator.update(price)
    print(f"  第{i+1}点: 价格={price:6.2f}, EMA={ema_val:6.4f}")
print(f"\n最终 EMA 值: {ema_val:.4f}")

===== 流式 EMA 计算演示（逐点递推） =====
周期 (period): 3
平滑系数 (alpha): 0.5000

价格流与 EMA 更新过程:
  第1点: 价格=100.00, EMA=100.0000
  第2点: 价格=101.50, EMA=100.7500
  第3点: 价格= 99.80, EMA=100.2750
  第4点: 价格=102.20, EMA=101.2375
  第5点: 价格=100.50, EMA=100.8688

最终 EMA 值: 100.8688

📝 动手练一练

下面编写完整的指标计算与单元测试代码，验证批量向量化与在线递推计算的一致性：

import pandas as pd
import numpy as np

class StreamingEMA:
    """
    实盘高频事件驱动 EMA 递推计算引擎（状态机实现，避免未来函数）
    """
    def __init__(self, period: int):
        self.period = period
        self.alpha = 2.0 / (period + 1.0)
        self.current_ema = None
        
    def update(self, price: float) -> float:
        if self.current_ema is None:
            self.current_ema = price
        else:
            self.current_ema = self.alpha * price + (1.0 - self.alpha) * self.current_ema
        return self.current_ema

def run_indicator_validation():
    rng = np.random.default_rng(42)
    prices = 100.0 + np.cumsum(rng.normal(0.05, 1.0, 100))
    df = pd.DataFrame({'close': prices})
    
    # 1. 批量向量化计算
    df['EMA_10'] = df['close'].ewm(span=10, adjust=False).mean()
    
    # 2. 递推状态机计算
    streaming_engine = StreamingEMA(period=10)
    streaming_values = [streaming_engine.update(p) for p in prices]
    
    df['EMA_STREAM'] = streaming_values
    max_diff = np.max(np.abs(df['EMA_10'] - df['EMA_STREAM']))
    
    print("================ 指标计算一致性测试 ================")
    print(f"数据量: {len(df)} 条")
    print(f"批量计算与递推计算最大绝对误差: {max_diff:.8e}")
    assert max_diff < 1e-6, "指标计算一致性校验失败！"
    print("✅ 指标编写精度验证 PASS")
    print("==================================================")

run_indicator_validation()

运行结果输出：

================ 指标计算一致性测试 ================
数据量: 100 条
批量计算与递推计算最大绝对误差: 0.00000000e+00
✅ 指标编写精度验证 PASS
==================================================

本章小结

1. 熟练掌握向量化与递推式计算：离线回测使用 Pandas 向量化提升速度，实盘交易采用状态机递推降低延迟。
2. 严防未来函数：所有移动窗口必须仅包含当前及历史 Bar，涉及交叉信号时必须明确对齐生效周期。
3. 指标精度对齐：在策略上线前必须严格校验 Python 实现与行情软件/商业平台在初始化权重上的微小差异。

📋 行动清单 (Action Checklist)

• 运行本讲单元测试代码，验证 EMA 在不同初始值条件下的收敛步数；
• 编写 RSI（相对强弱指标）与布林带的向量化无未来函数实现；
• 预习下一节，深入探究指标背后的物理意义与市场动能映射。

—— 小象教研组