← 返回博客列表

【零依赖量化数据实战 #22】港股通成交排名与南北向资金总览

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

摘要:【零依赖量化数据实战 #22】港股通成交排名与南北向资金总览 系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK 适用:想用 Python 把「北向/南向资金流向、港股通个股

系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK
适用:想用 Python 把「北向/南向资金流向、港股通个股成交排名」拉成结构化数据的量化爱好者;数据由智兔数服提供,不依赖任何券商终端。

1. 你将得到什么

  • 4 个官方接口的最小可用封装,覆盖南北向资金总览港股通个股成交排名两类需求:
  • GET /ht/nbzj/bxzl:北向资金累计净流入总览(北向合计 / 沪股通 / 深股通)
  • GET /ht/nbzj/nxzl:南向资金历史总览
  • GET /ht/nbzj/bxpm/{zq}:北向个股周期排名
  • GET /ht/nbzj/hgpm/{zq}:沪股通个股周期排名
  • 一个对字段名不敏感的排名抽取函数:上游返回里「持股市值」可能叫 mv / market_value / hold_value / value 等,用候选键命中,不写死。
  • 实测提醒:bxpm / hgpm 在 2026-08 上游已停更日频明细、降级为季频持股快照,周期增持字段(zq*)多为 null,调用时请知悉口径。

2. 端点语义表

GET https://api.zhituapi.com/ht/nbzj/bxzl?token=你的智兔token
  -> 北向资金累计净流入总览
     返回(累计口径):bxall(北向合计) / hgtall(沪股通) / sgtall(深股通)

GET https://api.zhituapi.com/ht/nbzj/nxzl?token=你的智兔token
  -> 南向资金历史总览(字段名可能异于北向,用候选键命中)

GET https://api.zhituapi.com/ht/nbzj/bxpm/{zq}?token=你的智兔token
  -> 北向个股周期排名;{zq}=周期参数(当前上游为季频快照)
     返回:list,每项含个股代码与持股市值类字段

GET https://api.zhituapi.com/ht/nbzj/hgpm/{zq}?token=你的智兔token
  -> 沪股通个股周期排名(同季频口径;zq* 多为 null)

鉴权:token 走查询参数;排名走路径参数 {zq};返回形态 bxzl/nxzl 为 dict、bxpm/hgpm 为 list。数据来自 智兔数服(www.zhituapi.com)。

3. 字段名不固定?用候选键命中

上游三类返回(北向总览 / 南向总览 / 个股排名)字段名都不统一。统一用 _hit_key + _to_float 兜底抽取:

def _hit_key(d, candidates):
    for k in candidates:
        if k in d:
            return k
    return None

def _to_float(v):
    try:
        return None if v is None else float(v)
    except (TypeError, ValueError):
        return None

4. 核心模板函数

import requests

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

def _hit_key(d, candidates):
    for k in candidates:
        if k in d:
            return k
    return None

def _to_float(v):
    try:
        return None if v is None else float(v)
    except (TypeError, ValueError):
        return None

def fetch(path, params=None):
    p = dict(params or {})
    p["token"] = TOKEN
    try:
        r = requests.get(f"{BASE}{path}", params=p, timeout=15)
    except requests.RequestException as e:
        return None, f"网络异常:{e}"
    if r.status_code != 200:
        return None, f"{r.status_code} {r.text.strip()[:140]}"
    try:
        payload = r.json()
    except ValueError:
        return None, f"非 JSON:{r.text[:140]}"
    if isinstance(payload, dict):
        detail = payload.get("detail") or payload.get("error")
        return None, f"业务错误:{detail or list(payload)[:6]}"
    return payload, None

# ===== #22 港股通成交排名与南北向资金总览 =====

_MV_CAND = ["mv", "market_value", "hold_value", "value", "持股", "持股市值"]

def northsouth_total():
    """北向累计净流入总览 bxzl + 南向总览 nxzl(字段名不固定,靠候选键命中)"""
    b, err1 = fetch("/ht/nbzj/bxzl")
    s, err2 = fetch("/ht/nbzj/nxzl")
    if err1 or err2:
        return None, (err1 or err2)
    def pick(d):
        if not isinstance(d, dict):
            return None
        k = _hit_key(d, ["bxall", "total", "value", "north"])
        return _to_float(d.get(k)) if k else None
    def pick_s(d):
        if not isinstance(d, dict):
            return None
        k = _hit_key(d, ["sxall", "total", "value", "south"])
        return _to_float(d.get(k)) if k else None
    return {"north": pick(b), "south": pick_s(s)}, None

def _rank(data):
    rows = data if isinstance(data, list) else []
    return sorted(rows, key=lambda x: _to_float(x.get(_hit_key(x, _MV_CAND))) or 0.0, reverse=True)

def northbound_rank(zq="d"):
    """北向个股周期排名 bxpm/{zq};返回按持股市值倒序的列表"""
    data, err = fetch(f"/ht/nbzj/bxpm/{zq}")
    if err:
        return None, err
    return _rank(data), None

def hg_rank(zq="d"):
    """沪股通个股周期排名 hgpm/{zq}(2026-08 上游停更日频,降级为季频持股快照,zq* 多为 null)"""
    data, err = fetch(f"/ht/nbzj/hgpm/{zq}")
    if err:
        return None, err
    return _rank(data), None

5. 代码自验结果

离线 selftest(合成数据,仅验证逻辑,不含任何真实行情):

PASS: #22 逻辑自验通过(合成数据,无真实行情)

联网实测(占位 token,真实返回):

--- 联网实测(占位 token,预期 404 102:Licence证书不存在)---
northsouth_total -> (None, '404 102:Licence证书(你的智兔token)不存在')
northbound_rank  -> (None, '404 102:Licence证书(你的智兔token)不存在')
hg_rank          -> (None, '404 102:Licence证书(你的智兔token)不存在')

TOKEN = "你的智兔token" 换成你申请的真实 token,上述函数即可打印真实资金流向与排名数据。本文未编造任何真实数值。

6. 坑与注意事项

  1. 102 不代表路径对404 102 是「证书不存在」(鉴权先于路由),路径合法与否要靠客户端白名单自查,不能凭报错反推接口可用。
  2. 季频口径变更bxpm / hgpm 上游在 2026-08 停更日频持股明细,降级为季频快照;zq* 周期增持字段多为 null,落盘时带 _freq=quarterly 标记更稳妥。
  3. 北向/南向字段名不同:北向总览返回 bxall/hgtall/sgtall,南向总览键名可能不同,务必用候选键命中,别硬写 bxall 抽南向。
  4. 排名 list 形态bxpm/hgpm 是数组,按「持股市值」类字段倒序;字段名随上游变,用 _MV_CAND 候选键兜底。
  5. 路径参数 {zq} 必填bxpm/{zq}zq 会 404 路由错误;zq 取值按官方文档(当前季频快照)。

7. 小结与下篇预告

本篇把「南北向资金总览 + 港股通个股成交排名」拧成了 4 个零依赖接口的最小封装,重点解决了字段名不统一(候选键命中)与季频口径变更两个坑,排名按持股市值倒序即可直接落表。

下一篇计划写 #23《沪深个股财务三表与股东结构》:讲解如何用官方接口拉取单只沪深股票的资产负债表、利润表、现金流量表、财务指标、十大股东与股本等基本面数据。

8. 免责声明

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


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

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

把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印港股通成交排名与持股数据。

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