【零依赖量化数据实战 #12】ETF 同类排名、规模与分红:3 个 URL 做选基三维度
摘要:【零依赖量化数据实战 #12】ETF 同类排名、规模与分红:3 个 URL 做选基三维度 系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK 适用:想做基金筛选、同类排名对比
系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK
适用:想做基金筛选、同类排名对比、规模/分红评估,不想装库或对接多个数据源的开发者
你将得到什么
- 3 个「基金排名/规模/分红」端点的路径、返回字段和语义(业绩排名 / 规模 / 分红),照公开文档核对过的,不是猜的
- 一段
fund_compare(code)把排名 + 规模 + 分红拉齐,算出名次、同类百分位、规模、分红四件套 - 字段名容错写法(候选键命中),对字段名不敏感、对形态(list / dict)不敏感
- 完整可复制运行代码,把
你的智兔token换成真实智兔证书即可直接跑
一、三个端点,一张语义表
「选基三维度」拆成 3 个端点,路径前缀统一 /js/,按「开放-股票型-最新」这一类目组织:
GET https://api.zhituapi.com/js/pm/kfpm/kfsjj_gpxjj_zsx 开放式股票型基金 · 业绩排名(按最新)
GET https://api.zhituapi.com/js/gm/kfgm/kfsjj_gpxjj_zsx 开放式股票型基金 · 规模
GET https://api.zhituapi.com/js/jf/kffh/kfsjj_gpxjj_zsx 开放式股票型基金 · 分红
鉴权统一 ?token=<你的智兔token>。三个都是整张榜单/全量返回(排名是列表,规模/分红可能是列表或聚合对象),适合做横向对比。路径里 kfpm/kfsjj_gpxjj_zsx 这类是「分类 key」,照文档原样拼,不要自己改。
一个实际用法:先用
kfpm排名榜定位某基金名次 → 再在kfgm/kffh里取出它的规模与分红 → 三者一合,就能回答「这只基在同类里排第几、盘子多大、分了多少红」。这就是选基三维度。
代码形态提醒:/js/* 路径里的分类 key(如 kfsjj_gpxjj_zsx)就是文档给的枚举值,原样使用;不要自创其他 key(文档里还有 kfjzq/fbjz/kfpm/fbpm/fzyj 等不同分类,按需替换,但必须在文档清单内)。
二、字段名不固定?用候选键命中
基金榜单的返回字段名可能因数据源更新而变化。与其硬编码字段名(改了就全崩),不如用候选键列表命中抽取:
def _hit_key(d, candidates):
"""从字典中按候选键名列表命中第一个存在的键。"""
for k in candidates:
if k in d:
return k
return None
def _to_float(v):
"""把可能是字符串/数字/None 的值安全转 float,转不了返回 None。"""
try:
if v is None:
return None
return float(v)
except (TypeError, ValueError):
return None
def find_in_rank(rank_data, code):
"""在排名列表中定位某基金,返回 (行, 名次, 同类总数)。"""
rows = [r for r in (rank_data or []) if isinstance(r, dict)]
ck = ["code", "dm", "fundcode", "jjdm", "symbol"]
for i, r in enumerate(rows):
if _hit_key(r, ck) and str(r.get(_hit_key(r, ck))) == str(code):
return r, i + 1, len(rows)
return None, None, len(rows)
这样就算数据源把 dm 改成 jjdm,代码不用动。命不中也不报错——只是那只基金取不到,不影响其他。
三、选基三维度:排名 + 规模 + 分红合成
核心不是「把三个接口都调通」,而是把三张榜归一后算出能用的指标。我们关心四件事:
- 名次 = 在排名榜中的第几位
- 同类百分位 = (总数 − 名次) / 总数 × 100(越高越好)
- 规模 = 该基金的规模(来自规模榜)
- 分红 = 该基金的分红(来自分红榜)
import sys, time
import requests
TOKEN = "你的智兔token" # ← 换成你的真实智兔token
BASE = "https://api.zhituapi.com"
RANK_ENDPOINTS = {
"rank": ("/js/pm/kfpm/kfsjj_gpxjj_zsx", "业绩排名"),
"scale": ("/js/gm/kfgm/kfsjj_gpxjj_zsx", "基金规模"),
"div": ("/js/jf/kffh/kfsjj_gpxjj_zsx", "基金分红"),
}
def fetch(name, token=TOKEN, timeout=15):
"""拉一个排名/规模/分红端点,返回 (data, err)。"""
path_tmpl, _ = RANK_ENDPOINTS[name]
url = f"{BASE}{path_tmpl}"
try:
r = requests.get(url, params={"token": token}, timeout=timeout)
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
def fund_compare(code, token=TOKEN, sleep=0.2):
"""把排名/规模/分红拉齐,做一只基金的「选基三维度」对比。"""
if token == "你的智兔token":
print("[演示] TOKEN 为占位符,下面请求会返回错误码契约;请换成真实 token 后再跑。")
rank, rerr = fetch("rank", token)
scale, serr = fetch("scale", token)
div, derr = fetch("div", token)
if rerr:
print(f" rank: 暂不可用:{rerr}")
else:
print(f" rank (业绩排名): {len(rank) if isinstance(rank, list) else type(rank).__name__} 条")
if serr:
print(f" scale: 暂不可用:{serr}")
else:
print(f" scale (基金规模): {len(scale) if isinstance(scale, list) else '聚合对象'}")
if derr:
print(f" div: 暂不可用:{derr}")
else:
print(f" div (基金分红): {len(div) if isinstance(div, list) else '聚合对象'}")
time.sleep(sleep)
row, pos, total = find_in_rank(rank, code)
percentile = round((total - pos) / total * 100, 1) if (row and pos) else None
ck = ["code", "dm", "fundcode", "jjdm", "symbol"]
sc, dv = None, None
if isinstance(scale, list):
for r in scale:
if isinstance(r, dict) and _hit_key(r, ck) and str(r.get(_hit_key(r, ck))) == str(code):
sc = _to_float(r.get(_hit_key(r, ["scale", "gm", "size", " asset", "total", "guimo"])))
break
if isinstance(div, list):
for r in div:
if isinstance(r, dict) and _hit_key(r, ck) and str(r.get(_hit_key(r, ck))) == str(code):
dv = _to_float(r.get(_hit_key(r, ["div", "fh", "fhsp", "dividend", "fhje"])))
break
print("=== 选基三维度 ===")
print(f" 代码={code} 排名名次={pos}/{total} 前{percentile}% 规模={sc} 分红={dv}")
return {"code": code, "pos": pos, "total": total, "percentile": percentile, "scale": sc, "div": dv}
sleep=0.2 是限频保护——3 个端点 + 0.2 秒间隔 ≈ 0.6 秒。要对比多只基金,把 fund_compare 套一层循环,外面再撑一个更大的间隔即可。
四、代码自验结果
离线自测 6 项全 PASS(真实输出,不联网):
selftest PASS: _hit_key 命中 / 排名定位与百分位 / 找不到兜底 / 规模分红抽取 / 空数据兜底 / 字段名不敏感 共 6 项
其中「排名定位与百分位」用的是合成数据(沪深300ETF 排第 2 / 共 3 只),断言百分位 = (3−2)/3 ≈ 33.3%;「规模分红抽取」用合成数据(规模 200.5、分红 0.15)——这些是逻辑自验用的假数,不是真实数值。
联网跑(占位 token)真实输出:
[演示] TOKEN 为占位符,下面请求会返回错误码契约;请换成真实 token 后再跑。
rank: 暂不可用:404 102:Licence证书(你的智兔token)不存在
scale: 暂不可用:404 102:Licence证书(你的智兔token)不存在
div: 暂不可用:404 102:Licence证书(你的智兔token)不存在
=== 选基三维度 ===
代码=510300 排名名次=None/0 前None% 规模=None 分红=None
三个端点契约一致,单端点失败不影响整体流程、程序友好退出、对空数据不崩。本文未编造任何排名/规模/分红数值——换成覆盖该接口的正式证书后重跑,即可打印真实选基三维度。
五、坑与注意事项
坑 #1:102 不代表路径写对了。 鉴权发生在路由匹配之前。路径故意写错配无效 token,同样返回 102。路径合法性只能靠客户端按白名单自查——本文 3 个路径是照公开文档核对的。实测 HTTP 状态码是 404 而不是 403,重点是看到 102:Licence证书... 就知道是证书问题。
坑 #2:分类 key 别自创。 /js/pm/kfpm/kfsjj_gpxjj_zsx 里的 kfpm/kfsjj_gpxjj_zsx 是文档定义的枚举;要换分类(如债券型、封闭型)必须用文档清单里真实存在的 key(如 kfjzq/fbjz/fzyj 等),不能自己拼。
坑 #3:排名榜可能不含规模/分红。 排名、规模、分红是三张独立的榜,同一只基金在三张榜里的行不一定对齐;定位某只基金时用统一的代码候选键(code/dm/fundcode/jjdm/symbol)去匹配,别假设行号一致。
坑 #4:百分位只在同一分类内有意义。 「前 33%」是相对「开放式股票型」这一类的;跨分类比百分位没有意义。对比时要保证排名/规模/分红用的是同一分类 key。
小结与下篇预告
这篇你拿到了 3 个基金排名/规模/分红端点(业绩排名 / 规模 / 分红)、字段名容错抽取写法、以及一个把三张榜合成「选基三维度」的模板,算出名次、同类百分位、规模、分红,顺带避开了 102 误判、分类 key 自创、榜单不对齐、跨类百分位四个坑。
下一篇(#13)讲ETF 盘中实时估值与估算涨跌扫描——用 /jh/gz/gz(估值列表)+ /jh/gz/zx/{code}(单只估值)拉全市场盘中估值,按「估算涨跌幅」筛异动,把选基升级成盘中实时盯估。
免费领取证书
数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。
领取路径:进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」(邮箱验证 3 步即可拿到 token)。
把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印基金排名规模与分红数据。
免责声明
本文仅演示公开数据接口的用法,所有代码示例均以占位 token 自验,未含任何真实排名/规模/分红数据;文中合成数据仅为逻辑自验用途,不构成投资建议,亦不承诺收益。投资决策请基于你自己的判断与风险承受力。