You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
120 lines
4.5 KiB
120 lines
4.5 KiB
"""东方财富 CHOICE API 所属指数抓取工具。
|
|
- 输入: 股票代码列表 (如 ['603233.SH'])
|
|
- 输出: pandas DataFrame (指数分类 / 纳入指数统计 / 指数成分权重)
|
|
- 不需要登录, 不需要 cookie, 无文件落盘
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pandas as pd
|
|
|
|
from utils._emchoice import fetch_report
|
|
|
|
|
|
def _fetch_one(secucode: str, params: dict[str, str]) -> pd.DataFrame:
|
|
"""按给定参数拉一只股票的数据."""
|
|
rows = fetch_report(params)
|
|
df = pd.DataFrame(rows)
|
|
if not df.empty:
|
|
df["SECUCODE"] = secucode
|
|
return df
|
|
|
|
|
|
def _concat(frames: list[pd.DataFrame]) -> pd.DataFrame:
|
|
if not frames:
|
|
return pd.DataFrame()
|
|
return pd.concat(frames, ignore_index=True)
|
|
|
|
|
|
# ---------------------------------------------------------------- 指数分类
|
|
|
|
def _build_classif_params(secucode: str) -> dict[str, str]:
|
|
"""构造 RPT_HSF9_BELONGINDEX_CLASSIF 查询参数 (股票所属的指数分类)."""
|
|
return {
|
|
"reportName": "RPT_HSF9_BELONGINDEX_CLASSIF",
|
|
"columns": "INDEX_CLASSIF_CODE,INDEX_CLASSIF",
|
|
"quoteColumns": "",
|
|
"filter": f'(SECUCODE="{secucode}")(STR_TYPE="1")',
|
|
"distinct": "INDEX_CLASSIF_CODE,INDEX_CLASSIF",
|
|
"pageNumber": "",
|
|
"pageSize": "",
|
|
"source": "CHOICE",
|
|
"client": "SW",
|
|
}
|
|
|
|
|
|
def get_index_classif(secucodes: list[str]) -> pd.DataFrame:
|
|
"""抓取指定股票所属的指数分类列表."""
|
|
codes = [code.strip() for code in secucodes if code.strip()]
|
|
if not codes:
|
|
return pd.DataFrame()
|
|
|
|
frames = [_fetch_one(code, _build_classif_params(code)) for code in codes]
|
|
return _concat([df for df in frames if not df.empty])
|
|
|
|
|
|
# ---------------------------------------------------------------- 纳入指数统计
|
|
|
|
def _build_belong_index_params(secucode: str) -> dict[str, str]:
|
|
"""构造 RPT_CUSTOM_HSF9_BELONG_INDEX 查询参数 (纳入指数数量统计)."""
|
|
return {
|
|
"reportName": "RPT_CUSTOM_HSF9_BELONG_INDEX",
|
|
"filter": f'(SECUCODE="{secucode}")',
|
|
"source": "CHOICE",
|
|
"client": "SW",
|
|
}
|
|
|
|
|
|
def get_belong_index(secucodes: list[str]) -> pd.DataFrame:
|
|
"""抓取指定股票纳入指数的统计 (指数/基金数量)."""
|
|
codes = [code.strip() for code in secucodes if code.strip()]
|
|
if not codes:
|
|
return pd.DataFrame()
|
|
|
|
frames = [_fetch_one(code, _build_belong_index_params(code)) for code in codes]
|
|
return _concat([df for df in frames if not df.empty])
|
|
|
|
|
|
# ---------------------------------------------------------------- 指数成分权重
|
|
|
|
# 只保留主要指数分类 (剔除行业/风格类), 沿用 docs/api_docs.md 抓包值
|
|
_INDEX_CLASSIF_LIKE = (
|
|
'(INDEX_CLASSIF_CODE like "%001001%")(|INDEX_CLASSIF_CODE like "%001002%")'
|
|
'(|INDEX_CLASSIF_CODE like "%001003%")(|INDEX_CLASSIF_CODE like "%001004%")'
|
|
'(|INDEX_CLASSIF_CODE like "%001005%")(|INDEX_CLASSIF_CODE like "%001007%")'
|
|
'(|INDEX_CLASSIF_CODE like "%001009%")(|INDEX_CLASSIF_CODE like "%001010%")'
|
|
'(|INDEX_CLASSIF_CODE like "%001011%")(|INDEX_CLASSIF_CODE like "%001017%")'
|
|
'(|INDEX_CLASSIF_CODE like "%001020%")(|INDEX_CLASSIF_CODE like "%001022%")'
|
|
'(|INDEX_CLASSIF_CODE like "%001025%")(|INDEX_CLASSIF_CODE like "%001030%")'
|
|
'(|INDEX_CLASSIF_CODE like "%001006%")(|INDEX_CLASSIF_CODE like "%001024%")'
|
|
'(|INDEX_CLASSIF_CODE like "%013003%")(|INDEX_CLASSIF_CODE like "%014%")'
|
|
)
|
|
|
|
|
|
def _build_membership_params(secucode: str) -> dict[str, str]:
|
|
"""构造 RPT_HSF9_BELONG_INDEX 查询参数 (指数成分及权重)."""
|
|
return {
|
|
"reportName": "RPT_HSF9_BELONG_INDEX",
|
|
"columns": (
|
|
"INDEX_CODE,INDEX_INNER_CODE,TRADE_DATE,SECUCODE,SECURITY_CODE,"
|
|
"SECURITY_NAME_ABBR,INCLUDE_DATE,MAKER_NAME,INDEX_NAME_ABBR,"
|
|
"SECURITY_INNER_CODE,INDEX_CLASSIF_CODE,INDEX_CLASSIF,WEIGHT,NUM,"
|
|
"TOTAL_SCALE,WEIGHT_DATE,CURRENCY"
|
|
),
|
|
"filter": f'(SECUCODE="{secucode}")({_INDEX_CLASSIF_LIKE})',
|
|
"sortColumns": "NUM,INDEX_CODE",
|
|
"sortTypes": "-1,1",
|
|
"pageNumber": "1",
|
|
"pageSize": "50",
|
|
"source": "CHOICE",
|
|
"client": "SW",
|
|
}
|
|
|
|
|
|
def get_index_membership(secucodes: list[str]) -> pd.DataFrame:
|
|
"""抓取指定股票所属的指数成分及权重."""
|
|
codes = [code.strip() for code in secucodes if code.strip()]
|
|
if not codes:
|
|
return pd.DataFrame()
|
|
|
|
frames = [_fetch_one(code, _build_membership_params(code)) for code in codes]
|
|
return _concat([df for df in frames if not df.empty])
|
|
|