← 返回博客列表

【智兔数服|别再到处找免费股票数据API:官方204个接口32篇讲透 #08】个股资金流向靠猜?主力净流入排名一次拉

2026年10月03日 08:43 · 智兔数服 · 别再到处找免费股票数据API:官方204个接口32篇讲透

摘要:【智兔数服|别再到处找免费股票数据API:官方204个接口32篇讲透 #08】个股资金流向靠猜?主力净流入排名一次拉 系列:智兔数服|别再到处找免费股票数据API:官方204个接口32篇讲透|连载项目 · 纯 GET 取数 &

系列:智兔数服|别再到处找免费股票数据API:官方204个接口32篇讲透|连载项目 · 纯 GET 取数 · 仅依赖 requests
适用:想做资金流因子 / 主力追踪,但还在行情软件里肉眼看主力净流入、还分不清大单中单的读者;数据由智兔数服提供。本篇给沪深A股「历史成交(资金流) / 指定日期成交 / 基础信息」3 个端点的分组地图、一次拉全的代码、主力净流入字段的坑,全部只依赖 requests,所有示例均为演示数据,不构成投资建议。

1. 你将得到什么

读完这一篇,你能拿走四样东西:

  1. 一张分组地图:3 个资金流端点按「逐日资金流 / 单日明细 / 基础档案」分成 3 类;
  2. 一次拉全的代码:/hs/history/transaction 拿逐日主力净流入,/hs/instrument 拿股本 / 市值底表;
  3. 主力净流入的取法:接口返回的是「主力 / 超大单 / 大单 / 中单 / 小单」五档净流入,要挑你需要的档;
  4. 三个真实踩坑点,都是第一次用几乎一定会踩的。

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

2. 本篇取数约定

  • 全部接口都是 GET + query 参数,token 放在查询串里(?token=xxx);
  • 统一基址 https://api.zhituapi.com;
  • 代码块里的 你的智兔token 是占位符,换成你的 token 即可;
  • 所有接口路径均取自官方已验证文档,跨篇零重复。
  • 数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。

3. 3 个端点分 3 类

先建立地图。资金流类一共 3 个端点,按用途分:

类 端点 用途 更新频率
逐日资金流 /hs/history/transaction 逐日资金流向(主力 / 大单净流入) 每日盘后
单日明细 /hs/history/transaction/{日期} 指定日期的资金流明细 盘后
基础档案 /hs/instrument 股本 / 市值 / 行业等基础信息 每日盘后

4. 核心模板函数

import requests, time

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

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


def _get(path, params=None, timeout=10, retries=2, backoff=0.6, default=None):
    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}


def fetch_cashflow(code):
    return _get("/hs/history/transaction", {"code": code}, default=[])
def fetch_cashflow_date(code, date):
    return _get("/hs/history/transaction/%s" % date, {"code": code}, default=[])
def fetch_instrument(code):
    return _get("/hs/instrument", {"code": code}, default={})


def latest_main_net(code):
    rows = fetch_cashflow(code)
    if isinstance(rows, dict) and "_error" in rows:
        return None, rows["_error"]
    row = (rows or [{}])[-1]
    return {
        "date":  _hit_key(row, "date", "rq", "日期", default=""),
        "main_net": _to_float(_hit_key(row, "main_net", "zljlr", "主力净流入", default=None)),
        "big_net":  _to_float(_hit_key(row, "big_net", "ddjlr", "大单净流入", default=None)),
    }, None


def run_check():
    fake = [{"date": "20240909", "main_net": "1234567", "big_net": "456000"},
            {"date": "20240910", "main_net": "-890123", "big_net": "-120000"}]
    _orig = fetch_cashflow
    fetch_cashflow = lambda c: fake
    m, err = latest_main_net("000001.SZ")
    fetch_cashflow = _orig
    assert err is None and m["date"] == "20240910" and m["main_net"] == -890123
    assert fetch_cashflow_date("000001.SZ", "20240910") is not None
    print("校验通过")


if __name__ == "__main__":
    run_check()
    print("-" * 62)
    code = "000001.SZ"
    for name, fn in [("逐日资金流", lambda: fetch_cashflow(code)),
                     ("单日明细", lambda: fetch_cashflow_date(code, "20240910")),
                     ("基础信息", lambda: fetch_instrument(code))]:
        data = fn()
        if isinstance(data, dict) and "_error" in data:
            print("%-10s -> %s" % (name, data["_error"][:60]))
        else:
            print("%-10s -> %d 条" % (name, len(data) if isinstance(data, list) else 1))

5. 跑通示例

把上面的代码复制到本地,填入你的 智兔token 即可直接运行:传入代码,latest_main_net 取最近交易日的主力净流入与大单净流入,fetch_cashflow_date 还能拉指定日期明细(各字段含义见前文各小节)。

6. 坑与注意事项

坑 1:资金流字段是「五档净流入」,别只盯主力。
/hs/history/transaction 返回「主力 / 超大单 / 大单 / 中单 / 小单」五档净流入,单位通常是元。做主力追踪取 main_net,做游资追踪取 big_net,别把中单小单误当主力,否则信号会反。

坑 2:净流入有正负,单位要看清。
净流入为正表示资金净流入、为负表示净流出,字段值通常是元(可能很大)。做排行前先确认单位,别把「-890123」当成「-0.89 亿」直接排序,量级错了排名全乱。

坑 3:历史成交是盘后数据,当日盘中不在里面。
/hs/history/transaction 是每日盘后更新的历史资金流,不含当日实时。做盘中主力监控要配合实时接口(见实时行情篇),别把历史资金流当实时看,会滞后一个交易日。

7. 小结与下篇预告

本篇把沪深A股 3 个资金流端点分成 3 类,latest_main_net 一行取最近交易日主力净流入,_to_float 处理净流入单位。

下一篇:《【智兔数服|别再到处找免费股票数据API:官方204个接口32篇讲透 #09】MACD_KDJ还要自己算?4大技术指标一次拿》:用 /hs/history/{macd,ma,boll,kdj} 一组接口,把 MACD / MA / BOLL / KDJ 四大技术指标直接取全。

8. 免责声明

本文仅演示沪深A股资金流数据的取数方法,所有代码示例均为演示数据,未含任何真实行情数值,不构成投资建议,亦不承诺收益。


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

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

把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印个股资金流向数据。

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