系列：《Python 量化取数指南》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
数据：由智兔数服提供。更多接口见 智兔数服技术博客。

1. 你将得到什么

• 北向资金 4 类核心端点的完整可跑代码：沪股通、深股通排名、沪/深港通成交额、连续性
• 一套「先打原始结构、再容错取值」的解析习惯
• 一个把北向净额做情绪指标的小示例

2. 本篇取数约定

• 北向资金统一挂在 /ht/nbzj/ 下；ah=沪股通、sgpm/all=深股通排名、hgtc=沪港通成交、sgtc=深港通成交、lxgl=连续性
• 请求：GET https://api.zhituapi.com<path>?token=<你的智兔token>
• 字段名不稳定（如 value/净额/je），一律用 _hit_key 容错
• 北向资金是日频，盘中为实时累计；历史口径区别见跨市场系列，本篇只取实时/当日

3. 核心模板（全系列复用）

import time, json, requests

BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token"      # 演示证书（免费版）即可起步

def _get(path, params=None, timeout=15, retry=3, backoff=1.5):
    params = dict(params or {})
    params["token"] = TOKEN
    url = BASE + path
    last = None
    for i in range(retry):
        try:
            r = requests.get(url, params=params, timeout=timeout)
            if r.status_code != 200:
                last = f"HTTP {r.status_code} {r.text[:120]}"
                time.sleep(backoff * (i + 1)); continue
            try:
                return r.json(), None
            except ValueError:
                last = f"非JSON响应: {r.text[:120]}"
                return None, last
        except requests.RequestException as e:
            last = str(e); time.sleep(backoff * (i + 1))
    return None, last

def _hit_key(d, *keys, default=None):
    if not isinstance(d, dict):
        return default
    for k in keys:
        if k in d and d[k] not in (None, "", []):
            return d[k]
    return default

def _to_float(x, default=float("nan")):
    try:
        return float(x)
    except (TypeError, ValueError):
        return default

4. 跑通示例：北向 4 类端点

def demo_north():
    # 4.1 沪股通当日净买入 /ht/nbzj/ah
    data, err = _get("/ht/nbzj/ah")
    if err:
        print("沪股通失败:", err)
    else:
        val = _to_float(_hit_key(data, "value", "je", "净额", "net"))
        print(f"  沪股通净买入: {val}")

    # 4.2 深股通排名 /ht/nbzj/sgpm/all
    data, err = _get("/ht/nbzj/sgpm/all")
    if err:
        print("深股通排名失败:", err)
    else:
        items = data if isinstance(data, list) else (data.get("data") or [])
        for it in (items or [])[:5]:
            name = _hit_key(it, "name", "mc", "名称")
            net = _to_float(_hit_key(it, "net", "je", "净额"))
            print(f"  深股通 {name} 净买: {net}")

    # 4.3 沪港通成交 /ht/nbzj/hgtc
    data, err = _get("/ht/nbzj/hgtc")
    if err:
        print("沪港通成交失败:", err)
    else:
        print("  沪港通成交原始结构:", type(data).__name__)

    # 4.4 深港通成交 /ht/nbzj/sgtc
    data, err = _get("/ht/nbzj/sgtc")
    if err:
        print("深港通成交失败:", err)
    else:
        print("  深港通成交原始结构:", type(data).__name__)

if __name__ == "__main__":
    demo_north()

返回字段说明：沪股通/深股通通常返回 date（日期）、value/je/net（净额）、name/mc（标的名）。排名类返回 list，每项含 code/dm、name/mc、net/je。先 print(type(data), data) 看真实结构最稳妥。

5. 坑与注意事项

1. 北向是日频累计：盘中拉到的是「当日累计」，不是分时序列；要历史请走历史口径接口。
2. 沪/深分开：沪股通（ah）和深股通（sgpm）是两个端点，没有「北向合一」单端点，自己相加。
3. 字段名三家不统一：净额可能叫 value/je/net，用 _hit_key 候选键。
4. 休市日无数据：周末/节假日返回空或老数据，别当成异常。
5. 成交额 ≠ 净买入：hgtc/sgtc 是成交总额，和「净买入」是两回事，分析时别混。
6. 频率：免费证书拉排名类 list 也有限流，循环多日请加 sleep。

6. 常见报错速查

7. 小结与下一篇预告

小结：北向资金 4 类端点已能拉通，关键是「沪/深分开 + _hit_key 容错 + 区分成交与净买」。这路数据做市场情绪指标很顺手。

下一篇计划写 #04《行情接口选型：实时与历史行情实测》：横向对比实时行情、历史 K 线、历史成交三类端点，告诉你不同场景该选哪个。

8. 免责声明

本文仅演示公开数据接口的用法，所有代码示例均为演示数据，不构成任何投资建议；实际返回字段以接口文档与你的证书权限为准。数据由 智兔数服 提供，更多接口示例见 技术博客。

免费领取证书 / 查看完整接口文档，可前往 智兔数服官网。