"""正式发布事件驱动局部重算；独立本地函数，不监听公告、不注册调度。"""
import copy
import json
from datetime import datetime
from decimal import Decimal as D, InvalidOperation

DEPENDENCIES = {"valuation": {"price", "eps"}, "growth": {"growth"}, "liquidity": {"turnover"}}


def finite(value):
    try:
        number = D(str(value))
    except (InvalidOperation, ValueError, TypeError) as exc:
        raise ValueError("invalid_numeric_input") from exc
    if not number.is_finite():
        raise ValueError("nonfinite_financial_field")
    return number


def rule(name, row):
    if name == "valuation":
        price, eps = finite(row["price"]), finite(row["eps"])
        if price <= 0 or eps <= 0:
            raise ValueError("invalid_pe_inputs")
        return price/eps < 20
    if name == "growth":
        return finite(row["growth"]) >= D("0.10")
    if name == "liquidity":
        return finite(row["turnover"]) >= 2
    raise ValueError("unsupported_rule")


class EventScreen:
    def __init__(self):
        self.rows = {"AAA": {"price": "18", "eps": "1", "growth": "0.08", "turnover": "3"},
                     "BBB": {"price": "16", "eps": "1", "growth": "0.15", "turnover": "3"}}
        self.results = {s: {name: rule(name, row) for name in DEPENDENCIES} for s, row in self.rows.items()}
        self.done = {}

    def apply(self, event, now="2026-10-26T16:00:00+08:00"):
        if event.get("kind") != "official_release" or event.get("verified") is not True:
            raise ValueError("not_verified_official_release")
        released, decision = map(datetime.fromisoformat, [event["released_at"], now])
        if released.utcoffset() is None or decision.utcoffset() is None or released > decision:
            raise ValueError("release_not_available")
        if event.get("year") != 2026 or event.get("currency") != "CNY" or event.get("basis") != "diluted":
            raise ValueError("incompatible_event_scope")
        symbol = event["symbol"]
        if symbol not in self.rows or not event.get("source_id") or not event.get("id"):
            raise ValueError("unmapped_event")
        if not event["fields"] or set(event["fields"])-{"eps", "growth"}:
            raise ValueError("unsupported_released_fields")
        for value in event["fields"].values():
            finite(value)
        key = (symbol, event["id"])
        fingerprint = json.dumps(event, sort_keys=True)
        if key in self.done:
            if self.done[key] != fingerprint:
                raise ValueError("conflicting_event_id")
            return {"status": "duplicate_suppressed", "calls": []}
        candidate = {**self.rows[symbol], **event["fields"]}
        affected = sorted(name for name, fields in DEPENDENCIES.items() if fields & set(event["fields"]))
        results = {**self.results[symbol], **{name: rule(name, candidate) for name in affected}}
        self.rows[symbol], self.results[symbol] = candidate, results
        self.done[key] = fingerprint
        return {"status": "updated", "calls": [symbol+":"+name for name in affected], "selected": self.selected()}

    def selected(self):
        return sorted(s for s, results in self.results.items() if all(results.values()))


event = {"symbol": "AAA", "id": "filing-1", "source_id": "official:AAA:FY2026", "kind": "official_release",
         "verified": True, "released_at": "2026-10-26T15:30:00+08:00", "year": 2026,
         "currency": "CNY", "basis": "diluted", "fields": {"eps": "1.2", "growth": "0.12"}}
job = EventScreen()
bbb_before = copy.deepcopy((job.rows["BBB"], job.results["BBB"]))
assert job.selected() == ["BBB"]
main = job.apply(event)
assert main["calls"] == ["AAA:growth", "AAA:valuation"] and main["selected"] == ["AAA", "BBB"]
assert bbb_before == (job.rows["BBB"], job.results["BBB"])
assert job.apply(event)["calls"] == []
blocked = []
for bad in [{**event, "kind": "scheduled_release"}, {**event, "fields": {"eps": "9"}},
            {**event, "verified": "true"}, {**event, "released_at": "2026-10-27T15:30:00+08:00"}]:
    before = copy.deepcopy((job.rows, job.results, job.done))
    try:
        job.apply(bad)
    except ValueError as exc:
        blocked.append(str(exc))
        assert before == (job.rows, job.results, job.done)
    else:
        raise AssertionError("无效事件必须阻断")
for token in ["NaN", "Infinity", "-Infinity"]:
    try:
        EventScreen().apply({**event, "fields": {"eps": token}})
    except ValueError:
        pass
    else:
        raise AssertionError("非有限事件必须拒绝")
print(json.dumps({"main": main, "duplicate": job.apply(event), "blocked": blocked,
                  "unchanged_bbb": job.rows["BBB"]}, sort_keys=True))