【跨市场数据实战 #01】北向资金数据怎么取：21个接口、2种口径，别把季频当日用

【跨市场数据实战 #01】北向资金数据怎么取：21个接口、2种口径，别把季频当日用

系列：《跨市场数据实战》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想看北向/南向资金、但被「日频还是季频」「万元还是百万」绕晕的读者。本篇给港股通板块 21 个端点的分组地图、三套口径的归一化代码、季频降级守卫，全部只依赖 requests，所有示例均为演示数据，不构成收益承诺。

1. 你将得到什么

读完这一篇，你能拿走四样东西：

代码全部自包含，复制进 .py 直接能跑，不依赖 numpy / pandas。

2. 本篇取数约定

• 全部接口都是 GET + query 参数，token 放在查询串里（?token=xxx）；
• 统一基址 https://api.zhituapi.com；
• 代码块里的 你的token 是占位符，换成你的 token 即可；
• 所有接口路径均取自官方文档。

3. 21 个端点分 8 组

先建立地图。港股通板块一共 21 个端点，按用途分：

注意最后一行——前 7 组都是日频，只有个股排名组是季频。这就是全篇最大的坑。

4. 核心模板函数

import requests, time

BASE = "https://api.zhituapi.com"
TOKEN = "你的token"

# ———- 1. 字段容错与类型归一 ———-
def _hit_key(d, *cands, default=None):
"""字段容错：接口偶发大小写/中英文混用时，按顺序取第一个非空值"""
if not isinstance(d, dict):
return default
for c in cands:
if c in d and d[c] not in (None, "", "-", "null"):
return d[c]
low = {str(k).lower(): v for k, v in d.items()}
for c in cands:
v = low.get(str(c).lower())
if v not in (None, "", "-", "null"):
return v
return default

def _to_float(v, default=None):
try:
if v in (None, "", "-", "null", "None"):
return default
return float(v)
except (TypeError, ValueError):
return default

# ———- 2. 统一请求：重试 + 退避 + 降级 ———-
def _get(path, params=None, timeout=10, retries=2, backoff=0.6, default=None):
"""返回 JSON；失败重试 retries 次仍失败则返回 {'_error': 原因}"""
q = {"token": TOKEN}
if params:
q.update(params)
last = ""
for i in range(retries + 1):
try:
r = requests.get(BASE + path, params=q, timeout=timeout)
if r.status_code == 200:
try:
return r.json()
except ValueError:
return default
last = "HTTP %s %s" % (r.status_code, (r.text or "").strip()[:80])
except Exception as e:
last = "%s: %s" % (type(e).__name__, e)
if i < retries:
time.sleep(backoff * (i + 1))
return {"_error": last}

# ———- 3. 单位归一：全部换算到「亿元」 ———-
UNIT_IN_YI = { # 各接口原始单位 -> 1 单位等于多少亿元
"万": 1e-4, # lxgl 的 netbuy / netin / remain
"百万": 1e-2, # bxls / nxls 的 bx / hgt / sgt
"万元": 1e-4, # bxzl 的 bxall / hgtall / sgtall
"元": 1e-8,
}

def to_yi(v, unit="万"):
"""原始金额 -> 亿元；None / 空串 / 异常值返回 None（不猜、不填 0）"""
f = _to_float(v)
if f is None:
return None
return round(f * UNIT_IN_YI.get(unit, 1.0), 4)

# ———- 4. 三套口径的归一化 ———-
def norm_overview(rows):
"""/ht/nbzj/lxgl -> 只留北向四条通道（单位：亿元）"""
out = []
for r in rows or []:
if _hit_key(r, "dir", default="") != "北向":
continue
out.append({
"板块": _hit_key(r, "tname", default="-"),
"净买额": to_yi(_hit_key(r, "netbuy"), "万"),
"净流入": to_yi(_hit_key(r, "netin"), "万"),
"余额": to_yi(_hit_key(r, "remain"), "万"),
"上涨": _to_float(_hit_key(r, "up"), 0),
"下跌": _to_float(_hit_key(r, "down"), 0),
"状态": _hit_key(r, "status", default="-"),
})
return out

def norm_series(rows, unit="百万"):
"""/ht/nbzj/bxls|nxls/{jd} -> 统一 [{'日期','北向','沪股通','深股通'}]，亿元"""
out = []
for r in rows or []:
out.append({
"日期": _hit_key(r, "t", "date", default="-"),
"北向": to_yi(_hit_key(r, "bx", "nx"), unit),
"沪股通": to_yi(_hit_key(r, "hgt", "ggth"), unit),
"深股通": to_yi(_hit_key(r, "sgt", "ggts"), unit),
})
return out

