← 返回博客列表

【零依赖量化数据实战 #28】沪深A股市场指标与资金流向

2026年09月01日 15:45 · 智兔数服 · 零依赖量化数据实战

摘要:【零依赖量化数据实战 #28】沪深A股市场指标与资金流向 系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK 适用:想用 Python 把 沪深A股 的市场指标(连续涨跌、涨

系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK
适用:想用 Python 把沪深A股的市场指标(连续涨跌、涨跌幅排名、流通市值、市盈率、市净率、ROE)与资金流向(行业/板块/个股净流入、主力连续流入)一次拉齐的量化爱好者;数据由智兔数服提供,不依赖任何行情终端。

1. 你将得到什么

  • 14 个官方接口的最小可用封装,分两组:
  • 市场指标(9):/himk/{lxsl|lxsz|lxxd|zzd|yzd|ltszph|syl|sjl|roe} —— 量缩 / 连涨 / 连跌、本周/本月涨跌幅排名、流通市值排名、市盈率、市净率、ROE 排名(均无路径参数,直接 ?token)。
  • 资金流向(5):/hizj/zjh(证监会行业资金)、/hizj/bk(概念板块资金)、/hizj/ggzl(个股净流入总览)、/hizj/lxlr(主力连续流入)、/hizj/ggjd/{近几日}(个股阶段净流入,阶段 3/5/10/20)。
  • 一个对字段名不敏感的排名函数 rank_by:按候选键(如 流通市值/market_cap)降序取前 N。

2. 端点语义表

GET https://api.zhituapi.com/himk/lxsl?token=你的智兔token     -> 成交量连续缩小
GET https://api.zhituapi.com/himk/lxsz?token=你的智兔token     -> 连续上涨
GET https://api.zhituapi.com/himk/lxxd?token=你的智兔token     -> 连续下跌
GET https://api.zhituapi.com/himk/zzd?token=你的智兔token      -> 本周涨跌幅排名(复权价)
GET https://api.zhituapi.com/himk/yzd?token=你的智兔token      -> 本月涨跌幅排名(复权价)
GET https://api.zhituapi.com/himk/ltszph?token=你的智兔token   -> 流通市值排名
GET https://api.zhituapi.com/himk/syl?token=你的智兔token      -> 市盈率排名(静态/动态)
GET https://api.zhituapi.com/himk/sjl?token=你的智兔token      -> 市净率排名
GET https://api.zhituapi.com/himk/roe?token=你的智兔token      -> ROE 排名(倒序)

GET https://api.zhituapi.com/hizj/zjh?token=你的智兔token      -> 近3/5/10天 证监会行业资金流入
GET https://api.zhituapi.com/hizj/bk?token=你的智兔token       -> 近3/5/10天 概念板块资金流入
GET https://api.zhituapi.com/hizj/ggzl?token=你的智兔token     -> 个股阶段净流入总览(代码升序)
GET https://api.zhituapi.com/hizj/lxlr?token=你的智兔token     -> 主力连续净流入/流出(流入天数倒序)
GET https://api.zhituapi.com/hizj/ggjd/5?token=你的智兔token    -> 个股近5日净流入(阶段 3/5/10/20)

鉴权:token 走查询参数;ggjd 的阶段为路径参数,只接受 3/5/10/20。数据来自 智兔数服(www.zhituapi.com)。

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

排名返回的「流通市值」可能叫 流通市值 / market_cap / lt_mv;「市盈率」可能叫 市盈率 / pe。统一候选键命中:

def _hit_key(d, keys):
    if not isinstance(d, dict):
        return None
    for k in keys:
        if k in d and d[k] is not None:
            return d[k]
    low = {str(x).lower(): x for x in d.keys()}
    for k in keys:
        kl = k.lower()
        if kl in low:
            return d[low[kl]]
    return None

4. 核心模板函数

import sys, requests

BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token"  # 占位,换成你申请的真实 token

def _hit_key(d, keys):
    if not isinstance(d, dict):
        return None
    for k in keys:
        if k in d and d[k] is not None:
            return d[k]
    low = {str(x).lower(): x for x in d.keys()}
    for k in keys:
        kl = k.lower()
        if kl in low:
            return d[low[kl]]
    return None

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

def _get(path, params=None):
    p = dict(params or {})
    p["token"] = TOKEN
    try:
        r = requests.get(f"{BASE}{path}", params=p, timeout=10)
    except Exception as e:
        return None, f"网络异常:{e}"
    if r.status_code != 200:
        return None, f"{r.status_code} {r.text.strip()[:140]}"
    try:
        return r.json(), None
    except Exception:
        return None, f"非 JSON:{r.text.strip()[:140]}"

# 市场指标(9 个,无路径参数)
def fetch_mk(kind):
    return _get(f"/himk/{kind}")

