Initial EMWeb API implementation

Add financial summary and quarterly endpoints with project configuration and documentation.
2026-08-15 00:20:47 +08:00
commit cd7ca92cff
8 changed files with 1724 additions and 0 deletions
@@ -0,0 +1,88 @@
+"""东方财富 CHOICE API 季度财务摘要抓取工具。"""
+from __future__ import annotations
+
+import pandas as pd
+import requests
+
+from utils.fin_summary import BASE_URL, HEADERS, TIMEOUT
+
+
+_REPORT_NAME = "RPT_HSF9_FIN_SUMMARYQUARTERNEW"
+_COLUMNS = "HSF9_FIN_SUMMARYQUARTER"
+
+
+def _build_filter(
+    secucode: str,
+    years: int,
+    report_dates: list[str] | None,
+) -> str:
+    if report_dates is not None:
+        dates = ",".join(f"'{date}'" for date in report_dates)
+        return f'(secuCode="{secucode}")(REPORTDATE in ({dates}))'
+
+    today = pd.Timestamp.today()
+    start = (today - pd.DateOffset(years=years)).strftime("%Y-%m-%d")
+    end = today.strftime("%Y-%m-%d")
+    return (
+        f'(secuCode="{secucode}")'
+        f"((REPORTDATE>='{start}')(REPORTDATE<='{end}')"
+        '(dateType in ("1","2","3","4","7"))(|IS_NEWEST="1"))'
+        '(TYPE_CODE in ("1"))'
+    )
+
+
+def _fetch_one(
+    secucode: str,
+    years: int,
+    report_dates: list[str] | None,
+) -> pd.DataFrame:
+    params = {
+        "reportName": _REPORT_NAME,
+        "columns": _COLUMNS,
+        "filter": _build_filter(secucode, years, report_dates),
+        "pageNumber": "1",
+        "pageSize": "500",
+        "sortTypes": "-1,1" if report_dates is None else "-1",
+        "sortColumns": "REPORTDATE,TYPE" if report_dates is None else "REPORTDATE",
+        "source": "CHOICE",
+        "client": "SW",
+    }
+    response = requests.post(BASE_URL, headers=HEADERS, data=params, timeout=TIMEOUT)
+    response.raise_for_status()
+    payload = response.json()
+    if not payload.get("success"):
+        raise RuntimeError(f"{secucode}: 接口返回失败: {payload}")
+
+    rows = payload.get("result", {}).get("data") or []
+    df = pd.DataFrame(rows)
+    if not df.empty:
+        df["SECUCODE"] = secucode
+    return df
+
+
+def get_fin_quarterly_summary(
+    secucodes: list[str],
+    years: int = 3,
+    report_dates: list[str] | None = None,
+) -> 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, report_dates)
+        if not df.empty:
+            frames.append(df)
+
+    if not frames:
+        return pd.DataFrame()
+
+    result = pd.concat(frames, ignore_index=True)
+    if {"SECUCODE", "REPORT_DATE"}.issubset(result.columns):
+        result = result.drop_duplicates(["SECUCODE", "REPORT_DATE"], keep="first")
+        result = result.sort_values(
+            ["SECUCODE", "REPORT_DATE"], ascending=[True, False]
+        )
+    return result.reset_index(drop=True)