from __future__ import annotations
|
|
from typing import Any
|
|
from .cache import canonical_bytes, sha256_bytes
|
|
|
class CoreInputError(ValueError):
|
pass
|
|
|
FINANCIAL_LABELS = ("annual", "current_cumulative", "prior_year_same_period")
|
FINANCIAL_VALUES = ("revenue", "attributable_profit", "deduct_profit", "cfo", "capex")
|
BALANCE_VALUES = (
|
"equity",
|
"cash_available",
|
"non_operating_financial_assets",
|
"interest_bearing_debt",
|
"minority_interest",
|
)
|
CORE_VALUE_PATHS = (
|
"market.price",
|
"market.diluted_shares",
|
"market.platform_market_cap",
|
*(f"financials.{label}.{field}" for label in FINANCIAL_LABELS for field in FINANCIAL_VALUES),
|
*(f"balance_sheet.{field}" for field in BALANCE_VALUES),
|
)
|
|
|
def protected_projection(snapshot: dict[str, Any]) -> dict[str, Any]:
|
return {
|
"meta": snapshot.get("meta"),
|
"market": snapshot.get("market"),
|
"financials": snapshot.get("financials"),
|
"balance_sheet": snapshot.get("balance_sheet"),
|
"sources": snapshot.get("sources"),
|
}
|
|
|
def protected_hash(snapshot: dict[str, Any]) -> str:
|
return sha256_bytes(canonical_bytes(protected_projection(snapshot)))
|
|
|
def _a1_for_period(
|
sources: list[dict[str, Any]], period: str, support: str
|
) -> dict[str, Any] | None:
|
candidates = [
|
source
|
for source in sources
|
if source.get("source_type") in {"cninfo", "szse", "sse", "bse", "archived_primary"}
|
and source.get("period_end") == period
|
and support in source.get("supports", [])
|
and source.get("publish_date")
|
and source.get("id")
|
]
|
return max(candidates, key=lambda item: (item["publish_date"], item["id"])) if candidates else None
|
|
|
def _require_lineage(
|
path: str,
|
lineage: dict[str, Any],
|
source_by_id: dict[str, dict[str, Any]],
|
all_raw: set[str],
|
as_of: str,
|
) -> None:
|
missing = [
|
key for key in ("source_id", "raw_hash", "publish_date", "data_date", "provider", "raw_field", "unit")
|
if not lineage.get(key)
|
]
|
if missing:
|
raise CoreInputError(f"字段血缘缺项 {path}: {','.join(missing)}")
|
if lineage["source_id"] not in source_by_id:
|
raise CoreInputError(f"字段血缘 source_id 不存在 {path}: {lineage['source_id']}")
|
structured = source_by_id[lineage["source_id"]]
|
if structured.get("period_end") != lineage["data_date"]:
|
raise CoreInputError(f"字段血缘 data_date/source period 不一致：{path}")
|
if structured.get("publish_date") != lineage["publish_date"]:
|
raise CoreInputError(f"字段血缘 publish_date/source 不一致：{path}")
|
if not any(
|
path == support or path.startswith(f"{support}.")
|
for support in structured.get("supports", [])
|
):
|
raise CoreInputError(f"字段血缘结构化来源不支持字段：{path}")
|
if lineage["raw_hash"] not in all_raw or len(lineage["raw_hash"]) != 64:
|
raise CoreInputError(f"字段血缘 raw hash 不存在 {path}")
|
if lineage["publish_date"] > as_of or lineage["data_date"] > as_of:
|
raise CoreInputError(f"E_ASOF_VIOLATION：{path}")
|
|
|
def build_data_snapshot(
|
ticker: str, as_of: str, providers: dict[str, dict[str, Any]]
|
) -> tuple[dict[str, Any], dict[str, Any]]:
|
gaps = [gap for result in providers.values() for gap in result.get("gaps", [])]
|
blocking = [gap for gap in gaps if gap.get("blocking")]
|
if blocking:
|
raise CoreInputError("；".join(gap["reason"] for gap in blocking))
|
for required_provider in ("announcements", "market", "finance", "forecast"):
|
if required_provider not in providers:
|
raise CoreInputError(f"缺少 provider：{required_provider}")
|
announcements = providers["announcements"]
|
market = providers["market"]
|
finance = providers["finance"]
|
forecast = providers["forecast"]
|
for key, container in (
|
("identity", announcements),
|
("market", market),
|
("financials", finance),
|
("balance_sheet", finance),
|
):
|
if key not in container:
|
raise CoreInputError(f"缺少核心字段组：{key}")
|
|
sources: list[dict[str, Any]] = []
|
seen: set[str] = set()
|
for result in providers.values():
|
for source in result.get("sources", []):
|
source_id = source.get("id")
|
if not source_id:
|
raise CoreInputError("来源缺少 source_id")
|
if not source.get("publish_date"):
|
raise CoreInputError(f"来源缺少 publish_date：{source_id}")
|
if source["publish_date"] > as_of:
|
raise CoreInputError(f"E_ASOF_VIOLATION：{source_id}")
|
if source.get("period_end") and source["period_end"] > as_of:
|
raise CoreInputError(f"E_ASOF_VIOLATION：{source_id}")
|
if source_id not in seen:
|
seen.add(source_id)
|
sources.append(source)
|
source_by_id = {source["id"]: source for source in sources}
|
all_raw = {
|
digest
|
for result in providers.values()
|
for digest in result.get("raw_artifact_hashes", [])
|
}
|
|
financials = finance["financials"]
|
current_period = financials["current_cumulative"]["period_end"]
|
annual_period = financials["annual"]["period_end"]
|
annual_a1 = _a1_for_period(sources, annual_period, "financials.annual")
|
current_a1 = _a1_for_period(sources, current_period, "financials.current_cumulative")
|
comparative_a1 = _a1_for_period(
|
sources, current_period, "financials.prior_year_same_period_comparative"
|
)
|
balance_a1 = _a1_for_period(sources, current_period, "balance_sheet")
|
if annual_a1 is None or current_a1 is None or comparative_a1 is None or balance_a1 is None:
|
missing = []
|
if annual_a1 is None:
|
missing.append(annual_period)
|
if current_a1 is None:
|
missing.append(current_period)
|
if comparative_a1 is None:
|
missing.append(f"{current_period}:comparative")
|
if balance_a1 is None:
|
missing.append(f"{current_period}:balance")
|
raise CoreInputError(f"核心财务缺少 A1 报告期回链：{missing}")
|
announcement_hashes = announcements.get("raw_artifact_hashes", [])
|
if not announcement_hashes:
|
raise CoreInputError("A1 公告索引缺少 raw hash")
|
a1_raw_hash = announcement_hashes[-1]
|
|
field_lineage: dict[str, dict[str, Any]] = {}
|
for result in providers.values():
|
for path, item in result.get("field_lineage", {}).items():
|
if path in field_lineage:
|
raise CoreInputError(f"字段血缘重复：{path}")
|
field_lineage[path] = dict(item)
|
for path in CORE_VALUE_PATHS:
|
if path not in field_lineage:
|
raise CoreInputError(f"核心字段缺少血缘：{path}")
|
_require_lineage(path, field_lineage[path], source_by_id, all_raw, as_of)
|
if path.startswith("financials.") or path.startswith("balance_sheet."):
|
if path.startswith("financials.annual."):
|
a1 = annual_a1
|
a1_support = "financials.annual"
|
elif path.startswith("financials.prior_year_same_period."):
|
a1 = comparative_a1
|
a1_support = "financials.prior_year_same_period_comparative"
|
relation = field_lineage[path].get("comparison_relation")
|
expected_relation = {
|
"type": "same_response_comparative_row",
|
"current_period_end": current_period,
|
"comparison_period_end": financials["prior_year_same_period"]["period_end"],
|
"shared_publish_date": a1["publish_date"],
|
}
|
if relation != expected_relation:
|
raise CoreInputError(f"上年同期比较关系无效：{path}")
|
if field_lineage[path]["publish_date"] != a1["publish_date"]:
|
raise CoreInputError(f"上年同期不是同一 A1 报告比较列：{path}")
|
elif path.startswith("balance_sheet."):
|
a1 = balance_a1
|
a1_support = "balance_sheet"
|
else:
|
a1 = current_a1
|
a1_support = "financials.current_cumulative"
|
if a1_support not in a1.get("supports", []):
|
raise CoreInputError(f"A1 未声明字段支持：{path}")
|
field_lineage[path]["a1_source_id"] = a1["id"]
|
field_lineage[path]["a1_publish_date"] = a1["publish_date"]
|
field_lineage[path]["a1_period_end"] = a1["period_end"]
|
field_lineage[path]["a1_raw_hash"] = a1_raw_hash
|
field_lineage[path]["a1_support"] = a1_support
|
if a1_raw_hash not in all_raw or a1["publish_date"] > as_of:
|
raise CoreInputError(f"核心字段 A1 血缘无效：{path}")
|
|
identity = announcements["identity"]
|
snapshot = {
|
"snapshot_id": f"SNAPSHOT-{ticker.split('.')[0]}-{as_of.replace('-', '')}-V2",
|
"meta": {
|
"company": identity["company"],
|
"code": ticker,
|
"market": identity["market"],
|
"as_of_date": as_of,
|
"currency": "CNY",
|
"report_period_end": current_period,
|
"latest_operating_info_date": max(source["publish_date"] for source in sources),
|
},
|
"market": market["market"],
|
"financials": financials,
|
"balance_sheet": finance["balance_sheet"],
|
"institutions": forecast.get("institutions", {"coverage_status": "gap", "forecasts": []}),
|
"sources": sources,
|
}
|
report = {
|
"schema_version": 2,
|
"ticker": ticker,
|
"as_of": as_of,
|
"field_lineage": field_lineage,
|
"raw_hashes": sorted(all_raw),
|
"gaps": gaps,
|
"as_of_safe": True,
|
"mechanical_protected_sha256": protected_hash(snapshot),
|
}
|
return snapshot, report
|
