from __future__ import annotations from dataclasses import dataclass import json from pathlib import Path import urllib.parse import urllib.request import duckdb import pandas as pd from .config import Settings SECTOR_FIELDS = [ "SECUCODE", "REPORT_DATE", "NOTICE_DATE", "ROEJQ", "NET_INTEREST_MARGIN", "NET_INTEREST_SPREAD", "BLDKBBL", "LOAN_PROVISION_RATIO", "NEWCAPITALADER", "FIRST_ADEQUACY_RATIO", "LIQUIDITY_COVERAGE_RATIO", "RISK_COVERAGE", "NET_CAPITAL_LIABILITIES", "NET_ASSETS_LIABILITIES", "SOLVENCY_AR", ] @dataclass(frozen=True) class FinancialSectorDataResult: rows: int report_period: str rebuilt: bool output_path: Path warning: str = "" def _fetch_chunk(secucodes: list[str], report_period: str) -> list[dict[str, object]]: quoted = ",".join(f'"{code}"' for code in secucodes) params = { "reportName": "RPT_F10_FINANCE_MAINFINADATA", "columns": ",".join(SECTOR_FIELDS), "filter": f"(SECUCODE in ({quoted}))(REPORT_DATE='{report_period}')", "pageNumber": 1, "pageSize": 100, "source": "HSF10", "client": "PC", } url = "https://datacenter.eastmoney.com/securities/api/data/v1/get?" + urllib.parse.urlencode(params) request = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0", "Referer": "https://emweb.securities.eastmoney.com/"}) with urllib.request.urlopen(request, timeout=30) as response: payload = json.loads(response.read()) result = payload.get("result") or {} return result.get("data") or [] def refresh_financial_sector_data(settings: Settings, force: bool = False) -> FinancialSectorDataResult: output_path = settings.processed_dir / "financial_sector_metrics.parquet" finance = pd.read_parquet(settings.financial_pit_path, columns=["report_period"]) report_period = pd.Timestamp(finance["report_period"].max()).date().isoformat() if output_path.exists() and not force: cached = pd.read_parquet(output_path) if not cached.empty and str(pd.Timestamp(cached["report_period"].max()).date()) == report_period: return FinancialSectorDataResult(len(cached), report_period, False, output_path) master = pd.read_parquet(settings.processed_dir / "security_master.parquet") financial = master.loc[master["industry_l1_code"].isin(["X50", "X51"]), [ "source_code", "industry_l2_code", "industry_l2_name", ]].copy() financial["secucode"] = financial["source_code"].str[2:] + "." + financial["source_code"].str[:2].str.upper() rows: list[dict[str, object]] = [] codes = financial["secucode"].tolist() for start in range(0, len(codes), 40): rows.extend(_fetch_chunk(codes[start:start + 40], report_period)) frame = pd.DataFrame(rows).rename(columns={ "SECUCODE": "secucode", "REPORT_DATE": "report_period", "NOTICE_DATE": "notice_date", "ROEJQ": "sector_roe", "NET_INTEREST_MARGIN": "net_interest_margin", "NET_INTEREST_SPREAD": "net_interest_spread", "BLDKBBL": "provision_coverage_ratio", "LOAN_PROVISION_RATIO": "loan_provision_ratio", "NEWCAPITALADER": "capital_adequacy_ratio", "FIRST_ADEQUACY_RATIO": "tier1_capital_adequacy_ratio", "LIQUIDITY_COVERAGE_RATIO": "liquidity_coverage_ratio", "RISK_COVERAGE": "risk_coverage_ratio", "NET_CAPITAL_LIABILITIES": "net_capital_liabilities_ratio", "NET_ASSETS_LIABILITIES": "net_assets_liabilities_ratio", "SOLVENCY_AR": "solvency_adequacy_ratio", }) if frame.empty or "secucode" not in frame.columns: # The upstream data centre has no rows for this report period yet. Without this # guard, `pd.DataFrame([])` is a column-less frame and the merge below dies with # a cryptic `KeyError: 'secucode'` deep inside pandas. # # Policy: keep the previous period's table AS-IS (its report_period column still # says the older date, so nothing is relabelled or faked) and continue the run # with an explicit warning. Failing hard here would freeze the whole dashboard # for weeks, because M1 gates M2 and the dashboard build. if output_path.exists(): cached = pd.read_parquet(output_path) if not cached.empty and "report_period" in cached.columns: cached_period = str(pd.Timestamp(cached["report_period"].max()).date()) warning = ( f"金融行业指标（银行/证券/保险）上游尚未发布 {report_period} 期，" f"本次沿用 {cached_period} 期结果（该表 report_period 仍为 {cached_period}，" f"未改标期数）。受影响的是金融行业个股的行业专属指标，" f"其余个股使用最新期数据。上游发布后重跑即会自动更新。" ) return FinancialSectorDataResult( rows=len(cached), report_period=cached_period, rebuilt=False, output_path=output_path, warning=warning, ) # No cached fallback either: nothing truthful can be produced, so fail. raise RuntimeError( f"金融行业指标（银行/证券/保险）取数失败：上游数据中心尚未发布 {report_period} 期的数据，" f"接口返回 0 条记录，且本地没有可沿用的历史结果。" f"该期数据通常晚于普通行业披露；请等上游发布后重跑本步骤。" ) frame = financial.merge(frame, on="secucode", how="left") frame["report_period"] = pd.to_datetime(frame["report_period"]) frame["notice_date"] = pd.to_datetime(frame["notice_date"]) numeric = [column for column in frame.columns if column not in { "source_code", "industry_l2_code", "industry_l2_name", "secucode", "report_period", "notice_date" }] frame[numeric] = frame[numeric].apply(pd.to_numeric, errors="coerce") bank = frame["industry_l2_code"].isin(["X5001", "X5002"]) security = frame["industry_l2_code"].eq("X5101") insurance = frame["industry_l2_code"].eq("X5102") frame["financial_core_margin"] = frame["net_interest_margin"].where(bank, frame["sector_roe"]) frame["financial_solvency"] = frame["capital_adequacy_ratio"].where(bank) frame.loc[security, "financial_solvency"] = frame.loc[security, "risk_coverage_ratio"] frame.loc[insurance, "financial_solvency"] = frame.loc[insurance, "solvency_adequacy_ratio"] frame["financial_buffer"] = frame["provision_coverage_ratio"].where(bank) frame.loc[security, "financial_buffer"] = frame.loc[security, "net_capital_liabilities_ratio"] frame.loc[insurance, "financial_buffer"] = frame.loc[insurance, "solvency_adequacy_ratio"] frame.to_parquet(output_path, index=False) frame.to_csv(settings.processed_dir / "financial_sector_metrics.csv", index=False, encoding="utf-8-sig") with duckdb.connect(str(settings.database_dir / "fundamental_mvp.duckdb")) as connection: connection.register("sector_df", frame) connection.execute("CREATE OR REPLACE TABLE financial_sector_metrics AS SELECT * FROM sector_df") return FinancialSectorDataResult(len(frame), report_period, True, output_path)