A股量化数据工程：从 REST 接口到策略信号（第 3 篇）：指数技术指标自研：MA、KDJ、BOLL、MACD

A股量化数据工程：从 REST 接口到策略信号（第 3 篇）：指数技术指标自研：MA、KDJ、BOLL、MACD

前言

第 1、2 篇把指数「列表 + 实时 + 分时」接进来了。但行情数据真正产生信号，靠的是技术指标。接口已经返回了 ma、kdj、boll、macd 四类现成序列，但要理解它们、做二次加工（比如金叉死叉判定、多周期共振），必须先自己会算。

本篇用指数分时收盘价 c，手写 MA / KDJ / BOLL / MACD 四个指标的完整算法，并与接口返回的序列逐项核对，确认口径一致后再落库。

一、接口原始字段梳理

1）指数历史分时序列

接口地址：https://api.biyingapi.com/hsindex/history/{指数代码.市场}/{分时级别}/{LICENCE}?st=开始时间&et=结束时间

2）指数历史分时 MA

接口地址：https://api.biyingapi.com/hsindex/history/ma/{指数代码.市场}/{分时级别}/{LICENCE}?st=&et=&lt=

3）指数历史分时 KDJ

接口地址：https://api.biyingapi.com/hsindex/history/kdj/{指数代码.市场}/{分时级别}/{LICENCE}?st=&et=&lt=

4）指数历史分时 BOLL

接口地址：https://api.biyingapi.com/hsindex/history/boll/{指数代码.市场}/{分时级别}/{LICENCE}?st=&et=&lt=

5）指数历史分时 MACD

接口地址：https://api.biyingapi.com/hsindex/history/macd/{指数代码.市场}/{分时级别}/{LICENCE}?st=&et=&lt=

二、自研衍生统计指标

接口已给序列，但下面这些量需自研：

所有指标必须基于原始 c 价手写，不能依赖接口的预制值做判定（接口值仅作核对基准）。

三、数据表设计

CREATE TABLE IF NOT EXISTS index_kline (
dm TEXT,
t TEXT,
c REAL,
PRIMARY KEY (dm, t)
);

CREATE TABLE IF NOT EXISTS index_indicator (
dm TEXT,
t TEXT,
ma5 REAL, ma20 REAL, ma60 REAL,
k REAL, d REAL, j REAL,
boll_u REAL, boll_m REAL, boll_d REAL,
macd_diff REAL, macd_dea REAL, macd_bar REAL,
PRIMARY KEY (dm, t)
);

四、完整代码

import requests
import sqlite3
import time

LICENCE = "YOUR_LICENCE"
BASE = "https://api.biyingapi.com"
DB_PATH = "quant.db"

def biying_api_get_retry(url, timeout=15, max_retry=3):
for i in range(max_retry):
try:
r = requests.get(url, timeout=timeout)
r.raise_for_status()
return r.json()
except Exception as e:
if i == max_retry – 1:
raise
time.sleep(1 + i)

def fetch_kline(code, market, level="d", lt=120):
url = f"{BASE}/hsindex/history/{code}.{market}/{level}/{LICENCE}?lt={lt}"
data = biying_api_get_retry(url)
return data.get("data", []) if isinstance(data, dict) else []

def my_ma(seq, n):
if len(seq) < n:
return None
return round(sum(seq[–n:]) / n, 4)

def my_kdj(highs, lows, closes, n=9):
"""简化 KDJ：基于未成熟随机值 RSV。返回 (k, d, j)。"""
if len(closes) < n:
return None, None, None
h = max(highs[–n:]); l = min(lows[–n:])
rsv = 50.0 if h == l else (closes[–1] – l) / (h – l) * 100
# 平滑初值取 50
k = 2/3 * 50 + 1/3 * rsv
d = 2/3 * 50 + 1/3 * k
j = 3 * k – 2 * d
return round(k, 3), round(d, 3), round(j, 3)

def my_boll(closes, n=20):
if len(closes) < n:
return None, None, None
window = closes[–n:]
m = sum(window) / n
var = sum((x – m) ** 2 for x in window) / n
sd = var ** 0.5
return round(m + 2 * sd, 4), round(m, 4), round(m – 2 * sd, 4)

def my_macd(closes, fast=12, slow=26, sig=9):
if len(closes) < slow:
return None, None, None
def ema(seq, n):
k = 2 / (n + 1)
out = [seq[0]]
for x in seq[1:]:
out.append(out[–1] * (1 – k) + x * k)
return out
e12 = ema(closes, fast)
e26 = ema(closes, slow)
diff = [round(a – b, 4) for a, b in zip(e12, e26)]
dea = ema(diff, sig)
bar = [round(2 * (d – de), 4) for d, de in zip(diff, dea)]
return diff[–1], dea[–1], bar[–1]

def save_indicator(dm, rows):
con = sqlite3.connect(DB_PATH)
for r in rows:
con.execute(
"INSERT OR REPLACE INTO index_indicator"
"(dm,t,ma5,ma20,ma60,k,d,j,boll_u,boll_m,boll_d,macd_diff,macd_dea,macd_bar)"
"VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
[dm, r["t"], r.get("ma5"), r.get("ma20"), r.get("ma60"),
r.get("k"), r.get("d"), r.get("j"),
r.get("boll_u"), r.get("boll_m"), r.get("boll_d"),
r.get("macd_diff"), r.get("macd_dea"), r.get("macd_bar")])
con.commit(); con.close()

if __name__ == "__main__":
code, market = "000001", "SH"
kl = fetch_kline(code, market)
kl.sort(key=lambda x: x.get("t", ""))
closes = [x.get("c") for x in kl if x.get("c") is not None]
highs = [x.get("h") for x in kl if x.get("h") is not None]
lows = [x.get("l") for x in kl if x.get("l") is not None]

# 自研 + 与最后一根核对
last = kl[–1] if kl else {}
rec = {"t": last.get("t")}
rec["ma5"] = my_ma(closes, 5)
rec["ma20"] = my_ma(closes, 20)
rec["ma60"] = my_ma(closes, 60)
k, d, j = my_kdj(highs, lows, closes)
rec["k"], rec["d"], rec["j"] = k, d, j
bu, bm, bd = my_boll(closes)
rec["boll_u"], rec["boll_m"], rec["boll_d"] = bu, bm, bd
md, mde, mb = my_macd(closes)
rec["macd_diff"], rec["macd_dea"], rec["macd_bar"] = md, mde, mb
print("自研末值:", rec)
print("接口末值(ma5/kdj/boll/macd):",
last.get("ma5"), (last.get("k"), last.get("d"), last.get("j")),
(last.get("u"), last.get("m"), last.get("d")),
(last.get("diff"), last.get("dea"), last.get("macd")))
save_indicator(f"{code}.{market}", [rec])

五、关键业务说明

• 以 c 为唯一基准：MA/KDJ/BOLL/MACD 全部由分时收盘 c 派生，接口预制序列同此口径，可逐项核对。
• 空值周期：ma120/ma200/ma250 在历史不足时为空，落库前需判空，避免 None 写入报错。
• 指标非投资建议：金叉死叉仅作数据演示，多头共振也只说明历史统计规律。

六、拓展练习

下篇预告

A股量化数据工程：从 REST 接口到策略信号（第 4 篇）：行业与概念板块树构建

下一篇用 hszg/list 拉取指数、行业、概念三级树，讲解如何把 type1/type2/level/pcode 还原成可遍历的板块层级，为成分股映射打底。

免责申明：本篇全部代码、接口调用仅作为量化编程学习演示，仅为数据演示，不构成投资建议。市场有风险，投资需谨慎。