def norm_total(d):
"""/ht/nbzj/bxzl -> 累计净流入（亿元）"""
return {
"北向累计": to_yi(_hit_key(d, "bxall"), "万元"),
"沪股通累计": to_yi(_hit_key(d, "hgtall"), "万元"),
"深股通累计": to_yi(_hit_key(d, "sgtall"), "万元"),
}

# ———- 5. 季频守卫 ———-
SNAPSHOT_FIELDS = ("jrcg", "jrsz", "jrltb", "jrzgbb") # 快照日持股，仍有效
PERIOD_FIELDS = ("zqzc", "zqsz", "zqszzf", "zqltb", "zqzgbb") # 周期增持，多已停更

def norm_rank(rows):
"""/ht/nbzj/bxpm|hgpm|sgpm/{zq} -> (排名列表, 元信息)；自动丢弃全 null 的周期字段"""
if not rows:
return [], {"季度标记": "unknown", "快照日": "-", "丢弃字段": [], "条数": 0}
first = rows[0]
quarterly = str(_hit_key(first, "_freq", default="")).lower() == "quarterly"
dropped = [f for f in PERIOD_FIELDS
if all(_hit_key(r, f) in (None, "", "null") for r in rows)]
out = [{
"代码": _hit_key(r, "dm", default="-"),
"名称": _hit_key(r, "mc", default="-"),
"板块": _hit_key(r, "ssbk", default="-"),
"持股万股": _to_float(_hit_key(r, "jrcg")),
"持股市值万元": _to_float(_hit_key(r, "jrsz")),
"占流通股比": _to_float(_hit_key(r, "jrltb")),
} for r in rows]
meta = {"季度标记": "quarterly" if quarterly else "daily",
"快照日": _hit_key(first, "t", default="-"),
"丢弃字段": dropped, "条数": len(out)}
return out, meta

def assert_not_daily(meta):
"""把季频数据当日频用是本篇最想拦住的错误；季频返回 False"""
return meta.get("季度标记") != "quarterly"

# ———- 6. 成分列表交叉核对 ———-
def duplicate_slots(rows_a, rows_b):
"""ggth / ggts 两个港股通成分列表可能完全一致，返回重复代码数"""
a = set(_hit_key(r, "dm", default="") for r in rows_a or [])
b = set(_hit_key(r, "dm", default="") for r in rows_b or [])
if not a or not b:
return 0
return len(a & b)

# ———- 7. 取数封装 ———-
def fetch_overview(): return _get("/ht/nbzj/lxgl", default=[])
def fetch_north_series(jd="1"): return _get("/ht/nbzj/bxls/%s" % jd, default=[])
def fetch_south_series(jd="1"): return _get("/ht/nbzj/nxls/%s" % jd, default=[])
def fetch_north_total(): return _get("/ht/nbzj/bxzl", default={})
def fetch_slot(kind="hgtc"): return _get("/ht/nbzj/%s" % kind, default=[])
def fetch_rank(scope="bx", zq="1"):
m = {"bx": "bxpm", "hgt": "hgpm", "sgt": "sgpm"}
return _get("/ht/nbzj/%s/%s" % (m.get(scope, "bxpm"), zq), default=[])

# ———- 8. 校验 ———-
def run_check():
# 1) 字段容错
assert _hit_key({"Dm": "000001", "mc": "平安银行"}, "dm") == "000001"
assert _to_float("-") is None and _to_float("12.5") == 12.5

# 2) 单位换算：三个口径都归一到亿元
assert to_yi(10000, "万") == 1.0 # 10000 万 = 1 亿
assert to_yi(100, "百万") == 1.0 # 100 百万 = 1 亿
assert to_yi(10000, "万元") == 1.0
assert to_yi(None, "万") is None

# 3) 概览只留北向
fake = [
{"dir": "北向", "tname": "沪股通(港>沪)", "netbuy": "12345", "netin": "13000",
"remain": "98765", "up": "800", "down": "300", "status": "3"},
{"dir": "南向", "tname": "港股通(沪>港)", "netbuy": "9999", "netin": "9999",
"remain": "0", "up": "1", "down": "1", "status": "3"},
]
ov = norm_overview(fake)
assert len(ov) == 1 and ov[0]["板块"].startswith("沪股通")
assert ov[0]["净买额"] == 1.2345

# 4) 走势归一：分项之和应等于合计
ser = norm_series([{"t": "2026-08-28", "bx": "120", "hgt": "60", "sgt": "60"}])
assert ser[0]["日期"] == "2026-08-28" and ser[0]["北向"] == 1.2
assert abs(ser[0]["沪股通"] + ser[0]["深股通"] – 1.2) < 1e-9

# 5) 累计总览
tot = norm_total({"bxall": "190000", "hgtall": "100000", "sgtall": "90000"})
assert tot["北向累计"] == 19.0 and tot["沪股通累计"] == 10.0

