from __future__ import annotations
|
|
import html
|
import json
|
import re
|
import urllib.parse
|
from datetime import date, datetime, time as dt_time, timedelta, timezone
|
from typing import Any
|
from zoneinfo import ZoneInfo
|
|
from .cache import BlobIntegrityError
|
from .http_client import HttpClient, HttpRequest
|
|
|
def _decode_json(response: dict[str, Any]) -> dict[str, Any]:
|
return json.loads(response["body"].decode("utf-8-sig"))
|
|
|
def _secid(ticker: str) -> str:
|
code, market = ticker.split(".")
|
return f"{1 if market == 'SH' else 0}.{code}"
|
|
|
def _market_name(ticker: str) -> str:
|
return {"SZ": "深圳证券交易所", "SH": "上海证券交易所", "BJ": "北京证券交易所"}[ticker[-2:]]
|
|
|
def _result(provider_id: str, as_of: str, status: str = "OK") -> dict[str, Any]:
|
return {
|
"provider_id": provider_id,
|
"adapter_version": "1.0.0",
|
"status": status,
|
"fetched_at": datetime.now().astimezone().isoformat(),
|
"as_of_date": as_of,
|
"records": [],
|
"sources": [],
|
"gaps": [],
|
"warnings": [],
|
"raw_artifact_hashes": [],
|
"request_telemetry": [],
|
"field_lineage": {},
|
"data_kinds": {},
|
"baseline_data_kinds": [],
|
"cache_integrity_failure": False,
|
}
|
|
|
def _capture(result: dict[str, Any], response: dict[str, Any]) -> None:
|
meta = response["meta"]
|
result["raw_artifact_hashes"].append(meta["blob_hash"])
|
if response.get("from_baseline"):
|
result["baseline_data_kinds"].append(meta["data_kind"])
|
return
|
result["request_telemetry"].append(
|
{
|
"fingerprint": response["fingerprint"],
|
"raw_hash": meta["blob_hash"],
|
"bytes": meta["bytes"],
|
"http_status": meta["http_status"],
|
"from_cache": response["from_cache"],
|
"started_at": meta.get("started_at"),
|
"finished_at": meta.get("finished_at"),
|
"data_kind": meta.get("data_kind"),
|
"remaining_before": meta.get("remaining_before"),
|
"remaining_after": meta.get("remaining_after"),
|
"attempts": meta.get("attempts", []),
|
}
|
)
|
|
|
def _kind_entry(
|
*,
|
request: HttpRequest,
|
response: dict[str, Any],
|
ticker: str,
|
as_of: str,
|
data_date: str,
|
publish_date: str | None,
|
watermark: Any,
|
requires_publish_date: bool = False,
|
**extra: Any,
|
) -> dict[str, Any]:
|
meta = response["meta"]
|
return {
|
"data_kind": request.data_kind,
|
"provider_id": request.provider_id,
|
"request_fingerprint": response["fingerprint"],
|
"ticker": ticker,
|
"requested_as_of": as_of,
|
"baseline_as_of": as_of,
|
"data_date": data_date,
|
"publish_date": publish_date,
|
"requires_publish_date": requires_publish_date,
|
"fetched_at": meta["fetched_at"],
|
"expires_at": meta["expires_at"],
|
"adapter_version": request.adapter_version,
|
"raw_hash": meta["blob_hash"],
|
"schema_status": "PASS",
|
"as_of_status": "PASS",
|
"semantic_status": "OK",
|
"watermark": watermark,
|
**extra,
|
}
|
|
|
def _lineage(
|
source_id: str,
|
raw_hash: str,
|
publish_date: str,
|
data_date: str,
|
provider: str,
|
raw_field: str,
|
unit: str,
|
) -> dict[str, Any]:
|
return {
|
"source_id": source_id,
|
"raw_hash": raw_hash,
|
"publish_date": publish_date,
|
"data_date": data_date,
|
"provider": provider,
|
"raw_field": raw_field,
|
"unit": unit,
|
}
|
|
|
def acquire_announcements(client: HttpClient, ticker: str, as_of: str) -> dict[str, Any]:
|
result = _result("cninfo.announcement_index", as_of)
|
code, market = ticker.split(".")
|
if market == "BJ":
|
raise ValueError("CNInfo 首期 registry 不支持 BJ 公告索引")
|
recent_as_of = (client.process_start.date() - date.fromisoformat(as_of)).days <= 7
|
stock_request = HttpRequest(
|
"cninfo.announcement_index",
|
"1.0.0",
|
"stock_identity",
|
"GET",
|
"https://www.cninfo.com.cn/new/data/szse_stock.json",
|
ticker,
|
as_of,
|
"cninfo_stock_list",
|
ttl_seconds=7 * 86400,
|
)
|
stock = client.fetch(stock_request)
|
_capture(result, stock)
|
stock_payload = _decode_json(stock)
|
rows = stock_payload.get("stockList") or stock_payload.get("data") or []
|
identity = next(
|
(
|
row
|
for row in rows
|
if str(row.get("code") or row.get("secCode") or row.get("dm")) == code
|
),
|
None,
|
)
|
if not identity:
|
raise ValueError(f"CNInfo 无证券记录：{ticker}")
|
client.confirm_reusable(stock_request, stock)
|
org_id = str(identity.get("orgId") or identity.get("orgid") or identity.get("id"))
|
company = str(identity.get("zwjc") or identity.get("name") or identity.get("secName"))
|
plate = "sz" if market == "SZ" else "sh"
|
form = {
|
"pageNum": "1",
|
"pageSize": "30",
|
"column": "szse" if market == "SZ" else "sse",
|
"tabName": "fulltext",
|
"plate": plate,
|
"stock": f"{code},{org_id}",
|
"searchkey": "",
|
"secid": "",
|
"category": "category_ndbg_szsh;category_yjdbg_szsh;category_bndbg_szsh;category_sjdbg_szsh",
|
"trade": "",
|
"seDate": f"2020-01-01~{as_of}",
|
"sortName": "",
|
"sortType": "",
|
"isHLtitle": "true",
|
}
|
body = urllib.parse.urlencode(form).encode("utf-8")
|
announcement_request = HttpRequest(
|
"cninfo.announcement_index",
|
"1.0.0",
|
"announcement_index",
|
"POST",
|
"https://www.cninfo.com.cn/new/hisAnnouncement/query",
|
ticker,
|
as_of,
|
"cninfo_announcements",
|
body=body,
|
content_type="application/x-www-form-urlencoded",
|
ttl_seconds=6 * 3600 if recent_as_of else 30 * 86400,
|
)
|
announcements = client.fetch(announcement_request)
|
_capture(result, announcements)
|
payload = _decode_json(announcements)
|
result["identity"] = {
|
"ticker": ticker,
|
"company": company,
|
"market": _market_name(ticker),
|
"org_id": org_id,
|
}
|
fixture_sources = payload.get("v2_sources")
|
if fixture_sources is not None:
|
if not fixture_sources:
|
raise ValueError("CNInfo fixture A1 索引为空")
|
for source in fixture_sources:
|
if not source.get("publish_date") or source["publish_date"] > as_of:
|
raise ValueError("E_ASOF_VIOLATION：CNInfo fixture 发布日缺失或未来")
|
result["sources"] = fixture_sources
|
result["records"] = payload.get("announcements", [])
|
client.confirm_reusable(announcement_request, announcements)
|
max_publish = max(source["publish_date"] for source in fixture_sources)
|
result["data_kinds"] = {
|
"stock_identity": _kind_entry(
|
request=stock_request,
|
response=stock,
|
ticker=ticker,
|
as_of=as_of,
|
data_date=as_of,
|
publish_date=None,
|
watermark={"code": code, "market": market, "org_id": org_id},
|
),
|
"announcement_index": _kind_entry(
|
request=announcement_request,
|
response=announcements,
|
ticker=ticker,
|
as_of=as_of,
|
data_date=max_publish,
|
publish_date=max_publish,
|
watermark=max_publish,
|
requires_publish_date=True,
|
),
|
}
|
return result
|
cutoff = datetime.combine(
|
date.fromisoformat(as_of), dt_time.max, tzinfo=ZoneInfo("Asia/Shanghai")
|
)
|
for row in payload.get("announcements", []):
|
raw_time = row.get("announcementTime")
|
if isinstance(raw_time, (int, float)):
|
published = datetime.fromtimestamp(raw_time / 1000, timezone.utc).astimezone(
|
ZoneInfo("Asia/Shanghai")
|
)
|
else:
|
published = datetime.fromisoformat(str(raw_time).replace("Z", "+00:00"))
|
if published.tzinfo is None:
|
published = published.replace(tzinfo=ZoneInfo("Asia/Shanghai"))
|
else:
|
published = published.astimezone(ZoneInfo("Asia/Shanghai"))
|
if published > cutoff:
|
continue
|
url = "https://static.cninfo.com.cn/" + str(row.get("adjunctUrl", "")).lstrip("/")
|
source_id = f"CNINFO-{row.get('announcementId')}"
|
record = {
|
"source_id": source_id,
|
"announcement_id": str(row.get("announcementId")),
|
"title": html.unescape(re.sub("<[^>]+>", "", str(row.get("announcementTitle", "")))),
|
"publish_date": published.date().isoformat(),
|
"url": url,
|
}
|
result["records"].append(record)
|
period_end = None
|
supports = ["statutory_disclosure"]
|
normalized_title = record["title"].replace(" ", "")
|
annual_match = re.search(r"(\d{4})年年度报告", normalized_title)
|
q1_match = re.search(r"(\d{4})年(?:第一|一)季度报告", normalized_title)
|
if annual_match:
|
period_end = f"{annual_match.group(1)}-12-31"
|
supports = ["financials.annual", "financials.prior_year_same_period"]
|
elif q1_match:
|
period_end = f"{q1_match.group(1)}-03-31"
|
supports = [
|
"financials.current_cumulative",
|
"financials.prior_year_same_period_comparative",
|
"balance_sheet",
|
]
|
result["sources"].append(
|
{
|
"id": source_id,
|
"source_type": "cninfo",
|
"title": record["title"],
|
"publish_date": record["publish_date"],
|
"period_end": period_end,
|
"url": url,
|
"supports": supports,
|
"revision_status": "current",
|
}
|
)
|
if not result["records"]:
|
raise ValueError("CNInfo 公告索引为空")
|
client.confirm_reusable(announcement_request, announcements)
|
max_publish = max(source["publish_date"] for source in result["sources"])
|
result["data_kinds"] = {
|
"stock_identity": _kind_entry(
|
request=stock_request,
|
response=stock,
|
ticker=ticker,
|
as_of=as_of,
|
data_date=as_of,
|
publish_date=None,
|
watermark={"code": code, "market": market, "org_id": org_id},
|
),
|
"announcement_index": _kind_entry(
|
request=announcement_request,
|
response=announcements,
|
ticker=ticker,
|
as_of=as_of,
|
data_date=max_publish,
|
publish_date=max_publish,
|
watermark=max_publish,
|
requires_publish_date=True,
|
),
|
}
|
return result
|
|
|
def acquire_market(client: HttpClient, ticker: str, as_of: str) -> dict[str, Any]:
|
result = _result("eastmoney.market", as_of)
|
quote_request = HttpRequest(
|
"eastmoney.market",
|
"1.0.0",
|
"shares_market_cap",
|
"GET",
|
"https://push2.eastmoney.com/api/qt/stock/get",
|
ticker,
|
as_of,
|
"eastmoney_market_quote",
|
query={"secid": _secid(ticker), "fields": "f57,f58,f84,f116,f124"},
|
ttl_seconds=900,
|
)
|
quote = client.fetch(quote_request)
|
kline_request = HttpRequest(
|
"eastmoney.market",
|
"1.0.0",
|
"market_close",
|
"GET",
|
"https://push2his.eastmoney.com/api/qt/stock/kline/get",
|
ticker,
|
as_of,
|
"eastmoney_market_kline",
|
query={
|
"secid": _secid(ticker),
|
"klt": "101",
|
"fqt": "1",
|
"beg": (date.fromisoformat(as_of) - timedelta(days=14)).strftime("%Y%m%d"),
|
"end": as_of.replace("-", ""),
|
"fields1": "f1,f2,f3,f4,f5,f6",
|
"fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61",
|
},
|
ttl_seconds=30 * 86400,
|
)
|
kline = client.fetch(kline_request)
|
_capture(result, quote)
|
_capture(result, kline)
|
qdata = _decode_json(quote).get("data") or {}
|
kdata = _decode_json(kline).get("data") or {}
|
klines = kdata.get("klines") or []
|
if not klines:
|
raise ValueError("无 as-of 历史收盘")
|
eligible = [item for item in klines if str(item).split(",", 1)[0] <= as_of]
|
if not eligible:
|
raise ValueError("无不晚于 as-of 的历史收盘")
|
fields = str(max(eligible, key=lambda item: str(item).split(",", 1)[0])).split(",")
|
trade_date, close = fields[0], float(fields[2])
|
quote_epoch = int(qdata.get("f124") or 0)
|
if quote_epoch <= 0:
|
raise ValueError("E_HISTORICAL_SHARES_UNPROVEN：quote f124 缺失或为 0")
|
quote_time = datetime.fromtimestamp(
|
quote_epoch, ZoneInfo("Asia/Shanghai")
|
)
|
quote_date = quote_time.date().isoformat()
|
process_local_date = client.process_start.astimezone(ZoneInfo("Asia/Shanghai")).date().isoformat()
|
if quote_date != trade_date:
|
raise ValueError("E_HISTORICAL_SHARES_UNPROVEN：股本时间与收盘交易日不一致")
|
if client.fixture_dir is None and as_of != process_local_date:
|
raise ValueError("E_HISTORICAL_SHARES_UNPROVEN：历史 as-of 无同日可信 baseline")
|
shares = int(qdata["f84"])
|
market_cap = float(qdata["f116"])
|
if shares <= 0 or close <= 0 or market_cap <= 0:
|
raise ValueError("行情核心字段非正数")
|
if abs(close * shares - market_cap) > max(1.0, market_cap * 0.005):
|
raise ValueError("price×shares 与平台市值不一致")
|
client.confirm_reusable(quote_request, quote)
|
client.confirm_reusable(kline_request, kline)
|
result["market"] = {
|
"price": close,
|
"price_type": "收盘价",
|
"price_timestamp": f"{trade_date}T15:00:00+08:00",
|
"diluted_shares": shares,
|
"shares_date": quote_date,
|
"platform_market_cap": market_cap,
|
}
|
price_source_id = f"SRC-MARKET-CLOSE-{trade_date.replace('-', '')}"
|
shares_source_id = f"SRC-SHARES-{quote_date.replace('-', '')}"
|
result["sources"] = [
|
{
|
"id": price_source_id,
|
"source_type": "quote_provider",
|
"title": "东方财富历史行情接口",
|
"publish_date": trade_date,
|
"period_end": trade_date,
|
"url": f"https://push2his.eastmoney.com/api/qt/stock/kline/get?secid={_secid(ticker)}",
|
"supports": ["market.price"],
|
"revision_status": "current",
|
},
|
{
|
"id": shares_source_id,
|
"source_type": "quote_provider",
|
"title": "东方财富股本与总市值接口",
|
"publish_date": quote_date,
|
"period_end": quote_date,
|
"url": f"https://push2.eastmoney.com/api/qt/stock/get?secid={_secid(ticker)}",
|
"supports": ["market.diluted_shares", "market.platform_market_cap"],
|
"revision_status": "current",
|
},
|
]
|
result["field_lineage"] = {
|
"market.price": _lineage(
|
price_source_id, kline["meta"]["blob_hash"], trade_date, trade_date,
|
"eastmoney.market", "data.klines[].f53", "CNY/share",
|
),
|
"market.diluted_shares": _lineage(
|
shares_source_id, quote["meta"]["blob_hash"], quote_date, quote_date,
|
"eastmoney.market", "data.f84", "share",
|
),
|
"market.platform_market_cap": _lineage(
|
shares_source_id, quote["meta"]["blob_hash"], quote_date, quote_date,
|
"eastmoney.market", "data.f116", "CNY",
|
),
|
}
|
result["data_kinds"] = {
|
"market_close": _kind_entry(
|
request=kline_request, response=kline, ticker=ticker, as_of=as_of,
|
data_date=trade_date, publish_date=trade_date, watermark=trade_date,
|
requires_publish_date=True, is_latest_eligible_trade_date=True,
|
),
|
"shares_market_cap": _kind_entry(
|
request=quote_request, response=quote, ticker=ticker, as_of=as_of,
|
data_date=quote_date, publish_date=quote_date, watermark=quote_time.isoformat(),
|
requires_publish_date=True,
|
historical_capture_valid=(client.fixture_dir is not None or as_of == process_local_date),
|
),
|
}
|
return result
|
|
|
FINANCE_REPORTS = {
|
"main": "RPT_F10_FINANCE_MAINFINADATA",
|
"income": "RPT_DMSK_FN_INCOME",
|
"balance": "RPT_F10_FINANCE_GBALANCE",
|
"cashflow": "RPT_DMSK_FN_CASHFLOW",
|
}
|
|
|
def _finance_query(ticker: str, report_name: str) -> dict[str, str]:
|
return {
|
"reportName": report_name,
|
"columns": "ALL",
|
"filter": f'(SECUCODE="{ticker}")',
|
"pageNumber": "1",
|
"pageSize": "20",
|
"sortTypes": "-1",
|
"sortColumns": "REPORT_DATE",
|
}
|
|
|
def _records(payload: dict[str, Any]) -> list[dict[str, Any]]:
|
return list(((payload.get("result") or {}).get("data") or payload.get("data") or []))
|
|
|
def _pick(rows: list[dict[str, Any]], period: str, as_of: str) -> dict[str, Any]:
|
candidates = [
|
row
|
for row in rows
|
if str(row.get("REPORT_DATE", ""))[:10] == period
|
and bool(row.get("NOTICE_DATE") or row.get("UPDATE_DATE"))
|
and str(row.get("NOTICE_DATE") or row.get("UPDATE_DATE"))[:10] <= as_of
|
]
|
if not candidates:
|
raise ValueError(f"缺少财务期间 {period}")
|
return max(candidates, key=lambda row: str(row.get("NOTICE_DATE") or ""))
|
|
|
def _select_ttm_periods(rows: list[dict[str, Any]], as_of: str) -> tuple[str, str, str]:
|
available = sorted(
|
{
|
str(row.get("REPORT_DATE", ""))[:10]
|
for row in rows
|
if str(row.get("REPORT_DATE", ""))[:10] <= as_of
|
and bool(row.get("NOTICE_DATE") or row.get("UPDATE_DATE"))
|
and str(row.get("NOTICE_DATE") or row.get("UPDATE_DATE"))[:10] <= as_of
|
}
|
)
|
annuals = [period for period in available if period.endswith("-12-31")]
|
if not annuals:
|
raise ValueError("缺少 as-of 前完整年度财务")
|
annual = max(annuals)
|
cumulative = [period for period in available if period > annual and not period.endswith("-12-31")]
|
if not cumulative:
|
raise ValueError("缺少完整年度之后的最新累计期")
|
current = max(cumulative)
|
current_date = date.fromisoformat(current)
|
prior = current_date.replace(year=current_date.year - 1).isoformat()
|
if prior not in available:
|
raise ValueError(f"缺少上年同期 {prior}")
|
return annual, current, prior
|
|
|
def _number(row: dict[str, Any], *keys: str) -> float:
|
for key in keys:
|
value = row.get(key)
|
if value is not None and value != "":
|
return float(value)
|
raise ValueError(f"缺少字段 {'/'.join(keys)}")
|
|
|
LIQUID_FV_ALIAS_KEYS = (
|
"TRADE_FINASSET_NOTFVTPL",
|
"TRADE_FINASSET",
|
"FVTPL_FINASSET",
|
"APPOINT_FVTPL_FINASSET",
|
"AVAILABLE_SALE_FINASSET",
|
)
|
FINANCIAL_SINGLE_KEYS = ("DERIVE_FINASSET", "BUY_RESALE_FINASSET")
|
DEBT_KEYS = (
|
"SHORT_LOAN",
|
"NONCURRENT_LIAB_1YEAR",
|
"LONG_LOAN",
|
"BOND_PAYABLE",
|
"LEASE_LIAB",
|
"SHORT_BOND_PAYABLE",
|
)
|
|
|
def _require_keys(row: dict[str, Any], keys: tuple[str, ...]) -> None:
|
missing = [key for key in keys if key not in row]
|
if missing:
|
raise ValueError(f"E_BALANCE_SCHEMA_DRIFT：缺少键 {','.join(missing)}")
|
|
|
def _nonnegative(value: Any, key: str) -> float:
|
if value is None or value == "":
|
return 0.0
|
number = float(value)
|
if number < 0:
|
raise ValueError(f"E_BALANCE_SCHEMA_DRIFT：{key} 为负")
|
return number
|
|
|
def parse_balance_record(row: dict[str, Any]) -> dict[str, float]:
|
_require_keys(row, LIQUID_FV_ALIAS_KEYS + FINANCIAL_SINGLE_KEYS + DEBT_KEYS + ("MINORITY_EQUITY",))
|
alias_values = [
|
_nonnegative(row[key], key)
|
for key in LIQUID_FV_ALIAS_KEYS
|
if row[key] is not None and row[key] != ""
|
]
|
distinct = set(alias_values)
|
if len(distinct) > 1:
|
raise ValueError("E_BALANCE_ALIAS_CONFLICT：流动公允价值金融资产 alias 数值冲突")
|
liquid_fv = alias_values[0] if alias_values else 0.0
|
financial_assets = liquid_fv + sum(
|
_nonnegative(row[key], key) for key in FINANCIAL_SINGLE_KEYS
|
)
|
debt = sum(_nonnegative(row[key], key) for key in DEBT_KEYS)
|
minority = _nonnegative(row["MINORITY_EQUITY"], "MINORITY_EQUITY")
|
return {
|
"non_operating_financial_assets": financial_assets,
|
"interest_bearing_debt": debt,
|
"minority_interest": minority,
|
}
|
|
|
def acquire_finance(client: HttpClient, ticker: str, as_of: str) -> dict[str, Any]:
|
result = _result("eastmoney.finance", as_of)
|
recent_as_of = (client.process_start.date() - date.fromisoformat(as_of)).days <= 7
|
payloads: dict[str, dict[str, Any]] = {}
|
responses: dict[str, tuple[HttpRequest, dict[str, Any]]] = {}
|
for kind, report in FINANCE_REPORTS.items():
|
request = HttpRequest(
|
"eastmoney.finance",
|
"1.0.0",
|
f"finance_{kind}",
|
"GET",
|
"https://datacenter-web.eastmoney.com/api/data/v1/get",
|
ticker,
|
as_of,
|
f"eastmoney_finance_{kind}",
|
query=_finance_query(ticker, report),
|
ttl_seconds=6 * 3600 if recent_as_of else 30 * 86400,
|
)
|
response = client.fetch(request)
|
_capture(result, response)
|
payloads[kind] = _decode_json(response)
|
responses[kind] = (request, response)
|
income = _records(payloads["income"])
|
cashflow = _records(payloads["cashflow"])
|
balance = _records(payloads["balance"])
|
main = _records(payloads["main"])
|
periods = _select_ttm_periods(income or main, as_of)
|
labels = ("annual", "current_cumulative", "prior_year_same_period")
|
financials: dict[str, Any] = {}
|
selected: dict[str, dict[str, dict[str, Any]]] = {}
|
for period, label in zip(periods, labels):
|
inc = _pick(income or main, period, as_of)
|
cash = _pick(cashflow, period, as_of)
|
selected[label] = {"income": inc, "cashflow": cash}
|
financials[label] = {
|
"period_end": period,
|
"basis": {
|
"annual": "audited",
|
"current_cumulative": "quarterly_report_unaudited",
|
"prior_year_same_period": "reported_comparative",
|
}[label],
|
"revenue": _number(inc, "TOTAL_OPERATE_INCOME", "TOTALOPERATEREVE"),
|
"attributable_profit": _number(inc, "PARENT_NETPROFIT", "PARENTNETPROFIT"),
|
"deduct_profit": _number(inc, "DEDUCT_PARENT_NETPROFIT", "KCFJCXSYJLR"),
|
"cfo": _number(cash, "NETCASH_OPERATE"),
|
"capex": _number(cash, "CONSTRUCT_LONG_ASSET"),
|
}
|
bal = _pick(balance, periods[1], as_of)
|
parsed_balance = parse_balance_record(bal)
|
result["financials"] = financials
|
result["balance_sheet"] = {
|
"period_end": periods[1],
|
"equity": _number(bal, "TOTAL_EQUITY", "TOTAL_EQUITY_PARENT"),
|
"cash_available": _number(bal, "MONETARYFUNDS"),
|
**parsed_balance,
|
}
|
selected["balance_sheet"] = {"balance": bal}
|
raw_by_kind = {
|
kind: response["meta"]["blob_hash"]
|
for kind, (_, response) in responses.items()
|
}
|
source_ids: dict[tuple[str, str], str] = {}
|
for label, tables in selected.items():
|
for table, row in tables.items():
|
period = str(row["REPORT_DATE"])[:10]
|
publish = str(row.get("NOTICE_DATE") or row.get("UPDATE_DATE"))[:10]
|
source_id = f"SRC-EM-FINANCE-{table.upper()}-{period.replace('-', '')}"
|
source_ids[(label, table)] = source_id
|
result["sources"].append(
|
{
|
"id": source_id,
|
"source_type": "financial_mirror",
|
"title": f"东方财富结构化财务 {table} {period}",
|
"publish_date": publish,
|
"period_end": period,
|
"url": "https://datacenter-web.eastmoney.com/api/data/v1/get",
|
"supports": [f"financials.{label}" if label != "balance_sheet" else "balance_sheet"],
|
"revision_status": "current",
|
}
|
)
|
for label in labels:
|
period = financials[label]["period_end"]
|
inc = selected[label]["income"]
|
cash = selected[label]["cashflow"]
|
inc_publish = str(inc.get("NOTICE_DATE") or inc.get("UPDATE_DATE"))[:10]
|
cash_publish = str(cash.get("NOTICE_DATE") or cash.get("UPDATE_DATE"))[:10]
|
for field, raw_field, unit in (
|
("revenue", "TOTAL_OPERATE_INCOME|TOTALOPERATEREVE", "CNY"),
|
("attributable_profit", "PARENT_NETPROFIT|PARENTNETPROFIT", "CNY"),
|
("deduct_profit", "DEDUCT_PARENT_NETPROFIT|KCFJCXSYJLR", "CNY"),
|
):
|
result["field_lineage"][f"financials.{label}.{field}"] = _lineage(
|
source_ids[(label, "income")], raw_by_kind["income"], inc_publish,
|
period, "eastmoney.finance", raw_field, unit,
|
)
|
for field, raw_field in (("cfo", "NETCASH_OPERATE"), ("capex", "CONSTRUCT_LONG_ASSET")):
|
result["field_lineage"][f"financials.{label}.{field}"] = _lineage(
|
source_ids[(label, "cashflow")], raw_by_kind["cashflow"], cash_publish,
|
period, "eastmoney.finance", raw_field, "CNY",
|
)
|
if label == "prior_year_same_period":
|
current_period = financials["current_cumulative"]["period_end"]
|
current_publish = str(
|
selected["current_cumulative"]["income"].get("NOTICE_DATE")
|
or selected["current_cumulative"]["income"].get("UPDATE_DATE")
|
)[:10]
|
if inc_publish != current_publish or cash_publish != current_publish:
|
raise ValueError(
|
"E_COMPARATIVE_LINEAGE：上年同期行未与本期报告共享发布日期"
|
)
|
relation = {
|
"type": "same_response_comparative_row",
|
"current_period_end": current_period,
|
"comparison_period_end": period,
|
"shared_publish_date": current_publish,
|
}
|
for field in ("revenue", "attributable_profit", "deduct_profit", "cfo", "capex"):
|
result["field_lineage"][f"financials.{label}.{field}"][
|
"comparison_relation"
|
] = relation
|
bal_publish = str(bal.get("NOTICE_DATE") or bal.get("UPDATE_DATE"))[:10]
|
balance_fields = {
|
"equity": "TOTAL_EQUITY|TOTAL_EQUITY_PARENT",
|
"cash_available": "MONETARYFUNDS",
|
"non_operating_financial_assets": "+".join(LIQUID_FV_ALIAS_KEYS + FINANCIAL_SINGLE_KEYS),
|
"interest_bearing_debt": "+".join(DEBT_KEYS),
|
"minority_interest": "MINORITY_EQUITY",
|
}
|
for field, raw_field in balance_fields.items():
|
result["field_lineage"][f"balance_sheet.{field}"] = _lineage(
|
source_ids[("balance_sheet", "balance")], raw_by_kind["balance"], bal_publish,
|
periods[1], "eastmoney.finance", raw_field, "CNY",
|
)
|
ttl_suffix = "recent" if recent_as_of else "historical"
|
required_periods = set(periods)
|
for kind, rows_for_kind in (("main", main), ("income", income), ("balance", balance), ("cashflow", cashflow)):
|
request, response = responses[kind]
|
eligible_rows = [
|
row for row in rows_for_kind
|
if row.get("REPORT_DATE") and (row.get("NOTICE_DATE") or row.get("UPDATE_DATE"))
|
and str(row["REPORT_DATE"])[:10] <= as_of
|
and str(row.get("NOTICE_DATE") or row.get("UPDATE_DATE"))[:10] <= as_of
|
]
|
max_row = max(
|
eligible_rows,
|
key=lambda row: (
|
str(row.get("NOTICE_DATE") or row.get("UPDATE_DATE"))[:10],
|
str(row["REPORT_DATE"])[:10],
|
),
|
)
|
present_periods = {str(row["REPORT_DATE"])[:10] for row in eligible_rows}
|
complete = (
|
bool(present_periods & {periods[0]}) if kind == "main"
|
else required_periods.issubset(present_periods) if kind in {"income", "cashflow"}
|
else periods[1] in present_periods
|
)
|
entry = _kind_entry(
|
request=request,
|
response=response,
|
ticker=ticker,
|
as_of=as_of,
|
data_date=str(max_row["REPORT_DATE"])[:10],
|
publish_date=str(max_row.get("NOTICE_DATE") or max_row.get("UPDATE_DATE"))[:10],
|
watermark=[
|
str(max_row.get("NOTICE_DATE") or max_row.get("UPDATE_DATE"))[:10],
|
str(max_row["REPORT_DATE"])[:10],
|
],
|
requires_publish_date=True,
|
required_periods_complete=complete,
|
ttl_class=f"finance_{kind}_{ttl_suffix}",
|
)
|
result["data_kinds"][f"finance_{kind}"] = entry
|
for request, response in responses.values():
|
client.confirm_reusable(request, response)
|
return result
|
|
|
def acquire_forecast(client: HttpClient, ticker: str, as_of: str) -> dict[str, Any]:
|
result = _result("eastmoney.forecast", as_of)
|
code, market = ticker.split(".")
|
summary_request = HttpRequest(
|
"eastmoney.forecast_summary",
|
"1.0.0",
|
"forecast_summary",
|
"GET",
|
"https://datacenter-web.eastmoney.com/api/data/v1/get",
|
ticker,
|
as_of,
|
"eastmoney_forecast_summary",
|
query={
|
"reportName": "RPT_WEB_RESPREDICT",
|
"columns": "ALL",
|
"filter": f'(SECURITY_CODE="{code}")',
|
"pageNumber": "1",
|
"pageSize": "50",
|
},
|
ttl_seconds=6 * 3600,
|
)
|
summary = client.fetch(summary_request)
|
_capture(result, summary)
|
summary_payload = _decode_json(summary)
|
rows = _records(summary_payload)
|
eligible_summary = []
|
for row in rows:
|
raw_date = row.get("REPORT_DATE") or row.get("UPDATE_DATE")
|
if not raw_date:
|
continue
|
record_date = str(raw_date)[:10]
|
if record_date <= as_of:
|
eligible_summary.append(row)
|
summary_count = int(
|
(eligible_summary[0] if eligible_summary else {}).get("RATING_ORG_NUM")
|
or summary_payload.get("summary_count")
|
or 0
|
)
|
summary_date = max(
|
(str(row.get("REPORT_DATE") or row.get("UPDATE_DATE"))[:10] for row in eligible_summary),
|
default=None,
|
)
|
forecasts: list[dict[str, Any]] = []
|
detail_error: str | None = None
|
detail_pair: tuple[HttpRequest, dict[str, Any]] | None = None
|
try:
|
detail_request = HttpRequest(
|
"eastmoney.forecast_detail",
|
"1.0.0",
|
"forecast_detail",
|
"GET",
|
"https://emweb.eastmoney.com/PC_HSF10/ProfitForecast/Index",
|
ticker,
|
as_of,
|
"eastmoney_forecast_detail",
|
query={"code": f"{market}{code}", "type": "web"},
|
content_type="text/html",
|
ttl_seconds=6 * 3600,
|
)
|
detail = client.fetch(detail_request)
|
detail_pair = (detail_request, detail)
|
_capture(result, detail)
|
text = detail["body"].decode("utf-8", errors="replace")
|
match = re.search(r'<script id="v2-fixture" type="application/json">(.*?)</script>', text, re.S)
|
if match:
|
parsed = json.loads(html.unescape(match.group(1)))["forecasts"]
|
invalid = [
|
item for item in parsed
|
if not item.get("report_date") or str(item["report_date"])[:10] > as_of
|
]
|
forecasts = [item for item in parsed if item not in invalid]
|
if invalid:
|
detail_error = "预测明细缺少真实报告日期或包含 as-of 之后记录"
|
else:
|
detail_error = "固定预测明细表未找到或 schema 漂移"
|
except Exception as exc: # forecast is explicitly non-core
|
detail_error = str(exc)
|
result["cache_integrity_failure"] = isinstance(exc, BlobIntegrityError)
|
result["institutions"] = {"coverage_status": "available" if forecasts else "gap", "forecasts": forecasts}
|
if not summary_date or summary_count != len(forecasts) or detail_error:
|
result["gaps"].append(
|
{
|
"gap_id": "W_FORECAST_COVERAGE",
|
"provider": "eastmoney.forecast",
|
"field": "institutions.forecasts",
|
"reason": detail_error or f"汇总 {summary_count} 家、可见明细 {len(forecasts)} 家",
|
"impact": "机构覆盖明细不完整，不影响法定财务计算",
|
"blocking": False,
|
"budget_used_seconds": None,
|
"manual_action": "如需逐家核对，人工补充缺失机构原报告",
|
}
|
)
|
result["sources"] = []
|
if forecasts:
|
detail_date = max(str(item["report_date"])[:10] for item in forecasts)
|
result["sources"].append({
|
"id": "SRC-INSTITUTION-DETAIL",
|
"source_type": "institution_aggregator",
|
"title": "东方财富盈利预测明细",
|
"publish_date": detail_date,
|
"period_end": detail_date,
|
"url": f"https://emweb.eastmoney.com/PC_HSF10/ProfitForecast/Index?code={market}{code}&type=web",
|
"supports": ["institutions"],
|
"revision_status": "current",
|
})
|
else:
|
detail_date = None
|
if summary_date:
|
result["sources"].append({
|
"id": "SRC-INSTITUTION-CONSENSUS",
|
"source_type": "institution_aggregator",
|
"title": "东方财富机构盈利预测汇总",
|
"publish_date": summary_date,
|
"period_end": summary_date,
|
"url": "https://datacenter-web.eastmoney.com/api/data/v1/get?reportName=RPT_WEB_RESPREDICT",
|
"supports": ["institutions"],
|
"revision_status": "current",
|
})
|
if summary_count > 0:
|
client.confirm_reusable(summary_request, summary)
|
if forecasts and detail_pair:
|
client.confirm_reusable(*detail_pair)
|
result["data_kinds"]["forecast_summary"] = _kind_entry(
|
request=summary_request,
|
response=summary,
|
ticker=ticker,
|
as_of=as_of,
|
data_date=summary_date or "9999-12-31",
|
publish_date=summary_date,
|
watermark=summary_date or "",
|
requires_publish_date=True,
|
summary_complete=bool(summary_date and summary_count > 0),
|
)
|
if detail_pair:
|
result["data_kinds"]["forecast_detail"] = _kind_entry(
|
request=detail_pair[0],
|
response=detail_pair[1],
|
ticker=ticker,
|
as_of=as_of,
|
data_date=detail_date or "9999-12-31",
|
publish_date=detail_date,
|
watermark=detail_date or "",
|
requires_publish_date=True,
|
detail_complete=bool(forecasts and detail_date),
|
)
|
return result
|
