3. 技术指标计算

任务目标

实现一组量化研究中最常用的技术指标计算函数:简单移动平均线(SMA)、指数移动平均线(EMA)、相对强弱指标(RSI)和异同移动平均线(MACD),并使用公开可验证的输入数据校验计算结果的正确性。

环境与依赖

仅需 Python 3.8+ 标准库,无需第三方包。所有指标均从零实现,不依赖 pandas/numpy,便于核对算法逻辑。

核心实现

from typing import List, Tuple

def sma(values: List[float], period: int) -> List[float]:
    """简单移动平均线。输出长度与输入相同,前 period-1 个位置为 None。"""
    if period <= 0:
        raise ValueError("period must be positive")
    result: List[float] = [None] * len(values)
    window_sum = 0.0
    for i, v in enumerate(values):
        window_sum += v
        if i >= period:
            window_sum -= values[i - period]
        if i >= period - 1:
            result[i] = window_sum / period
    return result

def ema(values: List[float], period: int) -> List[float]:
    """指数移动平均线。以第一个值为种子,平滑因子 alpha = 2/(period+1)。"""
    if period <= 0:
        raise ValueError("period must be positive")
    result: List[float] = [None] * len(values)
    if not values:
        return result
    alpha = 2.0 / (period + 1)
    result[0] = values[0]
    for i in range(1, len(values)):
        result[i] = alpha * values[i] + (1 - alpha) * result[i-1]
    return result

def rsi(values: List[float], period: int = 14) -> List[float]:
    """相对强弱指标(Wilder 平滑法)。输出前 period 个位置为 None。"""
    if period <= 0:
        raise ValueError("period must be positive")
    result: List[float] = [None] * len(values)
    if len(values) <= period:
        return result
    gains, losses = [], []
    for i in range(1, len(values)):
        diff = values[i] - values[i-1]
        gains.append(max(diff, 0.0))
        losses.append(max(-diff, 0.0))
    # Wilder 平滑:初始为简单平均,之后递推
    avg_gain = sum(gains[:period]) / period
    avg_loss = sum(losses[:period]) / period
    if avg_loss == 0:
        result[period] = 100.0
    else:
        rs = avg_gain / avg_loss
        result[period] = 100.0 - 100.0 / (1.0 + rs)
    for i in range(period + 1, len(values)):
        avg_gain = (avg_gain * (period - 1) + gains[i-1]) / period
        avg_loss = (avg_loss * (period - 1) + losses[i-1]) / period
        if avg_loss == 0:
            result[i] = 100.0
        else:
            rs = avg_gain / avg_loss
            result[i] = 100.0 - 100.0 / (1.0 + rs)
    return result

def macd(values: List[float], fast: int = 12, slow: int = 26, signal: int = 9) -> Tuple[List[float], List[float], List[float]]:
    """MACD 线(DIF)、信号线(DEA)、柱状图(MACD Histogram)。"""
    ema_fast = ema(values, fast)
    ema_slow = ema(values, slow)
    dif = [None] * len(values)
    for i in range(len(values)):
        if ema_fast[i] is not None and ema_slow[i] is not None:
            dif[i] = ema_fast[i] - ema_slow[i]
    # 信号线:对 DIF 做 EMA,DIF 前 slow-1 个为 None,从第一个非 None 开始
    first_valid = next(i for i, v in enumerate(dif) if v is not None)
    dif_valid = dif[first_valid:]
    dea_valid = ema(dif_valid, signal)
    dea = [None] * first_valid + dea_valid
    hist = [None] * len(values)
    for i in range(len(values)):
        if dif[i] is not None and dea[i] is not None:
            hist[i] = 2.0 * (dif[i] - dea[i])
    return dif, dea, hist

if __name__ == "__main__":
    # 验证数据:10 个连续整数 1~10
    data = [float(i) for i in range(1, 11)]
    print("SMA(3):", [round(x, 4) if x is not None else None for x in sma(data, 3)])
    print("EMA(3):", [round(x, 4) if x is not None else None for x in ema(data, 3)])
    print("RSI(3):", [round(x, 4) if x is not None else None for x in rsi(data, 3)])
    dif, dea, hist = macd(data)
    print("MACD DIF:", [round(x, 4) if x is not None else None for x in dif])
    print("MACD DEA:", [round(x, 4) if x is not None else None for x in dea])
    print("MACD HIST:", [round(x, 4) if x is not None else None for x in hist])

运行与验证

将上述代码保存为 technical_indicators.py,在终端执行:

python technical_indicators.py

程序会打印各指标的计算结果。由于输入数据为严格递增的 1~10,可手工推导验证:

  • SMA(3):第3个位置为 (1+2+3)/3=2.0,第4个为 (2+3+4)/3=3.0,依此类推,最后一个为 (8+9+10)/3=9.0。
  • EMA(3):alpha=0.5,第1个值为1.0,第2个为 0.5×2+0.5×1=1.5,第3个为 0.5×3+0.5×1.5=2.25,后续按递推公式计算。
  • RSI(3):数据单调递增,所有变化均为正收益,平均损失为0,因此从第4个位置起 RSI=100.0。
  • MACD:DIF 从第25个位置(slow=26 的 EMA 有效)才开始有值;对递增序列,DIF 始终为正且逐渐收敛,DEA 为 DIF 的 EMA,柱状图为两者差值的2倍。

通过条件:程序无异常输出,且打印结果与上述手工推导的前几个数值一致,即说明指标实现正确。

常见问题

  • RSI 初始值:本实现采用 Wilder 平滑法,与多数交易软件(如 TradingView)默认算法一致。若使用简单平均法,前几个值会有差异,属正常现象。
  • MACD 柱状图定义:国内软件常用 2×(DIF−DEA),国际平台(如 Bloomberg)直接用 DIF−DEA。本实现采用国内惯例,如需对齐国际口径,将 hist 计算中的系数 2 去掉即可。
  • 边界处理:所有指标在数据不足时返回 None,避免产生无意义的填充值,便于后续直接过滤使用。