【零依赖量化数据实战 #25】北交所技术指标与基本面
摘要:【零依赖量化数据实战 #25】北交所技术指标与基本面 系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK 适用:想用 Python 把 北交所(BJ) 股票的技术指标(MAC
系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK
适用:想用 Python 把北交所(BJ)股票的技术指标(MACD / MA / KDJ / BOLL)与基本面(股本、十大股东、流通股东、股东数)一次拉齐的量化爱好者;数据由智兔数服提供,不依赖任何行情终端。
1. 你将得到什么
- 10 个官方接口的最小可用封装,分两组:
- 北交所技术指标(4 个):
/bj/history/{macd|ma|kdj|boll}/{code}/{period}/{adj},与 #24 指数技术指标同构,只是路径多了「除权类型」参数{adj}。 - 北交所基本面 + 盘口(6 个):
/bj/fin/{capital|topholder|flowholder|hm}/{code}(股本表 / 十大股东 / 十大流通股东 / 股东数)、/bj/list/index(股票列表)、/bj/stock/real/mmwp/{code}(实时五档盘口)。 - 复用 #24 的
macd_signal/ma_alignment/kdj_zone/boll_band信号函数(直接消费已算好的指标序列)。 - 一套对字段名中英文混用不敏感的股东/股本抽取函数。
2. 端点语义表
GET https://api.zhituapi.com/bj/history/macd/920000.BJ/d/n?token=你的智兔token<=30
-> 北交所股票 日线 MACD(除权类型 n=不复权 / q=前复权 / f=后复权)
GET https://api.zhituapi.com/bj/history/ma/920000.BJ/d/n?token=你的智兔token<=30
GET https://api.zhituapi.com/bj/history/kdj/920000.BJ/d/n?token=你的智兔token<=30
GET https://api.zhituapi.com/bj/history/boll/920000.BJ/d/n?token=你的智兔token<=30
GET https://api.zhituapi.com/bj/fin/capital/920547.BJ?token=你的智兔token&st=20240101&et=20241231
-> 公司股本表
GET https://api.zhituapi.com/bj/fin/topholder/920547.BJ?token=你的智兔token&st=20240101&et=20241231
-> 公司十大股东
GET https://api.zhituapi.com/bj/fin/flowholder/920547.BJ?token=你的智兔token&st=20240101&et=20241231
-> 公司十大流通股东
GET https://api.zhituapi.com/bj/fin/hm/920547.BJ?token=你的智兔token&st=20240101&et=20241231
-> 公司股东数
GET https://api.zhituapi.com/bj/list/index?token=你的智兔token
-> 北交所股票代码+名称列表(参数传入用)
GET https://api.zhituapi.com/bj/stock/real/mmwp/920000.BJ?token=你的智兔token
-> 实时买卖五档盘口
鉴权:token 走查询参数;{code} 必须带 .BJ 后缀(如 920000.BJ);技术指标路径含 {adj} 除权类型(n/q/f);st/et 格式 YYYYMMDD;lt 取最近 N 条。数据来自 智兔数服(www.zhituapi.com)。
3. 字段名不固定?用候选键命中
北交所财务返回的「股东姓名」可能叫 name / 股东 / 股东名称 / holder,「持股比例」可能叫 ratio / 比例 / 持股比例 / hold_ratio。统一候选键命中:
def _hit_key(d, keys):
if not isinstance(d, dict):
return None
for k in keys:
if k in d and d[k] is not None:
return d[k]
low = {str(x).lower(): x for x in d.keys()}
for k in keys:
kl = k.lower()
if kl in low:
return d[low[kl]]
return None
4. 核心模板函数
import sys, requests
BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token" # 占位,换成你申请的真实 token
def _hit_key(d, keys):
if not isinstance(d, dict):
return None
for k in keys:
if k in d and d[k] is not None:
return d[k]
low = {str(x).lower(): x for x in d.keys()}
for k in keys:
kl = k.lower()
if kl in low:
return d[low[kl]]
return None
def _get(path, params=None):
p = dict(params or {})
p["token"] = TOKEN
try:
r = requests.get(f"{BASE}{path}", params=p, timeout=10)
except Exception as e:
return None, f"网络异常:{e}"
if r.status_code != 200:
return None, f"{r.status_code} {r.text.strip()[:140]}"
try:
return r.json(), None
except Exception:
return None, f"非 JSON:{r.text.strip()[:140]}"
# ===== 技术指标信号(与 #24 同构)=====
def macd_signal(rows):
if not rows or len(rows) < 2:
return None
a, b = rows[-2], rows[-1]
da = _hit_key(a, ["dif", "DIF"]); db = _hit_key(a, ["dea", "DEA"])
ea = _hit_key(b, ["dif", "DIF"]); eb = _hit_key(b, ["dea", "DEA"])
if None in (da, db, ea, eb):
return None
if da <= db and ea > eb: return "golden"
if da >= db and ea < eb: return "dead"
return "hold"
def ma_alignment(row):
mas = {}
for k, v in row.items():
if v is None: continue
kl = str(k).lower()
for pp in (5, 10, 20, 30, 60, 120, 250):
if kl in (f"ma{pp}", f"ma{pp}_close", f"ma{pp}close"):
try: mas[pp] = float(v)
except Exception: pass
if len(mas) < 2: return "mix"
vals = [mas[x] for x in sorted(mas)]
if all(vals[i] <= vals[i+1] for i in range(len(vals)-1)): return "bull"
if all(vals[i] >= vals[i+1] for i in range(len(vals)-1)) return "bear"
return "mix"
def kdj_zone(row):
j = _hit_key(row, ["j", "J"]); k = _hit_key(row, ["k", "K"])
if j is None and k is None: return None
if j is not None:
j = float(j)
if j > 100: return "overbought"
if j < 0: return "oversold"
if k is not None:
k = float(k)
if k > 80: return "overbought"
if k < 20: return "oversold"
return "neutral"
def boll_band(row):
up = _hit_key(row, ["up", "upper", "boll_up", "BOLL_UP"])
mid = _hit_key(row, ["mid", "middle", "BOLL_MID", "mb"])
low = _hit_key(row, ["low", "lower", "boll_low", "BOLL_LOW"])
if None in (up, mid, low): return None
return float(up), float(mid), float(low)
def fetch_bj_indicator(kind, code="920000.BJ", period="d", adj="n", lt=30):
data, err = _get(f"/bj/history/{kind}/{code}/{period}/{adj}", {"lt": lt})
if err:
return None, err
rows = data
if isinstance(data, dict):
for k in ("data", "list", "rows", "items"):
if isinstance(data.get(k), list):
rows = data[k]; break
else:
rows = [data]
if not isinstance(rows, list):
rows = [rows]
return rows, None
# ===== 基本面 + 盘口 =====
def fetch_bj_fin(kind, code="920547.BJ", st="20240101", et="20241231"):
data, err = _get(f"/bj/fin/{kind}/{code}", {"st": st, "et": et})
if err:
return None, err
if isinstance(data, dict):
for k in ("data", "list", "rows", "items"):
if isinstance(data.get(k), list):
return data[k], None
return [data], None
if isinstance(data, list):
return data, None
return None, f"形态异常:{type(data)}"
def fetch_bj_list_index():
return _get("/bj/list/index")
def fetch_bj_mmwp(code="920000.BJ"):
return _get(f"/bj/stock/real/mmwp/{code}")
def _to_float(v):
try:
return None if v is None else float(v)
except (TypeError, ValueError):
return None
def top_holder(rows):
"""十大股东:候选键抽 姓名+持股比例,按持股比例倒序取首条。"""
if not isinstance(rows, list) or not rows:
return None
def score(x):
return _to_float(_hit_key(x, ["ratio", "比例", "持股比例", "hold_ratio"])) or 0.0
rows = sorted(rows, key=score, reverse=True)
top = rows[0]
name = _hit_key(top, ["name", "股东", "股东名称", "holder"])
return name, score(top)
def selftest():
# 合成数据仅逻辑自验,非真实行情
assert macd_signal([{"dif": 0.10, "dea": 0.20}, {"dif": 0.30, "dea": 0.25}]) == "golden"
assert ma_alignment({"ma5": 1.0, "ma10": 1.1, "ma20": 1.2}) == "bull"
assert kdj_zone({"j": 110}) == "overbought"
assert boll_band({"up": 1.3, "mid": 1.0, "low": 0.7}) == (1.3, 1.0, 0.7)
holders = [
{"name": "张三", "ratio": 12.3},
{"股东": "李四", "持股比例": 8.1},
{"holder": "王五", "hold_ratio": 5.0},
]
assert top_holder(holders) == ("张三", 12.3)
print("selftest PASS")
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "--selftest":
selftest()
else:
for kind in ("macd", "ma", "kdj", "boll"):
rows, err = fetch_bj_indicator(kind)
print(f"{kind} -> ({rows}, {err})")
for kind in ("capital", "topholder", "flowholder", "hm"):
rows, err = fetch_bj_fin(kind)
print(f"fin.{kind} -> ({rows}, {err})")
if kind == "topholder" and rows:
print(" 第一大股东", top_holder(rows))
print("list_index ->", fetch_bj_list_index())
print("mmwp ->", fetch_bj_mmwp())
5. 代码自验结果
离线 selftest(合成数据,仅验证逻辑,不含任何真实行情):
selftest PASS
联网实测(占位 token,真实返回):
--- 联网实测(占位 token,预期 404 102:Licence证书不存在)---
macd -> (None, 404 102:Licence证书(你的智兔token)不存在)
ma -> (None, 404 102:Licence证书(你的智兔token)不存在)
kdj -> (None, 404 102:Licence证书(你的智兔token)不存在)
boll -> (None, 404 102:Licence证书(你的智兔token)不存在)
fin.capital -> (None, 404 102:Licence证书(你的智兔token)不存在)
fin.topholder -> (None, 404 102:Licence证书(你的智兔token)不存在)
fin.flowholder-> (None, 404 102:Licence证书(你的智兔token)不存在)
fin.hm -> (None, 404 102:Licence证书(你的智兔token)不存在)
list_index -> (None, '404 102:Licence证书(你的智兔token)不存在')
mmwp -> (None, '404 102:Licence证书(你的智兔token)不存在')
把
TOKEN = "你的智兔token"换成你申请的真实 token,上述函数即可打印北交所技术指标与基本面数据。本文未编造任何真实数值。
6. 坑与注意事项
- 102 不代表路径对:
404 102是「证书不存在」(鉴权先于路由),路径合法与否要靠客户端白名单自查。 - 代码必须带
.BJ后缀:写920000会路由错误,必须920000.BJ。 - 技术指标多一个
{adj}除权参数:北交所路径是.../{code}/{period}/{adj},adj取n(不复权)/q(前复权)/f(后复权),漏写会 404。 - 财务字段名中英文混用:「股东姓名」可能是
name/股东/holder,「持股比例」可能是ratio/持股比例/hold_ratio,务必候选键命中。 - 盘口是盘中实时:
mmwp只在交易时段有数据,非交易时段返回可能为空或上一笔。
7. 小结与下篇预告
本篇把「北交所技术指标 + 基本面 + 盘口」拧成了 10 个零依赖接口的最小封装,重点解决了代码带 .BJ 后缀、技术指标含 {adj} 除权参数、财务字段名中英文混用三个坑,复用 #24 的信号函数即可一行出指标信号。
下一篇计划写 #26《港股通更多成交维度》:讲解如何用官方接口拉取港股通的个股通、历史成交、个股盘口等更多维度数据,把「港股通」的成交面也接进零依赖工作流。
8. 免责声明
本文仅演示公开数据接口的用法,所有代码示例均以占位 token 自验,未含任何真实数据;文中合成数据仅为逻辑自验用途,不构成投资建议,亦不承诺收益。
免费领取证书
数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。
领取路径:进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」(邮箱验证 3 步即可拿到 token)。
把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印北交所技术指标与基本面数据。