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.
80 lines
2.6 KiB
80 lines
2.6 KiB
"""东方财富 CHOICE API 杜邦分析抓取工具。
|
|
- 输入: 股票代码列表 (如 ['603233.SH'])
|
|
- 输出: pandas DataFrame (按报告期的杜邦分析指标)
|
|
- 不需要登录, 不需要 cookie, 无文件落盘
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pandas as pd
|
|
|
|
from utils._emchoice import fetch_report
|
|
|
|
|
|
_REPORT_NAME = "RPT_HSF9_FINA_DUPONT"
|
|
# 列清单沿用 docs/api_docs.md 抓包值
|
|
_COLUMNS = (
|
|
"SECUCODE,SECURITY_CODE,SECURITY_NAME_ABBR,ORG_CODE,SECURITY_INNER_CODE,"
|
|
"TRADE_MARKET_CODE,TRADE_MARKET,SECURITY_TYPE,SECURITY_TYPE_CODE,"
|
|
"REPORT_DATE,ROE,NETPROFIT,TOTAL_OPERATE_INCOME,PARENT_NETPROFIT_RATIO,"
|
|
"ASSET_TURNOVER_RATIO,TOTAL_ASSETS,EQUITY_MULTIPLIER,ROE_AVERAGE_PRE,"
|
|
"ROE_DIF,NETPROFIT_TOI,NP_TP,TP_EBIT,EBIT_TOI,STR_YEAR,STR_MONTH,"
|
|
"TOTAL_PROFIT,EBIT,PARENT_NETPROFIT,PRE_TOTAL_ASSETS,AVG_TOTAL_ASSETS,"
|
|
"TOTAL_PARENT_EQUITY,PRETOTAL_PARENT_EQUITY,AVGTOTAL_PARENT_EQUITY,"
|
|
"PARENT_EQUITY_NETMARGIN,IS_NEWEST"
|
|
)
|
|
# 报告期: 1=一季报, 3=三季报, 5=中报, 6=年报
|
|
_REPORT_PERIOD_CODES = '"1","5","3","6","7"'
|
|
|
|
|
|
def _build_params(
|
|
secucode: str,
|
|
start_date: str,
|
|
end_date: str,
|
|
) -> dict[str, str]:
|
|
"""构造 RPT_HSF9_FINA_DUPONT 查询参数."""
|
|
return {
|
|
"reportName": _REPORT_NAME,
|
|
"columns": _COLUMNS,
|
|
"quoteColumns": "",
|
|
"filter": (
|
|
f'(SECUCODE="{secucode}")'
|
|
f"((REPORT_DATE>='{start_date}')(REPORT_DATE<='{end_date}')"
|
|
f"(STR_MONTH in ({_REPORT_PERIOD_CODES}))(|IS_NEWEST=''1''))"
|
|
),
|
|
"pageNumber": "1",
|
|
"pageSize": "",
|
|
"sortTypes": "-1",
|
|
"sortColumns": "REPORT_DATE",
|
|
"source": "choice",
|
|
"client": "SW",
|
|
}
|
|
|
|
|
|
def _fetch_one(secucode: str, years: int) -> pd.DataFrame:
|
|
"""拉一只股票的杜邦分析指标."""
|
|
today = pd.Timestamp.today()
|
|
start = (today - pd.DateOffset(years=years)).strftime("%Y-%m-%d")
|
|
end = today.strftime("%Y-%m-%d")
|
|
|
|
rows = fetch_report(_build_params(secucode, start, end))
|
|
df = pd.DataFrame(rows)
|
|
if not df.empty:
|
|
df["SECUCODE"] = secucode
|
|
return df
|
|
|
|
|
|
def get_dupont(secucodes: list[str], years: int = 3) -> pd.DataFrame:
|
|
"""抓取指定股票的杜邦分析指标并按股票合并。"""
|
|
codes = [code.strip() for code in secucodes if code.strip()]
|
|
if not codes:
|
|
return pd.DataFrame()
|
|
|
|
frames: list[pd.DataFrame] = []
|
|
for code in codes:
|
|
df = _fetch_one(code, years)
|
|
if not df.empty:
|
|
frames.append(df)
|
|
|
|
if not frames:
|
|
return pd.DataFrame()
|
|
return pd.concat(frames, ignore_index=True)
|
|
|