← 返回博客列表

【零依赖量化数据实战 #25】北交所技术指标与基本面

2026年08月28日 11:11 · 智兔数服 · 零依赖量化数据实战

摘要:【零依赖量化数据实战 #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&lt=30
  -> 北交所股票 日线 MACD(除权类型 n=不复权 / q=前复权 / f=后复权)
GET https://api.zhituapi.com/bj/history/ma/920000.BJ/d/n?token=你的智兔token&lt=30
GET https://api.zhituapi.com/bj/history/kdj/920000.BJ/d/n?token=你的智兔token&lt=30
GET https://api.zhituapi.com/bj/history/boll/920000.BJ/d/n?token=你的智兔token&lt=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 格式 YYYYMMDDlt 取最近 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. 坑与注意事项

  1. 102 不代表路径对404 102 是「证书不存在」(鉴权先于路由),路径合法与否要靠客户端白名单自查。
  2. 代码必须带 .BJ 后缀:写 920000 会路由错误,必须 920000.BJ
  3. 技术指标多一个 {adj} 除权参数:北交所路径是 .../{code}/{period}/{adj}adjn(不复权)/q(前复权)/f(后复权),漏写会 404。
  4. 财务字段名中英文混用:「股东姓名」可能是 name/股东/holder,「持股比例」可能是 ratio/持股比例/hold_ratio,务必候选键命中。
  5. 盘口是盘中实时mmwp 只在交易时段有数据,非交易时段返回可能为空或上一笔。

7. 小结与下篇预告

本篇把「北交所技术指标 + 基本面 + 盘口」拧成了 10 个零依赖接口的最小封装,重点解决了代码带 .BJ 后缀技术指标含 {adj} 除权参数财务字段名中英文混用三个坑,复用 #24 的信号函数即可一行出指标信号。

下一篇计划写 #26《港股通更多成交维度》:讲解如何用官方接口拉取港股通的个股通、历史成交、个股盘口等更多维度数据,把「港股通」的成交面也接进零依赖工作流。

8. 免责声明

本文仅演示公开数据接口的用法,所有代码示例均以占位 token 自验,未含任何真实数据;文中合成数据仅为逻辑自验用途,不构成投资建议,亦不承诺收益。


免费领取证书
数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。

领取路径:进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」(邮箱验证 3 步即可拿到 token)。

把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印北交所技术指标与基本面数据。

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