【量化系统从零构建 #11】技术指标(下):KDJ · BOLL · 量价衍生 · 统一接口
摘要:【量化系统从零构建 #11】技术指标(下):KDJ · BOLL · 量价衍生 · 统一接口 系列:《量化系统从零构建》|连载项目 · 纯 GET 取数 · 仅依赖 re
系列:《量化系统从零构建》|连载项目 · 纯 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。数据来自 智兔数服(www.zhituapi.com)。
3. 指标一览
| name | 附加列 | 说明 |
|---|---|---|
ma |
ma{n} |
收盘价均线 |
kdj |
k / d / j |
随机指标 |
boll |
boll_mid / boll_up / boll_lo |
布林带 |
vol_ma |
vol_ma{n} |
成交量均线 |
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. 坑与注意事项
- KDJ 初值用 50:首根 K/D 缺前值,本篇用 50 起平滑;口径不同结果略有差异,固定即可。
- BOLL 标准差用总体标准差:
/N而非/(N-1),与多数行情软件一致;要样本标准差改/(N-1)。 indicator就地改 rows:返回同一对象,调用多次可叠加多列(ma + kdj + boll)。- None 要跳过:指标前段为 None,画图/信号先
if v is None: continue。
7. 小结与下篇预告
本篇把指标层补全(KDJ / BOLL / 量价)并封装统一入口 indicator,下游择时与选股直接调名字即可。指标层完工。
下一篇计划写 #12《选股引擎(上):因子库》:基于落库的基本面(#06)与指标(#10/#11),搭估值 / 盈利 / 成长 / 质量 / 资金流 / 动量六类因子,进入策略层。
8. 免责声明
本文仅演示技术指标的算法实现,所有代码示例均以合成数据校验,未含任何真实数据;文中示例数据仅作演示用途,不构成投资建议,亦不承诺收益。
免费领取证书
数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。
领取路径:进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」(邮箱验证 3 步即可拿到 token)。
把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印KDJ与布林带指标数据。