# 资金流向(5 个;ggjd 带阶段路径参数 3/5/10/20)
def fetch_zj(kind, days=None):
    path = f"/hizj/{kind}"
    if kind == "ggjd" and days is not None:
        path = f"/hizj/ggjd/{days}"
    return _get(path)

def rank_by(rows, keys, descending=True, topn=None):
    """按候选键排名(中性,不写红绿)。"""
    if not isinstance(rows, list):
        return rows
    def sc(x):
        return _to_float(_hit_key(x, keys)) or 0.0
    out = sorted(rows, key=sc, reverse=descending)
    return out[:topn] if topn else out

def selftest():
    # 合成数据仅逻辑自验,非真实行情
    mv = [
        {"code": "A", "流通市值": 100.0},
        {"code": "B", "market_cap": 300.0},
        {"code": "C", "lt_mv": 200.0},
    ]
    top = rank_by(mv, ["流通市值", "market_cap", "lt_mv"], topn=2)
    assert [x["code"] for x in top] == ["B", "C"]
    pe = [
        {"code": "X", "市盈率": 12.0},
        {"code": "Y", "pe": 8.0},
    ]
    assert rank_by(pe, ["市盈率", "pe"], topn=1)[0]["code"] == "X"
    for d in (3, 5, 10, 20):
        assert d in (3, 5, 10, 20)
    print("selftest PASS")

if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--selftest":
        selftest()
    else:
        for kind in ("lxsl", "lxsz", "lxxd", "zzd", "yzd", "ltszph", "syl", "sjl", "roe"):
            print(f"mk.{kind} ->", fetch_mk(kind))
        for kind in ("zjh", "bk", "ggzl", "lxlr"):
            print(f"zj.{kind} ->", fetch_zj(kind))
        print("zj.ggjd ->", fetch_zj("ggjd", 5))

5. 代码自验结果

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

selftest PASS

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

--- 联网实测(占位 token,预期 404 102:Licence证书不存在)---
mk.lxsl -> (None, '404 102:Licence证书(你的智兔token)不存在')
mk.lxsz -> (None, '404 102:Licence证书(你的智兔token)不存在')
mk.lxxd -> (None, '404 102:Licence证书(你的智兔token)不存在')
mk.zzd  -> (None, '404 102:Licence证书(你的智兔token)不存在')
mk.yzd  -> (None, '404 102:Licence证书(你的智兔token)不存在')
mk.ltszph -> (None, '404 102:Licence证书(你的智兔token)不存在')
mk.syl  -> (None, '404 102:Licence证书(你的智兔token)不存在')
mk.sjl  -> (None, '404 102:Licence证书(你的智兔token)不存在')
mk.roe  -> (None, '404 102:Licence证书(你的智兔token)不存在')
zj.zjh  -> (None, '404 102:Licence证书(你的智兔token)不存在')
zj.bk   -> (None, '404 102:Licence证书(你的智兔token)不存在')
zj.ggzl -> (None, '404 102:Licence证书(你的智兔token)不存在')
zj.lxlr -> (None, '404 102:Licence证书(你的智兔token)不存在')
zj.ggjd -> (None, '404 102:Licence证书(你的智兔token)不存在')

TOKEN = "你的智兔token" 换成你申请的真实 token,上述函数即可打印沪深A股市场指标与资金流向数据。本文未编造任何真实数值。

6. 坑与注意事项

  1. 102 不代表路径对404 102 是「证书不存在」(鉴权先于路由),路径合法与否要靠客户端白名单自查。
  2. ggjd 阶段仅 3/5/10/20/hizj/ggjd/{近几日} 只接受 3/5/10/20,写别的会路由错误。
  3. 涨跌幅中性处理zzd/yzd 按涨跌幅排名返回,代码里只排序不映射颜色(A股涨红跌绿,但排名本身与颜色无关),避免写死涨跌色。
  4. 排名字段名中英文混用:「流通市值」可能是 流通市值/market_cap/lt_mv,「市盈率」可能是 市盈率/pe,务必候选键命中。
  5. 资金是「阶段」口径zjh/bk 是近 3/5/10 天行业/板块资金,ggjd 是个股阶段净流入,比较时口径要对齐。

7. 小结与下篇预告

本篇把「沪深A股市场指标 + 资金流向」拧成了 14 个零依赖接口的最小封装,重点解决了ggjd 阶段参数涨跌幅排名中性处理排名字段名中英文混用三个坑,配 rank_by 候选键排名即可一行出榜。

本篇(#28)为 #24–#28 阶段收尾:本篇用零依赖接口拉取了沪深A股的市场指标(连续涨跌、涨跌幅排名、流通市值、市盈率、市净率、ROE)与资金流向(行业/板块/个股净流入、主力连续流入)数据,至此《零依赖量化数据实战》#24–#28 完结。

8. 免责声明

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


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

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

把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印沪深A股市场指标与资金流向数据。

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