← 返回博客列表

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

2026年09月09日 08:47 · 智兔数服 · 量化系统从零构建

摘要:【量化系统从零构建 #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. 坑与注意事项

  1. KDJ 初值用 50:首根 K/D 缺前值,本篇用 50 起平滑;口径不同结果略有差异,固定即可。
  2. BOLL 标准差用总体标准差/N 而非 /(N-1),与多数行情软件一致;要样本标准差改 /(N-1)
  3. indicator 就地改 rows:返回同一对象,调用多次可叠加多列(ma + kdj + boll)。
  4. 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与布林带指标数据。

想亲自试一下?免费获取证书