A股量化数据工程：从 REST 接口到策略信号（第 16 篇）：北交所股东与持股分析
前言
股东户数变化、十大股东持股比例，是观察「筹码集中 / 分散」的窗口。北交所流动性薄，股东结构变动对股价更敏感。接口 bj/financial/topholder（十大股东）、flowholder（流通股东）、hm（股东户数）给出这套数据，本篇自研持股集中度与户数变化指标。
一、接口原始字段梳理
1）北交所十大股东
接口地址：https://api.biyingapi.com/bj/financial/topholder/{LICENCE}
| 字段 | 说明 |
|---|---|
| plrq | 披露日期 |
| jzrq | 截止日期 |
| gdmc | 股东名称 |
| gdlx | 股东类型 |
| cgsl | 持股数量 |
| bdyy | 变动原因 |
| cgbl | 持股比例 |
| gfxz | 股份性质 |
| cgpm | 持股排名 |
2）北交所十大流通股东
接口地址：https://api.biyingapi.com/bj/financial/flowholder/{LICENCE}
| 字段 | 说明 |
|---|---|
| ggrq | 公告日期 |
| jzrq | 截止日期 |
| gdmc | 股东名称 |
| gdlx | 股东类型 |
| cgsl | 持股数量 |
| bdyy | 变动原因 |
| cgbl | 持股比例 |
| gfxz | 股份性质 |
| cgpm | 持股排名 |
3）北交所股东户数
接口地址：https://api.biyingapi.com/bj/financial/hm/{LICENCE}
| 字段 | 说明 |
|---|---|
| plrq | 披露日期 |
| jzrq | 截止日期 |
| gdzs | 股东总数 |
| agdhs | A 股股东户数 |
| bgdhs | B 股股东户数 |
| hgdhs | H 股股东户数 |
| yltgdhs | 机构投资者户数 |
| wltgdhs | 流通股东户数 |
二、自研衍生统计指标
- 持股集中度：十大股东
cgbl之和，越高筹码越集中。 - 户数变化率：本期
gdzs与上期之差 / 上期，下降=筹码集中（通常偏多）。 - 机构占比：
yltgdhs / gdzs，衡量专业资金参与度。
集中度与变化率均基于原始
cgbl/gdzs字段二次计算。
三、数据表设计
CREATE TABLE IF NOT EXISTS bj_holder (
dm TEXT, jzrq TEXT, gdmc TEXT, gdlx TEXT,
cgsl REAL, cgbl REAL, cgpm INTEGER,
PRIMARY KEY (dm, jzrq, gdmc)
);
CREATE TABLE IF NOT EXISTS bj_hm (
dm TEXT, jzrq TEXT,
gdzs REAL, agdhs REAL, yltgdhs REAL, wltgdhs REAL,
PRIMARY KEY (dm, jzrq)
);
CREATE TABLE IF NOT EXISTS bj_holder_metric (
dm TEXT, jzrq TEXT,
top10_ratio REAL, hm_change REAL, inst_ratio REAL,
PRIMARY KEY (dm, jzrq)
);
四、完整代码
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(path):
d = biying_api_get_retry(f"{BASE}/bj/financial/{path}/{LICENCE}")
return d.get("data", []) if isinstance(d, dict) else []
def build_metric(dm, holders, hm_rows):
top10 = sum((h.get("cgbl") or 0) for h in holders if h.get("cgpm") <= 10)
con = sqlite3.connect(DB_PATH)
con.executemany(
"INSERT OR REPLACE INTO bj_holder(dm,jzrq,gdmc,gdlx,cgsl,cgbl,cgpm)"
"VALUES(?,?,?,?,?,?,?)",
[[dm, h.get("jzrq"), h.get("gdmc"), h.get("gdlx"), h.get("cgsl"),
h.get("cgbl"), h.get("cgpm")] for h in holders])
# 户数变化：按 jzrq 排序取相邻两期
hm_rows = sorted(hm_rows, key=lambda x: x.get("jzrq") or "")
out = []
for i, hr in enumerate(hm_rows):
gdzs = hr.get("gdzs") or 0
inst = hr.get("yltgdhs") or 0
chg = 0.0
if i > 0 and hm_rows[i - 1].get("gdzs"):
prev = hm_rows[i - 1]["gdzs"]
chg = round((gdzs - prev) / prev, 4) if prev else 0.0
inst_ratio = round(inst / gdzs, 4) if gdzs else 0.0
out.append((dm, hr.get("jzrq"), round(top10, 3), chg, inst_ratio))
con.executemany(
"INSERT OR REPLACE INTO bj_holder_metric(dm,jzrq,top10_ratio,hm_change,inst_ratio)"
"VALUES(?,?,?,?,?)", out)
con.commit(); con.close()
return out
if __name__ == "__main__":
holders = fetch("topholder")
hm_rows = fetch("hm")
# 演示：取前若干 dm
dms = list({h.get("dm") for h in holders})
for dm in dms[:3]:
h = [x for x in holders if x.get("dm") == dm]
hr = [x for x in hm_rows if x.get("dm") == dm]
m = build_metric(dm, h, hr)
print(f"{dm} 最新集中度={m[-1][2]} 户数变化={m[-1][3]} 机构占比={m[-1][4]}")
五、关键业务说明
- 集中度非越高越好：十大股东占比高可能是机构锁仓（稳），也可能是单一大股东（风险），需结合
gdlx类型判断。 - 户数下降的含义：
gdzs环比下降通常解读为筹码集中，但北交所样本小，单期噪声大，看趋势更稳。 - 机构占比看质量：
yltgdhs占比提升，代表专业资金关注，与第 15 篇财务质量交叉验证。
六、拓展练习
- 把
bj_holder_metric的top10_ratio与第 15 篇bj_factor盈利分 Join，找「高集中 + 高盈利」组合。 - 用
hm_change连续两期下降筛选「筹码持续集中」标的（仅统计）。 - 把
gdlx分类统计北交所中「自然人 vs 机构」股东结构差异。
下篇预告
A股量化数据工程：从 REST 接口到策略信号（第 17 篇）：创新高与新低扫描
下一篇进入市场扫描维度，用
himk/pzxg与himk/pzxd拉取盘中创新高 / 新低个股，自研「新高强度」扫描器。
免责申明：本篇全部代码、接口调用仅作为量化编程学习演示，仅为数据演示，不构成投资建议。市场有风险，投资需谨慎。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/2601_96840033/article/details/167345747
