【零依赖量化数据实战 #28】沪深A股市场指标与资金流向
摘要:【零依赖量化数据实战 #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. 坑与注意事项
- 102 不代表路径对:
404 102是「证书不存在」(鉴权先于路由),路径合法与否要靠客户端白名单自查。 ggjd阶段仅 3/5/10/20:/hizj/ggjd/{近几日}只接受3/5/10/20,写别的会路由错误。- 涨跌幅中性处理:
zzd/yzd按涨跌幅排名返回,代码里只排序不映射颜色(A股涨红跌绿,但排名本身与颜色无关),避免写死涨跌色。 - 排名字段名中英文混用:「流通市值」可能是
流通市值/market_cap/lt_mv,「市盈率」可能是市盈率/pe,务必候选键命中。 - 资金是「阶段」口径:
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股市场指标与资金流向数据。