量化数据开发实战系列（第 16 篇）：股东数据分析：十大股东、十大流通股东、股东户数变化跟踪

量化数据开发实战系列（第 16 篇）：股东数据分析：十大股东、十大流通股东、股东户数变化跟踪

前言

第 15 篇我们用财务主要指标接口做了「高 ROE、合理资产负债率」的自研质量筛选，并反复强调：接口返回的 ROE、毛利率、资产负债率都只是原始裸字段，没有任何预制因子。本篇承接第 15 篇结尾的预告：

第 16 篇：股东数据实战，十大股东、十大流通股东、股东户数。

如果说财务质量回答的是「公司赚不赚钱、稳不稳」，那股东数据回答的就是「谁在持有、筹码在往哪里走」。必盈提供三个同属 hsstock/financial/ 路径的股东类接口：topholder（十大股东）、flowholder（十大流通股东）、hm（股东户数）。这三个接口都带 .SZ/.SH 后缀 + ?st=&et= 时间区间，返回的全部是原始裸字段——股东名称、持股数量、持股比例、股东总数等，没有任何「筹码集中度因子」。

本篇把这三类数据清洗、按报告期归档入库，并全部用 Python 自研计算两类信号：股东户数环比增减、筹码集中/分散信号统计。所有衍生指标均基于接口原始字段运算得到，文中明确标注「接口仅返回原始裸字段，以下指标为自研」。

衔接说明：本篇紧接第 15 篇「财务指标」，是大纲第三部分（基本面量化专题）的第 4 篇；代码复用基础代码块与第 13、14、15 篇的 fetch_stock_list()、with_exchange_suffix() 约定。

一、本篇涉及接口与字段解读

三个接口同属 hsstock/financial/ 路径，结构一致，股票代码均需带交易所后缀（如 000001.SZ / 000001.SH），后缀由《股票列表》的 jys 字段决定。st/et 格式均为 YYYYMMDD（如 20240101），不设置则为全部数据。

1.1 十大股东 financial/topholder

• 接口地址：http://api.biyingapi.com/hsstock/financial/topholder/{股票代码（如000001.SZ）}/{LICENCE}?st=开始时间&et=结束时间
• 接口说明：根据《股票列表》得到的股票代码获取公司十大股东。
• 数据更新：每日 0 点
• 请求频率：1 分钟 300 次 | 包年版 1 分钟 3 千次 | 白金版 1 分钟 6 千次
• 返回格式：标准 JSON 数组 [{},…{}]

1.2 十大流通股东 financial/flowholder

• 接口地址：http://api.biyingapi.com/hsstock/financial/flowholder/{股票代码（如000001.SZ）}/{LICENCE}?st=开始时间&et=结束时间
• 接口说明：根据《股票列表》得到的股票代码获取公司十大流通股东。
• 数据更新：每日 0 点
• 返回格式：标准 JSON 数组

1.3 股东户数 financial/hm

• 接口地址：http://api.biyingapi.com/hsstock/financial/hm/{股票代码（如000001.SZ）}/{LICENCE}?st=开始时间&et=结束时间
• 接口说明：根据《股票列表》得到的股票代码获取公司股东数。
• 数据更新：每日 0 点
• 返回格式：标准 JSON 数组

1.4 关键返回字段（逐字取自官方 docx）

topholder / flowholder 字段完全一致：

hm 股东户数字段：

⚠️ 三个接口不返回任何 chip_concentration（筹码集中度）、holder_change_rate（户数变化率）之类的预制指标。股东户数环比、筹码信号必须自己算。

二、自研衍生指标（接口仅返回原始裸字段，以下指标为自研）

接口仅返回原始裸字段，以下指标为自研。

本篇在原始字段之上，用代码自研以下三类指标，全部不依赖任何预制因子：

【仅为数据演示，不构成投资建议】

三、数据表设计

三张表分别承载三类数据，均带 UNIQUE 防重，按报告期（jzrq）归档，保证重复采集幂等。

3.1 top_holder（十大股东）

3.2 flow_holder（十大流通股东）

3.3 holder_count（股东户数）

四、完整可运行代码

以下基础代码块逐字复用全系列统一约定（签名不变），随后为本篇新增逻辑。演示用 limit=N 控制规模，避免刷爆接口配额。

import requests
import logging
import time
import pandas as pd
import numpy as np
import sqlite3
from apscheduler.schedulers.background import BackgroundScheduler

# ========== 全局配置 ==========
LICENCE = "你的licence"
DB_PATH = "quant.db"
LOG_FILE = "quant_collect.log"

# ———-日志初始化———-
logging.basicConfig(
filename=LOG_FILE,
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
filemode="a"
)
logger = logging.getLogger(__name__)

# ———-带重试HTTP请求（复用）———-
def biying_api_get_retry(full_url, timeout=15, max_retry=3):
for attempt in range(1, max_retry + 1):
try:
resp = requests.get(full_url, timeout=timeout)
if resp.status_code == 200:
return resp.json()
logger.warning(f"HTTP状态码异常：{resp.status_code}，第{attempt}次重试")
except Exception as e:
logger.warning(f"网络请求异常，第{attempt}次重试，错误信息：{str(e)}")
time.sleep(2)
logger.error("达到最大重试次数，接口请求失败")
return []

