from __future__ import annotations from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass from datetime import date, timedelta import json from pathlib import Path import re import time import urllib.parse import urllib.request import duckdb import pandas as pd from .config import Settings GOVERNANCE_SCHEMA_VERSION = 1 EASTMONEY_DATA_API = "https://datacenter-web.eastmoney.com/api/data/v1/get" EASTMONEY_NOTICE_API = "https://np-anotice-stock.eastmoney.com/api/security/ann" RISK_PATTERNS: list[tuple[str, str, str]] = [ ("AUDIT_ADVERSE", "hard", r"无法表示意见|否定意见"), ("AUDIT_QUALIFIED", "warning", r"保留意见|非标准审计意见|非标审计意见|非无保留意见"), ("REGULATORY_MAJOR", "hard", r"立案告知书|立案调查|行政处罚决定书|公开谴责|纪律处分"), ("REGULATORY_WARNING", "warning", r"警示函|采取监管措施|责令改正|监管函"), ("FUNDS_OCCUPANCY", "hard", r"非经营性资金占用|违规资金占用|资金占用事项|违规担保"), ("MAJOR_LITIGATION_DEBT", "warning", r"重大诉讼|重大仲裁|债务逾期|申请重整|破产重整"), ("DELISTING_RISK", "hard", r"可能被终止上市|重大违法强制退市|退市风险警示"), ] RESOLUTION_PATTERN = re.compile(r"影响消除|事项消除|清偿完毕|整改完成|撤销风险警示|解除风险警示") @dataclass(frozen=True) class GovernanceBuildResult: pledge_rows: int announcement_rows: int risk_flags: int refreshed: bool output_path: Path def _get_json(url: str, params: dict[str, object], retries: int = 3) -> dict[str, object]: request_url = url + "?" + urllib.parse.urlencode(params) request = urllib.request.Request(request_url, headers={ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://data.eastmoney.com/", }) last_error: Exception | None = None for attempt in range(retries): try: with urllib.request.urlopen(request, timeout=30) as response: return json.loads(response.read()) except Exception as exc: # network errors are recorded by the caller last_error = exc time.sleep(0.5 * (attempt + 1)) raise RuntimeError(f"数据请求失败: {request_url[:180]}") from last_error def _fetch_pages(url: str, base_params: dict[str, object], total_pages: int, data_getter, workers: int = 8) -> list[dict[str, object]]: def fetch(page: int) -> list[dict[str, object]]: params = dict(base_params) params["pageNumber" if "pageNumber" in params else "page_index"] = page payload = _get_json(url, params) return data_getter(payload) rows: list[dict[str, object]] = [] with ThreadPoolExecutor(max_workers=workers) as executor: futures = {executor.submit(fetch, page): page for page in range(1, total_pages + 1)} for future in as_completed(futures): rows.extend(future.result()) return rows def _source_code(code: str) -> str | None: code = str(code).zfill(6) if code.startswith("6"): return "sh" + code if code.startswith(("0", "3")): return "sz" + code return None def _fetch_pledge() -> tuple[pd.DataFrame, str]: profile_params = { "sortColumns": "TRADE_DATE", "sortTypes": "-1", "pageSize": 1, "pageNumber": 1, "reportName": "RPT_CSDC_STATISTICS", "columns": "ALL", "source": "WEB", "client": "WEB", } profile = _get_json(EASTMONEY_DATA_API, profile_params) latest = str(profile["result"]["data"][0]["TRADE_DATE"])[:10] params = { "sortColumns": "PLEDGE_RATIO", "sortTypes": "-1", "pageSize": 500, "pageNumber": 1, "reportName": "RPT_CSDC_LIST", "columns": "ALL", "source": "WEB", "client": "WEB", "filter": f"(TRADE_DATE='{latest}')", } first = _get_json(EASTMONEY_DATA_API, params) pages = int(first["result"]["pages"]) rows = _fetch_pages(EASTMONEY_DATA_API, params, pages, lambda x: x["result"]["data"], workers=5) frame = pd.DataFrame(rows) frame["source_code"] = frame["SECURITY_CODE"].map(_source_code) selected = frame.rename(columns={ "SECURITY_CODE": "code", "SECURITY_NAME_ABBR": "source_name", "TRADE_DATE": "pledge_date", "PLEDGE_RATIO": "pledge_ratio_pct", "REPURCHASE_BALANCE": "pledged_shares_10k", "PLEDGE_DEAL_NUM": "pledge_deal_count", "PLEDGE_MARKET_CAP": "pledge_market_value_10k", })[["source_code", "code", "source_name", "pledge_date", "pledge_ratio_pct", "pledged_shares_10k", "pledge_deal_count", "pledge_market_value_10k"]] selected["pledge_date"] = pd.to_datetime(selected["pledge_date"]) return selected.dropna(subset=["source_code"]).drop_duplicates("source_code"), latest def _extract_notice_row(item: dict[str, object]) -> dict[str, object] | None: codes = item.get("codes") or [] selected = next((entry for entry in codes if str(entry.get("ann_type", "")).startswith("A")), None) if not selected: return None source_code = _source_code(str(selected.get("stock_code", ""))) if source_code is None: return None return { "source_code": source_code, "code": selected.get("stock_code"), "source_name": selected.get("short_name"), "announcement_id": item.get("art_code"), "announcement_date": item.get("notice_date"), "title": item.get("title"), "url": f"https://data.eastmoney.com/notices/detail/{selected.get('stock_code')}/{item.get('art_code')}.html", } def _fetch_risk_notices(start: date, end: date) -> pd.DataFrame: params = { "sr": -1, "page_size": 100, "page_index": 1, "ann_type": "A", "client_source": "web", "f_node": 3, "s_node": 0, "begin_time": start.isoformat(), "end_time": end.isoformat(), } first = _get_json(EASTMONEY_NOTICE_API, params) total_hits = int(first["data"]["total_hits"]) total_pages = (total_hits + 99) // 100 items = _fetch_pages(EASTMONEY_NOTICE_API, params, total_pages, lambda x: x["data"]["list"], workers=8) rows = [row for item in items if (row := _extract_notice_row(item)) is not None] frame = pd.DataFrame(rows).drop_duplicates("announcement_id") frame["announcement_date"] = pd.to_datetime(frame["announcement_date"]) return frame.sort_values(["announcement_date", "announcement_id"], ascending=[False, False]) def _classify_notices(notices: pd.DataFrame) -> pd.DataFrame: rows: list[dict[str, object]] = [] for row in notices.itertuples(index=False): title = str(row.title) if RESOLUTION_PATTERN.search(title): continue for risk_type, severity, pattern in RISK_PATTERNS: if re.search(pattern, title): rows.append({ "security_id": ("SH." if row.source_code.startswith("sh") else "SZ.") + row.source_code[2:], "source_code": row.source_code, "rule_code": risk_type, "severity": severity, "description": title, "event_date": row.announcement_date, "announcement_id": row.announcement_id, "source_url": row.url, }) return pd.DataFrame(rows, columns=[ "security_id", "source_code", "rule_code", "severity", "description", "event_date", "announcement_id", "source_url", ]).drop_duplicates(["source_code", "rule_code", "announcement_id"]) def refresh_governance_data(settings: Settings, force: bool = False, cache_days: int = 1) -> GovernanceBuildResult: settings.processed_dir.mkdir(parents=True, exist_ok=True) settings.report_dir.mkdir(parents=True, exist_ok=True) pledge_path = settings.processed_dir / "external_pledge_latest.parquet" notices_path = settings.processed_dir / "external_risk_notices.parquet" flags_path = settings.processed_dir / "external_governance_risk_flags.csv" metadata_path = settings.report_dir / "governance_metadata.json" fresh = metadata_path.exists() and (time.time() - metadata_path.stat().st_mtime) < cache_days * 86400 if fresh and not force and pledge_path.exists() and notices_path.exists() and flags_path.exists(): pledge = pd.read_parquet(pledge_path) notices = pd.read_parquet(notices_path) flags = pd.read_csv(flags_path) return GovernanceBuildResult(len(pledge), len(notices), len(flags), False, flags_path) started = time.time() status: dict[str, object] = {"schema_version": GOVERNANCE_SCHEMA_VERSION, "build_date": date.today().isoformat()} pledge, pledge_date = _fetch_pledge() start = date.today() - timedelta(days=730) notices = _fetch_risk_notices(start, date.today()) flags = _classify_notices(notices) pledge.to_parquet(pledge_path, index=False) notices.to_parquet(notices_path, index=False) flags.to_csv(flags_path, index=False, encoding="utf-8-sig") status.update({ "pledge_date": pledge_date, "pledge_rows": len(pledge), "notice_start": start.isoformat(), "notice_end": date.today().isoformat(), "risk_notice_rows": len(notices), "classified_risk_flags": len(flags), "source": "Eastmoney public data APIs", "elapsed_seconds": round(time.time() - started, 2), "limitations": [ "公告风险依据标题分类，需对高风险样本阅读原文复核。", "只拉取近两年风险提示类公告，不能代替交易所和证监会完整监管档案。", "未出现风险标题不等于已确认标准无保留审计意见。", ], }) metadata_path.write_text(json.dumps(status, ensure_ascii=False, indent=2), encoding="utf-8") _persist_governance(settings, pledge, notices, flags) _write_report(settings, status, flags, pledge) return GovernanceBuildResult(len(pledge), len(notices), len(flags), True, flags_path) def _persist_governance(settings: Settings, pledge: pd.DataFrame, notices: pd.DataFrame, flags: pd.DataFrame) -> None: with duckdb.connect(str(settings.database_dir / "fundamental_mvp.duckdb")) as connection: connection.register("pledge_df", pledge) connection.register("notices_df", notices) connection.register("flags_df", flags) connection.execute("CREATE OR REPLACE TABLE external_pledge_latest AS SELECT * FROM pledge_df") connection.execute("CREATE OR REPLACE TABLE external_risk_notices AS SELECT * FROM notices_df") connection.execute("CREATE OR REPLACE TABLE external_governance_risk_flags AS SELECT * FROM flags_df") def _write_report(settings: Settings, metadata: dict[str, object], flags: pd.DataFrame, pledge: pd.DataFrame) -> None: counts = flags.groupby(["rule_code", "severity"]).size().reset_index(name="count") lines = "\n".join(f"| {r.rule_code} | {r.severity} | {r.count} |" for r in counts.itertuples(index=False)) report = f"""# M1 治理与外部风险数据报告 生成日期：{metadata['build_date']} 质押数据日期：{metadata['pledge_date']} 有质押记录公司：{len(pledge)} 家 近两年风险提示公告：{metadata['risk_notice_rows']} 条 分类风险标记：{len(flags)} 条 | 风险类型 | 级别 | 记录数 | |---|---|---:| {lines} ## 口径限制 - 公告分类依据公告标题，硬风险样本必须点击保存的原文链接复核。 - 无风险标记只表示近两年未匹配已建规则，不表示公司已被证明没有治理风险。 - 质押数据是当前快照，后续回测必须保存每期历史快照。 """ (settings.report_dir / "M1_治理与外部风险报告.md").write_text(report, encoding="utf-8")