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'', 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