Appearance
用Python计算技术指标
技术指标是通过对价格、成交量等市场数据进行数学计算得到的衍生数据，用于辅助判断市场趋势、动量和超买超卖状态。用 Python 自动计算技术指标，是量化分析的基础能力。
本文将介绍两类方法：一是使用专业库 TA-Lib 快速计算，二是自己手写计算逻辑以深入理解指标原理。
环境准备
安装基础库
bash
pip install pandas numpy akshare安装 TA-Lib
TA-Lib（Technical Analysis Library）是最经典的技术分析计算库，用 C 语言编写，支持 150 多种技术指标。Python 通过封装层调用它。
Windows 安装（推荐用 whl 文件）：
bash
# 先到 https://github.com/cgohlke/talib-build/releases 下载对应版本的 .whl 文件
# 然后离线安装
pip install TA_Lib-0.4.28-cp312-cp312-win_amd64.whlmacOS 安装：
bash
brew install ta-lib
pip install ta-libLinux 安装：
bash
wget http://prdownloads.sourceforge.net/ta-lib/ta-lib-0.4.0-src.tar.gz
tar -xzf ta-lib-0.4.0-src.tar.gz
cd ta-lib/
./configure
make
make install
pip install ta-lib安装完成后验证：
python
import talib
print(talib.__version__) # 应输出类似 "0.4.28"如果安装遇到问题，可以使用纯 Python 实现的替代库：
pip install pandas-ta，它不需要编译 C 代码，但计算速度比 TA-Lib 慢。
获取测试数据
我们先用 akshare 获取一组测试数据，后续所有指标都基于这份数据计算：
python
import akshare as ak
import pandas as pd
import numpy as np
# 获取平安银行（000001）近250个交易日的日K数据
df = ak.stock_zh_a_hist(
symbol="000001",
period="daily",
start_date="20240101",
end_date="20241231",
adjust="qfq"
)
df = df.rename(columns={
"日期": "Date",
"开盘": "Open",
"收盘": "Close",
"最高": "High",
"最低": "Low",
"成交量": "Volume"
})
df["Date"] = pd.to_datetime(df["Date"])
df = df.set_index("Date")
df = df[["Open", "High", "Low", "Close", "Volume"]]
print(f"数据范围: {df.index[0].date()} ~ {df.index[-1].date()}")
print(f"共 {len(df)} 条记录")
print(df.tail())MACD 指标
MACD（Moving Average Convergence Divergence，指数平滑异同移动平均线） 是最常用的趋势跟踪指标，由 Gerald Appel 于 1970 年代发明。
MACD 包含三个组成部分：
| 组成部分 | 说明 |
|---|---|
| DIF（快线） | 12日EMA 减去 26日EMA，反映短期与长期趋势的差异 |
| DEA（慢线） | DIF 的 9日 EMA，是 DIF 的平滑版本 |
| MACD柱 | (DIF - DEA) × 2，反映多空力量的强弱变化 |
EMA（Exponential Moving Average，指数移动平均）：一种加权移动平均，近期数据权重更大，对价格变化更敏感。相比简单移动平均（SMA），EMA 反应更快。
用 TA-Lib 计算
python
import talib
close = df["Close"].values # TA-Lib 接受 numpy 数组
# 计算 MACD
# 参数：fastperiod=12, slowperiod=26, signalperiod=9
dif, dea, macd_bar = talib.MACD(
close,
fastperiod=12,
slowperiod=26,
signalperiod=9
)
# 将结果添加到 DataFrame
df["DIF"] = dif
df["DEA"] = dea
df["MACD"] = macd_bar * 2 # talib 返回的是 (DIF-DEA)，乘2才是国内习惯
print(df[["Close", "DIF", "DEA", "MACD"]].tail(10))手动计算 MACD
理解手动计算过程，有助于深入理解指标的含义：
python
# 第一步：计算 EMA12 和 EMA26
ema12 = df["Close"].ewm(span=12, adjust=False).mean()
ema26 = df["Close"].ewm(span=26, adjust=False).mean()
# 第二步：DIF = EMA12 - EMA26
dif_manual = ema12 - ema26
# 第三步：DEA = DIF 的 9日 EMA
dea_manual = dif_manual.ewm(span=9, adjust=False).mean()
# 第四步：MACD 柱 = (DIF - DEA) × 2
macd_manual = (dif_manual - dea_manual) * 2
# 验证与 talib 结果一致
print("DIF 差异:", abs(dif_manual.dropna() - pd.Series(dif).dropna()).max())
print("MACD 差异:", abs(macd_manual.dropna() - pd.Series(macd_bar * 2).dropna()).max())
ewm(span=N)是 pandas 提供的指数加权计算方法，span参数对应 EMA 的周期。adjust=False表示使用递归公式计算，与常用行情软件的结果一致。
MACD 的常见用法
- 金叉买入：DIF 从下方穿越 DEA（MACD柱由负转正）
- 死叉卖出：DIF 从上方穿越 DEA（MACD柱由正转负）
- 顶背离：股价创新高但 MACD 没有创新高，可能预示趋势反转
- 底背离：股价创新低但 MACD 没有创新低，可能预示反弹
RSI 指标
RSI（Relative Strength Index，相对强弱指标） 由 Welles Wilder 于 1978 年提出，用于衡量价格变动的速度和幅度，判断市场的超买超卖状态。
RSI 的取值范围为 0~100，一般使用 14 日作为默认周期：
- RSI > 70：超买区，价格可能过高，有回调风险
- RSI < 30：超卖区，价格可能过低，有反弹机会
- RSI 在 40~60 之间：中性区域，趋势不明朗
用 TA-Lib 计算
python
# 计算 RSI，默认周期14
rsi = talib.RSI(close, timeperiod=14)
df["RSI14"] = rsi
# 也可以计算不同周期
df["RSI6"] = talib.RSI(close, timeperiod=6) # 短期RSI，更灵敏
df["RSI24"] = talib.RSI(close, timeperiod=24) # 长期RSI，更平滑
print(df[["Close", "RSI14"]].tail(10))手动计算 RSI
手写 RSI 是理解其计算原理的最好方式。RSI 的核心公式为：
$$RSI = 100 - \frac{100}{1 + RS}$$
其中 $RS = \frac{平均涨幅}{平均跌幅}$
python
def calc_rsi(series, period=14):
"""
手动计算 RSI 指标
参数:
series: pd.Series，价格序列（通常是收盘价）
period: int，计算周期，默认14
返回:
pd.Series，RSI 值序列
"""
# 第一步：计算每日价格变动
delta = series.diff(1) # 今天收盘价 - 昨天收盘价
# 第二步：分离涨跌
gain = delta.where(delta > 0, 0.0) # 上涨日：取正值；下跌日：取0
loss = (-delta).where(delta < 0, 0.0) # 下跌日：取绝对值；上涨日：取0
# 第三步：计算平均涨跌幅
# 第一个平均用简单平均，之后用指数平滑（Wilder 平滑法）
avg_gain = gain.rolling(window=period, min_periods=period).mean()
avg_loss = loss.rolling(window=period, min_periods=period).mean()
# Wilder 平滑：从第二个值开始，用递推公式
for i in range(period, len(gain)):
if pd.notna(avg_gain.iloc[i]) and avg_gain.iloc[i] != 0:
avg_gain.iloc[i] = (
avg_gain.iloc[i - 1] * (period - 1) + gain.iloc[i]
) / period
avg_loss.iloc[i] = (
avg_loss.iloc[i - 1] * (period - 1) + loss.iloc[i]
) / period
# 第四步：计算 RS 和 RSI
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
return rsi
# 使用手写函数
df["RSI14_manual"] = calc_rsi(df["Close"], period=14)
# 对比与 talib 结果的差异
diff = abs(df["RSI14"].dropna() - df["RSI14_manual"].dropna())
print(f"与 talib 结果最大差异: {diff.max():.6f}")Wilder 平滑法：与普通 EMA 不同，Wilder 使用的平滑系数是
1/period（而不是2/(period+1)）。这是 RSI、ATR 等指标的标准计算方法。
KDJ 指标
KDJ（随机指标） 由 George Lane 提出，通过比较收盘价在一定周期内价格区间中的位置来判断超买超卖。它是 A 股投资者最常用的短线指标之一。
KDJ 由三条线组成：
- K 值（快线）：当日 RSV 的指数平滑
- D 值（慢线）：K 值的指数平滑
- J 值（方向线）：3K - 2D，对 K、D 的放大
RSV（Raw Stochastic Value，未成熟随机值）：
(收盘价 - N日最低价) / (N日最高价 - N日最低价) × 100，反映当日收盘价在近 N 日价格区间中的相对位置。
用 TA-Lib 计算
python
# TA-Lib 中 KDJ 对应的函数是 STOCH（随机指标）
# fastk_period=9, slowk_period=3, slowd_period=3 是国内常用参数
k, d = talib.STOCH(
df["High"].values,
df["Low"].values,
close,
fastk_period=9,
slowk_period=3,
slowk_matype=0, # 0 = SMA
slowd_period=3,
slowd_matype=0
)
df["K"] = k
df["D"] = d
df["J"] = 3 * k - 2 * d # J 值需要手动计算
print(df[["Close", "K", "D", "J"]].tail(10))手动计算 KDJ
python
def calc_kdj(df, n=9, m1=3, m2=3):
"""
手动计算 KDJ 指标
参数:
df: DataFrame，需要包含 High, Low, Close 列
n: int，RSV 计算周期，默认9
m1: int，K 值平滑周期，默认3
m2: int，D 值平滑周期，默认3
返回:
DataFrame，包含 K, D, J 三列
"""
# 计算 N 日内最低价和最高价
low_n = df["Low"].rolling(window=n).min()
high_n = df["High"].rolling(window=n).max()
# 计算 RSV
rsv = (df["Close"] - low_n) / (high_n - low_n) * 100
# 计算 K 值：K = (2/3) × 前日K + (1/3) × 当日RSV
k = pd.Series(np.nan, index=df.index)
k.iloc[n - 1] = 50 # 初始值取50
for i in range(n, len(df)):
k.iloc[i] = (2 / 3) * k.iloc[i - 1] + (1 / 3) * rsv.iloc[i]
# 计算 D 值：D = (2/3) × 前日D + (1/3) × 当日K
d = pd.Series(np.nan, index=df.index)
d.iloc[n - 1] = 50 # 初始值取50
for i in range(n, len(df)):
d.iloc[i] = (2 / 3) * d.iloc[i - 1] + (1 / 3) * k.iloc[i]
# 计算 J 值
j = 3 * k - 2 * d
result = pd.DataFrame({"K": k, "D": d, "J": j}, index=df.index)
return result
kdj_df = calc_kdj(df)
print(kdj_df.tail(10))KDJ 的常见用法
- K 线上穿 D 线（金叉），且 J 值从负值区域回升：买入信号
- K 线下穿 D 线（死叉），且 J 值从正值区域回落：卖出信号
- K > 80 且 D > 80：超买区，短期可能回调
- K < 20 且 D < 20：超卖区，短期可能反弹
布林带
布林带（Bollinger Bands，简称 BOLL） 由 John Bollinger 在 1980 年代提出，由中轨（20日SMA）、上轨（中轨+2倍标准差）、下轨（中轨-2倍标准差）三条线组成。
布林带的核心思想：价格在大多数时候会在上下轨之间波动。当价格触及上轨时可能偏高，触及下轨时可能偏低。
用 TA-Lib 计算
python
# 计算布林带
# 参数：timeperiod=20, nbdevup=2, nbdevdn=2, matype=0(SMA)
upper, middle, lower = talib.BBANDS(
close,
timeperiod=20,
nbdevup=2, # 上轨：中轨 + 2倍标准差
nbdevdn=2, # 下轨：中轨 - 2倍标准差
matype=0 # 0 = 简单移动平均
)
df["BOLL_upper"] = upper
df["BOLL_middle"] = middle
df["BOLL_lower"] = lower
print(df[["Close", "BOLL_upper", "BOLL_middle", "BOLL_lower"]].tail(10))手动计算布林带
python
def calc_boll(series, period=20, num_std=2):
"""
手动计算布林带
参数:
series: pd.Series，价格序列
period: int，中轨周期，默认20
num_std: float，标准差倍数，默认2
返回:
DataFrame，包含 upper, middle, lower 三列
"""
middle = series.rolling(window=period).mean()
std = series.rolling(window=period).std()
upper = middle + num_std * std
lower = middle - num_std * std
return pd.DataFrame({
"upper": upper,
"middle": middle,
"lower": lower
})
boll = calc_boll(df["Close"])
print(boll.tail(10))布林带的常见用法
- 布林带收窄：上下轨间距缩小，说明波动率降低，可能即将出现大幅突破
- 价格突破上轨：短期强势，但要注意是否过热
- 价格跌破下轨：短期弱势，但可能存在超卖反弹机会
- 价格沿上轨/下轨运行：趋势行情中的典型特征
批量计算多只股票指标
在实际量化分析中，我们通常需要对多只股票同时计算技术指标。以下是一个完整的批量计算框架：
python
def batch_calc_indicators(stock_list, start_date, end_date):
"""
批量计算多只股票的技术指标
参数:
stock_list: list，股票代码列表，如 ["000001", "600519", "000858"]
start_date: str，起始日期，如 "20240101"
end_date: str，结束日期，如 "20241231"
返回:
dict，key 为股票代码，value 为包含指标的 DataFrame
"""
result = {}
for symbol in stock_list:
try:
# 获取数据
df = ak.stock_zh_a_hist(
symbol=symbol,
period="daily",
start_date=start_date,
end_date=end_date,
adjust="qfq"
)
df = df.rename(columns={
"日期": "Date", "开盘": "Open", "收盘": "Close",
"最高": "High", "最低": "Low", "成交量": "Volume"
})
df["Date"] = pd.to_datetime(df["Date"])
df = df.set_index("Date")
df = df[["Open", "High", "Low", "Close", "Volume"]]
close = df["Close"].values
# 批量计算指标
df["MA5"] = talib.SMA(close, timeperiod=5)
df["MA20"] = talib.SMA(close, timeperiod=20)
df["MA60"] = talib.SMA(close, timeperiod=60)
dif, dea, macd_bar = talib.MACD(close)
df["DIF"] = dif
df["DEA"] = dea
df["MACD"] = macd_bar * 2
df["RSI14"] = talib.RSI(close, timeperiod=14)
upper, middle, lower = talib.BBANDS(close)
df["BOLL_upper"] = upper
df["BOLL_middle"] = middle
df["BOLL_lower"] = lower
result[symbol] = df
print(f"✓ {symbol} 指标计算完成，共 {len(df)} 条记录")
except Exception as e:
print(f"✗ {symbol} 处理失败: {e}")
return result
# 批量计算5只股票
stocks = ["000001", "600519", "000858", "601318", "000333"]
all_data = batch_calc_indicators(stocks, "20240101", "20241231")
# 查看某只股票的最新指标
if "600519" in all_data:
print(all_data["600519"].tail(5))批量筛选：RSI 超卖股票
python
# 从计算结果中筛选 RSI < 30 的股票（超卖状态）
print("=== RSI 超卖股票 ===")
for symbol, data in all_data.items():
latest = data.iloc[-1]
if pd.notna(latest["RSI14"]) and latest["RSI14"] < 30:
print(
f"{symbol}: 收盘价 {latest['Close']:.2f}, "
f"RSI {latest['RSI14']:.1f}"
)指标计算注意事项
1. 前置 NaN 值
所有基于滚动窗口计算的指标，在初始阶段都会产生 NaN（缺失值）。例如 MA60 需要至少 60 个交易日才能算出第一个值。在策略中要注意过滤掉这些 NaN：
python
# 方法一：用 dropna 移除 NaN 行
df_clean = df.dropna(subset=["MA20", "RSI14"])
# 方法二：用 fillna 填充默认值
df["RSI14"] = df["RSI14"].fillna(50) # RSI 中性值2. 复权对指标的影响
计算技术指标必须使用复权数据（前复权或后复权均可）。不复权的数据在除权除息日会产生价格跳空，导致均线和指标失真。
3. 指标的滞后性
所有基于历史数据计算的指标都具有滞后性。均线、MACD、布林带都是跟随价格的"滞后指标"，它们反映的是已经发生的趋势，而不是预测未来。RSI 和 KDJ 虽然属于"震荡指标"，反应更灵敏，但同样存在滞后和假信号的问题。
小结
本文介绍了四种常用技术指标的 Python 计算方法：
| 指标 | 类型 | 核心参数 | 主要用途 |
|---|---|---|---|
| MACD | 趋势跟踪 | 12, 26, 9 | 判断趋势方向和强弱 |
| RSI | 震荡指标 | 14 | 判断超买超卖 |
| KDJ | 震荡指标 | 9, 3, 3 | 短线买卖时机 |
| 布林带 | 波动率 | 20, 2 | 判断波动区间和突破 |
掌握了技术指标的计算方法后，下一步就是将这些指标组合起来，构建完整的交易策略。在下一篇《策略开发流程》中，我们将系统学习如何从零开始开发一个量化策略。
免责声明：本文所有代码仅用于技术教学目的，文中涉及的股票代码仅为数据示例，不构成任何投资建议。技术指标是辅助分析工具，不能作为投资决策的唯一依据。投资有风险，入市需谨慎。