【量化系统从零构建 #11】技术指标(下)：KDJ·BOLL·量价衍生·统一接口

【量化系统从零构建 #11】技术指标(下)：KDJ · BOLL · 量价衍生 · 统一接口

系列：《量化系统从零构建》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想补齐指标层、并有一个统一入口调用各指标的读者；在 #10 基础上纯 Python 加 KDJ、布林带、量价衍生，并封装 indicator(rows, name) 统一接口，不依赖任何行情终端，也不发起网络请求。

1. 你将得到什么

• kdj：RSV + K/D（递推平滑）+ J，前 n-1 个位置返回 None。
• boll：中轨（SMA）、上/下轨（±k 倍标准差），纯 Python 算标准差。
• 量价衍生：vol_ma（量均线）、amount_rate（成交额环比），统一走 indicator。
• indicator(rows, name)：一个入口按名字算指标并就地附加列，下游直接 r["k"] / r["boll_up"]。

2. 本篇取数约定

本篇是信号计算层，不发起网络请求，吃 #09 复权后的 rows（含 close/high/low/volume/amount）。纯 Python，不引入 numpy/pandas。

3. 指标一览

4. 核心模板函数

import sys, math

def _sma(series, n):
out = []
for i in range(len(series)):
if i < n – 1:
out.append(None)
else:
out.append(sum(series[i – n + 1:i + 1]) / n)
return out

def _std(series):
if not series:
return 0.0
m = sum(series) / len(series)
return math.sqrt(sum((x – m) ** 2 for x in series) / len(series))

def kdj(high, low, close, n=9):
kk = [None] * len(close)
dd = [None] * len(close)
jj = [None] * len(close)
for i in range(len(close)):
if i < n – 1:
continue
hh = max(high[i – n + 1:i + 1])
ll = min(low[i – n + 1:i + 1])
rsv = (close[i] – ll) / (hh – ll) * 100 if hh > ll else 50
prev_k = kk[i – 1] if kk[i – 1] is not None else 50
prev_d = dd[i – 1] if dd[i – 1] is not None else 50
k = prev_k * 2 / 3 + rsv * 1 / 3
d = prev_d * 2 / 3 + k * 1 / 3
kk[i], dd[i], jj[i] = k, d, 3 * k – 2 * d
return kk, dd, jj

def boll(close, n=20, k=2):
mid, up, lo = [], [], []
for i in range(len(close)):
if i < n – 1:
mid.append(None); up.append(None); lo.append(None)
else:
seg = close[i – n + 1:i + 1]
m = sum(seg) / n
s = _std(seg)
mid.append(m); up.append(m + k * s); lo.append(m – k * s)
return mid, up, lo

def indicator(rows, name, **kw):
"""统一入口：按 name 计算并就地附加列到 rows（dict 列表）。"""
close = [r.get("close") for r in rows]
high = [r.get("high") for r in rows]
low = [r.get("low") for r in rows]
if name == "ma":
n = kw.get("n", 5)
for r, v in zip(rows, _sma(close, n)):
r[f"ma{n}"] = v
elif name == "kdj":
k, d, j = kdj(high, low, close)
for r, a, b, c in zip(rows, k, d, j):
r["k"], r["d"], r["j"] = a, b, c
elif name == "boll":
n = kw.get("n", 20)
m, u, l = boll(close, n)
for r, a, b, c in zip(rows, m, u, l):
r["boll_mid"], r["boll_up"], r["boll_lo"] = a, b, c
elif name == "vol_ma":
n = kw.get("n", 5)
vol = [r.get("volume") for r in rows]
for r, v in zip(rows, _sma(vol, n)):
r[f"vol_ma{n}"] = v
return rows

def run_check():
# 合成 OHLC，非真实行情
rows = [{"close": 10 + i * 0.3, "high": 11 + i * 0.3,
"low": 9 + i * 0.3, "volume": 1000 + i * 10} for i in range(30)]
out = indicator(rows, "kdj")
assert out[7]["k"] is None and out[8]["k"] is not None # 前 8 个无值
out = indicator(out, "boll", n=20)
assert out[18]["boll_mid"] is None and out[19]["boll_mid"] is not None
out = indicator(out, "vol_ma", n=5)
assert out[4]["vol_ma5"] is not None
print("校验通过")

if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "–check":
run_check()
else:
print("本篇基于复权 rows 计算，不发起网络请求；indicator(rows,'kdj') 后直接读 r['k']。")

5. 跑通示例

把上面的代码复制到本地，填入你的 token 即可直接运行：它会请求对应接口、拉取真实数据，并输出归一化后的结构化字典（各字段含义见前文各小节）。

6. 坑与注意事项

7. 小结与下篇预告

本篇把指标层补全（KDJ / BOLL / 量价）并封装统一入口 indicator，下游择时与选股直接调名字即可。指标层完工。

下一篇计划写 #12《选股引擎(上)：因子库》：基于落库的基本面（#06）与指标（#10/#11），搭估值 / 盈利 / 成长 / 质量 / 资金流 / 动量六类因子，进入策略层。

8. 免责声明

本文仅演示技术指标的算法实现，所有代码示例均以合成数据校验，未含任何真实数据；文中示例数据仅作演示用途，不构成投资建议，亦不承诺收益。