from __future__ import annotations
|
|
import copy
|
import json
|
from pathlib import Path
|
from typing import Any
|
|
|
ALLOWED_TOP = {"normalization", "valuation", "analysis"}
|
PROTECTED_KEYS = {
|
"price",
|
"price_timestamp",
|
"diluted_shares",
|
"shares_date",
|
"platform_market_cap",
|
"revenue",
|
"attributable_profit",
|
"deduct_profit",
|
"period_end",
|
"publish_date",
|
"source_id",
|
"url",
|
"raw_hash",
|
}
|
|
|
class JudgmentConflict(ValueError):
|
pass
|
|
|
def _scan(value: Any, path: str = "") -> None:
|
if isinstance(value, dict):
|
for key, child in value.items():
|
current = f"{path}.{key}" if path else key
|
reference_only = current.startswith("normalization.adjustments[") and key == "source_id"
|
if key in PROTECTED_KEYS and not reference_only:
|
raise JudgmentConflict(f"E_JUDGMENT_CONFLICT：禁止字段 {current}")
|
_scan(child, current)
|
elif isinstance(value, list):
|
for index, child in enumerate(value):
|
_scan(child, f"{path}[{index}]")
|
|
|
def load_overlay(path: Path) -> dict[str, Any]:
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
unknown = set(payload) - ALLOWED_TOP
|
if unknown:
|
raise JudgmentConflict(f"E_JUDGMENT_CONFLICT：未知顶层字段 {sorted(unknown)}")
|
_scan(payload)
|
scenarios = ((payload.get("valuation") or {}).get("scenarios") or [])
|
if not scenarios or sum(1 for item in scenarios if item.get("role") == "base") != 1:
|
raise JudgmentConflict("judgment 必须包含且只包含一个 base 情景")
|
return payload
|
|
|
def apply_overlay(data_snapshot: dict[str, Any], overlay: dict[str, Any]) -> dict[str, Any]:
|
snapshot = copy.deepcopy(data_snapshot)
|
before = {
|
key: copy.deepcopy(snapshot.get(key))
|
for key in ("meta", "market", "financials", "balance_sheet", "institutions", "sources")
|
}
|
for key in ALLOWED_TOP:
|
if key in overlay:
|
snapshot[key] = copy.deepcopy(overlay[key])
|
known_sources = {item["id"] for item in data_snapshot.get("sources", [])}
|
for adjustment in snapshot.get("normalization", {}).get("adjustments", []):
|
source_id = adjustment.get("source_id")
|
if source_id not in known_sources:
|
raise JudgmentConflict(f"E_JUDGMENT_CONFLICT：未知只读来源引用 {source_id}")
|
after = {key: snapshot.get(key) for key in before}
|
if before != after:
|
raise JudgmentConflict("E_JUDGMENT_CONFLICT：保护数据被覆盖")
|
return snapshot
|