# ———-股票列表（复用，作为代码/交易所参数来源）———-
def fetch_stock_list():
url = f"http://api.biyingapi.com/hslt/list/{LICENCE}"
return biying_api_get_retry(url)

# ———-交易所后缀（复用第13篇逻辑）———-
def with_exchange_suffix(stock_code, exchange):
"""财报接口要求带交易所后缀：sh->.SH, sz->.SZ"""
suffix = "SH" if str(exchange).lower() == "sh" else "SZ"
return f"{stock_code}.{suffix}"

# =====================本篇新增：股东数据=====================
def init_holder_tables(db_name="quant.db"):
conn = sqlite3.connect(db_name)
cur = conn.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS top_holder (
id INTEGER PRIMARY KEY AUTOINCREMENT,
stock_code TEXT, ggrq TEXT, jzrq TEXT,
gdmc TEXT, gdlx TEXT, cgsl REAL, bdyy TEXT,
cgbl REAL, gfxz TEXT, cgpm INTEGER,
UNIQUE(stock_code,jzrq,cgpm)
)
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS flow_holder (
id INTEGER PRIMARY KEY AUTOINCREMENT,
stock_code TEXT, ggrq TEXT, jzrq TEXT,
gdmc TEXT, gdlx TEXT, cgsl REAL, bdyy TEXT,
cgbl REAL, gfxz TEXT, cgpm INTEGER,
UNIQUE(stock_code,jzrq,cgpm)
)
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS holder_count (
id INTEGER PRIMARY KEY AUTOINCREMENT,
stock_code TEXT, jzrq TEXT,
gdzs REAL, agdhs REAL, bgdhs REAL, hgdhs REAL,
yltgdhs REAL, wltgdhs REAL,
UNIQUE(stock_code,jzrq)
)
""")
conn.commit()
conn.close()
logger.info("股东数据表初始化完成")

HOLDER_COLS = ["ggrq","jzrq","gdmc","gdlx","cgsl","bdyy","cgbl","gfxz","cgpm"]
HM_COLS = ["jzrq","gdzs","agdhs","bgdhs","hgdhs","yltgdhs","wltgdhs"]

def fetch_holder_api(path_name, stock_code_with_suffix, st="", et=""):
"""path_name 取值：topholder / flowholder / hm，均带后缀 + st/et"""
base = f"http://api.biyingapi.com/hsstock/financial/{path_name}/{stock_code_with_suffix}/{LICENCE}"
url = f"{base}?st={st}&et={et}" if (st and et) else base
return biying_api_get_retry(url)

def clean_holder(raw_json, stock_code_with_suffix):
df = pd.DataFrame(raw_json)
if len(df) == 0:
return df
df = df.replace([None, "null", ""], np.nan)
for dcol in ["ggrq", "jzrq"]:
if dcol in df.columns:
df[dcol] = pd.to_datetime(df[dcol], errors="coerce").dt.strftime("%Y-%m-%d")
for c in ["cgsl", "cgbl"]:
if c in df.columns:
df[c] = pd.to_numeric(df[c], errors="coerce")
if "cgpm" in df.columns:
df["cgpm"] = pd.to_numeric(df["cgpm"], errors="coerce")
df["stock_code"] = stock_code_with_suffix
keep = ["stock_code"] + HOLDER_COLS
keep = [c for c in keep if c in df.columns]
return df[keep].dropna(subset=["jzrq", "cgpm"])

def clean_holder_count(raw_json, stock_code_with_suffix):
df = pd.DataFrame(raw_json)
if len(df) == 0:
return df
df = df.replace([None, "null", ""], np.nan)
if "jzrq" in df.columns:
df["jzrq"] = pd.to_datetime(df["jzrq"], errors="coerce").dt.strftime("%Y-%m-%d")
for c in HM_COLS:
if c in df.columns and c != "jzrq":
df[c] = pd.to_numeric(df[c], errors="coerce")
df["stock_code"] = stock_code_with_suffix
keep = ["stock_code"] + HM_COLS
keep = [c for c in keep if c in df.columns]
return df[keep].dropna(subset=["jzrq"])

def save_df_ignore(df, table, cols, db_name="quant.db"):
"""UNIQUE 防重 + INSERT OR IGNORE 幂等入库（满足 CONVENTIONS 入库约定）"""
if len(df) == 0:
return
conn = sqlite3.connect(db_name)
cur = conn.cursor()
placeholders = ",".join(["?"] * len(cols))
col_names = ",".join(cols)
sql = f"INSERT OR IGNORE INTO {table} ({col_names}) VALUES ({placeholders})"
for _, row in df[cols].iterrows():
cur.execute(sql, tuple(row))
conn.commit()
conn.close()
logger.info(f"{table} 入库 {len(df)} 条（已忽略重复）")

# ———-本篇核心：自研衍生指标（接口仅返回原始裸字段）———-
def calc_top10_holding_ratio(db_name="quant.db", table="top_holder"):
"""自研：前十大（流通）股东累计持股比例 = 按 jzrq 分组 sum(cgbl)"""
conn = sqlite3.connect(db_name)
df = pd.read_sql(f"SELECT stock_code, jzrq, cgbl FROM {table}", conn)
conn.close()
if len(df) == 0 or df["cgbl"].isna().all():
return pd.DataFrame()
grp = df.groupby(["stock_code", "jzrq"], as_index=False)["cgbl"].sum()
grp = grp.rename(columns={"cgbl": "top10_cgbl_sum"})
return grp.sort_values(["stock_code", "jzrq"])

def calc_holder_count_mom(db_name="quant.db"):
"""自研：股东户数(gdzs)环比变化率 + 筹码集中/分散信号，按 jzrq 升序"""
conn = sqlite3.connect(db_name)
df = pd.read_sql("SELECT stock_code, jzrq, gdzs FROM holder_count", conn)
conn.close()
if len(df) == 0 or df["gdzs"].isna().all():
return pd.DataFrame()
df = df.sort_values(["stock_code", "jzrq"]).reset_index(drop=True)
df["gdzs_prev"] = df.groupby("stock_code")["gdzs"].shift(1)
df["holder_mom_pct"] = (df["gdzs"] – df["gdzs_prev"]) / df["gdzs_prev"] * 100
# 自研筹码信号：户数下降->集中；上升->分散
df["chip_signal"] = np.where(df["holder_mom_pct"] < 0, "集中",
np.where(df["holder_mom_pct"] > 0, "分散", "持平"))
return df

def collect_one_holder(stock_code, exchange, st="", et=""):
code_suf = with_exchange_suffix(stock_code, exchange)
for path, table in [("topholder", "top_holder"), ("flowholder", "flow_holder")]:
raw = fetch_holder_api(path, code_suf, st, et)
if raw:
clean = clean_holder(raw, code_suf)
save_df_ignore(clean, table, ["stock_code"] + HOLDER_COLS)
raw_hm = fetch_holder_api("hm", code_suf, st, et)
if raw_hm:
clean_hm = clean_holder_count(raw_hm, code_suf)
save_df_ignore(clean_hm, "holder_count", ["stock_code"] + HM_COLS)

def collect_all_holder(limit=20, st="", et=""):
"""示例：仅采集前 limit 只作为演示，全量请去掉 .head(limit)【仅为数据演示，不构成投资建议】"""
init_holder_tables()
raw_list = fetch_stock_list()
if not raw_list:
logger.error("股票列表为空，终止采集")
return
df_list = pd.DataFrame(raw_list)
for _, row in df_list.head(limit).iterrows():
try:
collect_one_holder(row["dm"], row.get("jys", ""), st, et)
except Exception as e:
logger.warning(f"标的{row['dm']}股东数据采集失败：{str(e)}")
time.sleep(0.3) # 控制请求频率
# 自研指标演示输出
top10 = calc_top10_holding_ratio(table="top_holder")
flow10 = calc_top10_holding_ratio(table="flow_holder")
mom = calc_holder_count_mom()
print("=" * 64)
print("自研衍生指标演示：十大股东累计持股 / 股东户数环比 / 筹码信号")
print("【仅为数据演示，不构成投资建议】")
print("=" * 64)
if len(top10):
print("\\n[前十大股东累计持股比例] 最新一期：")
print(top10.sort_values("jzrq").groupby("stock_code").tail(1).to_string(index=False))
if len(flow10):
print("\\n[前十大流通股东累计持股比例] 最新一期：")
print(flow10.sort_values("jzrq").groupby("stock_code").tail(1).to_string(index=False))
if len(mom):
print("\\n[股东户数环比 + 筹码信号] 最新一期：")
print(mom.sort_values("jzrq").groupby("stock_code").tail(1)
[["stock_code","jzrq","gdzs","holder_mom_pct","chip_signal"]].to_string(index=False))
logger.info("股东数据采集与自研指标演示完成")

def start_scheduler():
"""股东类接口数据每日 0 点更新，定时 00:30 盘后采集"""
sched = BackgroundScheduler()
sched.add_job(collect_all_holder, "cron", hour=0, minute=30)
sched.start()
logger.info("股东数据定时任务已启动")

if __name__ == "__main__":
collect_all_holder(limit=20)

五、业务关键点

六、拓展练习

七、下篇预告

系列第 17 篇：解禁、打新收益数据接口实战

股东结构看的是「谁在持有」，而解禁数据看的是「未来筹码供给压力从哪来」、打新收益看的是「新股申购的历史回报」。下一篇用解禁提示、打新收益类接口，把供给端数据采回来，继续坚持「接口只给原始裸字段、全部指标代码自研」的系列铁律。

免责申明：文中所有数据处理逻辑仅为编程演示，仅为数据演示，不构成投资建议。市场有风险，投资需谨慎。