如何用 Python 计算量化技术指标

背景

在量化交易程序的开发过程中，技术指标的计算是一个绕不开的环节。常见的指标如 MA（移动平均线）、KDJ（随机指标）、RSI（相对强弱指数） 等，几乎是量化策略的“基础工具箱”。

很早之前，我在网上学习过这些指标的原理和实现方法。本文旨在对这些概念和计算方法做一个整理和记录：一方面是工作中的实际需求，另一方面也是作为再次学习时的备忘。

需要特别说明的是，本文 不涉及指标的具体应用场景和使用策略。原因有二：其一，每个人对指标的理解和使用方法可能不同；其二，如果展开讲述，篇幅会非常长。此外，文中可能存在不完善或不严谨的地方，还请多多包涵。

说明

在 Python 中，计算大部分量化指标并不需要从零实现。我们通常会借助 TA-Lib（全称 Technical Analysis Library）这个功能强大的库。它几乎涵盖了所有常用技术指标，如 MA、KDJ、RSI、MACD 等，只需几行代码即可直接得到计算结果，非常适合量化交易的快速开发。

在实际使用中，需要注意的是，有些指标的计算结果可能与行情软件上显示的略有差异，这时可能需要调整相关参数，或者自行进行一些处理以保证一致性。

数据准备

技术指标的计算需要有一份基础行情数据。数据的获取方式因人而异，可以来自交易所、券商 API、第三方数据平台，或者网络上的公开数据。我这里使用的是 螺纹钢（RB） 的日线行情，时间区间为 2023-03-01 至 2025-08-15。

在后续的示例中，我主要会用到 Pandas（用于数据处理）和 TA-Lib（用于指标计算）。这两个库我是用 uv add pandas Ta-Lib 来安装，如果没有使用 uv，可以直接通过 pip install pandas Ta-Lib 来安装试试。

代码

import pandas as pd
import talib

rb_df = pd.read_csv('SHFE.RB_1d.csv', index_col=0, parse_dates=True)
rb_df.index = rb_df.index.strftime('%Y-%m-%d')
rb_df.sort_index(ascending=True, inplace=True)
rb_df.tail()

图示

技术指标

MA

说明

• MA: 简单移动平均线，Moving Average
• EMA: 指数移动平均线，近期的价格分配更高的权重
• 均线滞后于价格
• 均线描述过去的走势
• 均线跟随价格走势
• 用于判断当下的趋势

公式

代码

# MA
ma = talib.MA(rb_df.close, timeperiod=30)
ma.tail()

# EMA
from talib import MA_Type
ema = talib.MA(rb_df.close, timeperiod=30, matype=MA_Type.EMA)
ema.tail()

例子

MACD

说明

• 指数平滑异同移动平均线
• 组成：快线 DIF，慢线 DEA，0轴，能量柱
• 参数：一般为 12， 26， 9

公式

1. 快线DIF $$快线DIF = \text{12EMA} - \text{26EMA}$$
2. 慢线DEA = 快线DIF的9EMA $$慢线DEA = \text{快线DIF的9EMA}$$
3. 能量柱 $$能量柱 = (\text{快线DIF} - \text{慢线DEA}) \times {\text{n}}$$

代码

# MACD
diff, dea, macd = talib.MACD(rb_df.close, fastperiod=12, slowperiod=26, signalperiod=9)
macd_df = pd.DataFrame({
    'diff': diff,
    'dea': dea,
    'macd': macd * 2
}, index=diff.index)

macd_df.tail()

例子

KDJ

说明

• 属于震荡指标，随机震荡指标，KD指标+J
• 组成：K线，D线，J线，20轴，80轴
• 参数：一般为 9，3， 3

公式

1. RSV，首先要计算 RSV未成熟随机值 $$RSV = \frac{\text{当前收盘价} - \text{近9根K线最低价}}{\text{近9根K线最高价} - \text{近9根K线最低价}} \times 100$$ 即： $$RSV = \frac{\text{涨幅}}{\text{振幅}} \times 100$$
2. 计算 K 值 $$k = \text{前一天K值(没有则为50)} \times \frac{2}{3} + \text{当前RSV} \times \frac{1}{3}$$
3. 计算 D 值 $$d = \text{前一天D值(没有则为50)} \times \frac{2}{3} + \text{当前K值} \times \frac{1}{3}$$
4. 计算 J 值 $$J = 3 \times K - 2 \times D$$

代码

k, d = talib.STOCH(rb_df.high, rb_df.low, rb_df.close, fastk_period=9, slowk_period=3, slowd_period=3)
j = 3 * k - 2 * d
kdj_df = pd.DataFrame({
    'K': k,
    'D': d,
    'J': j,
})

kdj_df.tail(10)

例子

RSI

说明

• 相对强弱指标
• 组成：RSI线，70轴，50轴，30轴
• 参数：一般为 14

公式

即：

代码

rsi = talib.RSI(rb_df.close, timeperiod=14)
rsi.tail()

例子

ATR

说明

• 平均真实波动幅度
• 组成：单线
• 参数：一般为 14

公式

1. 计算TR $$TR = \max(\text{最高价} - \text{最低价}, \text{最高价} - \text{前收盘价}, \text{前收盘价} - \text{最低价})$$
2. 计算ATR $$ATR = \text{近期N根K线TR值的求和平均}$$

代码

atr = talib.ATR(rb_df.high, rb_df.low, rb_df.close, timeperiod=14)
atr.tail()

例子

BOLL

说明

• 布林带，布林通道
• 组成：上轨，中轨，下轨
• 参数：一般为 20

公式

1. 中轨 $$中轨 = N日的移动平均线$$
2. 上轨 $$上轨 = 中轨 + 2倍标准差$$
3. 下轨 $$下轨 = 中轨 - 2倍标准差$$

代码

# BOLL
upper, middle, lower = talib.BBANDS(rb_df.close, timeperiod=14)
boll_df = pd.DataFrame({
    'upper': upper,
    'middle': middle,
    'lower': lower,
}, index=upper.index)

boll_df.tail()

例子

OBV

说明

• 能量潮指标，On Balance Volume

公式

代码

# OBV
obv = talib.OBV(rb_df.close, rb_df.volume)
obv.tail()

例子