# 6) 季频守卫：zq* 全 null 应被丢弃，且不可当日频用
fake_rank = [
{"dm": "600519", "mc": "X", "ssbk": "沪股通", "jrcg": "9000", "jrsz": "15000000",
"jrltb": "7.1", "zqzc": None, "zqsz": None, "zqszzf": None,
"zqltb": None, "zqzgbb": None, "t": "2026-06-30", "_freq": "quarterly"},
{"dm": "000858", "mc": "Y", "ssbk": "深股通", "jrcg": "5000", "jrsz": "800000",
"jrltb": "3.2", "zqzc": None, "zqsz": None, "zqszzf": None,
"zqltb": None, "zqzgbb": None, "t": "2026-06-30", "_freq": "quarterly"},
]
rank, meta = norm_rank(fake_rank)
assert meta["季度标记"] == "quarterly"
assert meta["快照日"] == "2026-06-30"
assert set(meta["丢弃字段"]) == set(PERIOD_FIELDS)
assert assert_not_daily(meta) is False # 季频 -> 不允许当日频用
assert rank[0]["持股市值万元"] > rank[1]["持股市值万元"]

# 7) 成分列表重复检测
a = [{"dm": "00700"}, {"dm": "00001"}]
assert duplicate_slots(a, a) == 2
assert duplicate_slots(a, [{"dm": "09988"}]) == 0

print("校验通过")

if __name__ == "__main__":
run_check()
print("-" * 62)
for name, path in [("当日概览", "/ht/nbzj/lxgl"),
("北向历史走势", "/ht/nbzj/bxls/1"),
("北向历史总览", "/ht/nbzj/bxzl"),
("沪股通成分股", "/ht/nbzj/hgtc"),
("AH股比价", "/ht/nbzj/ah"),
("北向个股排名", "/ht/nbzj/bxpm/1")]:
data = _get(path, default=[])
if isinstance(data, dict) and "_error" in data:
print("%-12s %-20s -> %s" % (name, path, data["_error"][:60]))
else:
print("%-12s %-20s -> %d 条" % (name, path, len(data)))

5. 跑通示例

把上面的代码复制到本地，填入你的 token 即可直接运行：它会请求对应接口、拉取真实数据，并输出归一化后的结构化字典（各字段含义见前文各小节）。

6. 坑与注意事项

坑 1：个股排名已经不是日频了。
/ht/nbzj/bxpm/{zq}、hgpm/{zq}、sgpm/{zq} 三个接口的上游日频持股明细已停更，现在返回的是最近季末的持股快照。返回的 zq* 系列字段（周期增持股数、增持市值、增幅、占流通股比、占总股本比）多数为 null，只有 jrcg / jrsz / jrltb / jrzgbb 这四个快照字段有值，_freq 会标成 quarterly。
所以"北向连续 5 日净增持某某股"这类逻辑现在已经做不出来了，能做的是"最新季末北向持股排名"。norm_rank 会自动把全 null 的周期字段丢掉，assert_not_daily 会在你误用时返回 False。

坑 2：单位有三种，别混着加。
lxgl 用「万」，bxls/nxls 用「百万」，bxzl 用「万元」。直接把 bx 和 netbuy 相加会差 100 倍。to_yi() 把三者统一到亿元，这是本篇最容易被忽略、也最容易出静默错误的地方。

坑 3：成交净买额 ≠ 资金净流入。
netbuy 是买入成交额减卖出成交额，netin 是当日限额减当日余额（含挂单未成交部分）。两者口径不同、数值不同，别当成同一个指标轮着用。

坑 4：总览接口只有累计口径。
/ht/nbzj/bxzl 只返回 bxall / hgtall / sgtall 三个累计值，没有近一月/近六月/近一年的分阶段字段。想要分阶段，去用 /ht/nbzj/bxls/{jd} 自己按日期切窗口。

坑 5：两个港股通成分列表可能完全一致。
/ht/nbzj/ggth（港股通·沪）与 /ht/nbzj/ggts（港股通·深）实测返回的成分列表可能完全相同。如果你的策略要区分这两个通道，先用 duplicate_slots() 比一下再决定要不要合并去重，别默认它们天然不同。

7. 小结与下篇预告

本篇把港股通板块 21 个端点分成 8 组，给出三套口径（万/百万/万元）统一到亿元的归一化代码，并用 norm_rank + assert_not_daily 把"季频当日频用"这个错误显式拦住。

下一篇计划写 #02《基金持仓穿透怎么做：从基金列表到重仓股变动的32个接口》：用 /jh 与 /js 两组接口，从基金代码反查它持有哪些股票，再统计哪些股票被最多基金共同重仓。

8. 免责声明

本文仅演示跨市场资金数据的取数与口径归一化方法，所有代码示例均为演示数据，未含任何真实行情数值，不构成投资建议，亦不承诺收益。