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.
 
 
 
EMWebApi/utils/fin_operate.py

70 lines
2.0 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_OPERATEABILITY"
_COLUMNS = "OPERATEABILITY"
# 报告期: 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_OPERATEABILITY 查询参数."""
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": "",
"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_operate_ability(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)