"""新报告驱动估值更新：事务式状态、重复/冲突、非盈利与晚到事件。"""
import copy
import hashlib
import json
from datetime import datetime
from decimal import Decimal as D


def stamp(value):
    result = datetime.fromisoformat(value)
    if result.tzinfo is None:
        raise ValueError("timezone_required")
    return result


def process(state, reports, quote, cutoff):
    price = D(quote["price"])
    if not price.is_finite() or price <= 0:
        raise ValueError("invalid_price")
    stop = stamp(cutoff)
    if stamp(quote["as_of"]) > stop:
        raise ValueError("future_quote")
    staged = copy.deepcopy(state)
    changes, ignored = [], []
    for report in sorted(reports, key=lambda r: (stamp(r["published"]), r["id"])):
        if report["symbol"] != quote["symbol"] or report["currency"] != quote["currency"]:
            raise ValueError("incompatible_quote")
        if report["basis"] != "diluted" or report["unit"] != "per_share":
            raise ValueError("incompatible_eps_context")
        eps = D(report["eps"])
        if not eps.is_finite():
            raise ValueError("nonfinite_eps")
        if stamp(report["published"]) > stop:
            ignored.append([report["id"], "future"])
            continue
        identity = (report["symbol"], report["broker"], report["id"], report["year"])
        digest = hashlib.sha256(json.dumps(report, sort_keys=True).encode()).hexdigest()
        if identity in staged["seen"]:
            if staged["seen"][identity] != digest:
                raise ValueError("same_identity_changed_payload")
            ignored.append([report["id"], "duplicate"])
            continue
        key = (report["symbol"], report["broker"], report["year"])
        previous = staged["latest"].get(key)
        if previous and stamp(report["published"]) == stamp(previous["published"]) and eps != D(previous["eps"]):
            raise ValueError("ambiguous_same_time_forecast")
        staged["seen"][identity] = digest
        if previous and stamp(report["published"]) < stamp(previous["published"]):
            ignored.append([report["id"], "late_older_report"])
            continue
        staged["latest"][key] = copy.deepcopy(report)
        if previous is not None and eps == D(previous["eps"]):
            ignored.append([report["id"], "numeric_unchanged"])
            continue
        pe = price / eps if eps > 0 else None
        staged["values"][key] = {"eps": str(eps), "forward_pe": str(pe) if pe is not None else None,
                                  "status": "calculated" if pe is not None else "nonpositive_eps",
                                  "calculation_source": report["id"], "price_as_of": quote["as_of"]}
        changes.append({"report": report["id"], "forward_pe": str(pe) if pe is not None else None})
    state.clear()
    state.update(staged)
    return {"recalculations": changes, "ignored": ignored}


quote = {"symbol": "AAA", "currency": "CNY", "price": "20", "as_of": "2026-10-19T15:00:00+08:00"}
base = {"symbol": "AAA", "broker": "B", "year": 2027, "currency": "CNY", "eps": "1",
        "basis": "diluted", "unit": "per_share",
        "id": "r1", "rating": "neutral", "published": "2026-10-20T10:00:00+08:00"}
state = {"seen": {}, "latest": {}, "values": {}}
initial = process(state, [base], quote, "2026-10-26T15:00:00+08:00")
rating = {**base, "id": "r2", "rating": "positive", "published": "2026-10-21T10:00:00+08:00"}
unchanged = process(state, [rating], quote, "2026-10-26T15:00:00+08:00")
upgrade = {**base, "id": "r3", "eps": "1.25", "published": "2026-10-22T10:00:00+08:00"}
changed = process(state, [upgrade], quote, "2026-10-26T15:00:00+08:00")
duplicate = process(state, [upgrade], quote, "2026-10-26T15:00:00+08:00")
assert initial["recalculations"] == [{"report": "r1", "forward_pe": "20"}]
assert unchanged["recalculations"] == [] and changed["recalculations"] == [{"report": "r3", "forward_pe": "16"}]
assert duplicate["recalculations"] == [] and len(state["seen"]) == 3
saved = copy.deepcopy(state)
for bad in ({**upgrade, "eps": "9"}, {**upgrade, "id": "bad", "eps": "NaN"}):
    try:
        process(state, [bad], quote, "2026-10-26T15:00:00+08:00")
    except ValueError:
        assert state == saved
    else:
        raise AssertionError("冲突或非有限数据不能更新状态")
older = process(state, [{**base, "id": "late-old"}], quote, "2026-10-26T15:00:00+08:00")
assert older["ignored"] == [["late-old", "late_older_report"]]
loss = process(state, [{**base, "id": "r4", "eps": "-0.2", "published": "2026-10-23T10:00:00+08:00"}], quote, "2026-10-26T15:00:00+08:00")
assert loss["recalculations"] == [{"report": "r4", "forward_pe": None}]
multi_state = {"seen": {}, "latest": {}, "values": {}}
multi_rows = [base, {**base, "year": 2028, "eps": "2"}]
multi = process(multi_state, multi_rows, quote, "2026-10-26T15:00:00+08:00")
assert len(multi["recalculations"]) == 2
assert multi_state["values"][("AAA", "B", 2027)]["forward_pe"] == "20"
assert multi_state["values"][("AAA", "B", 2028)]["forward_pe"] == "10"
assert process(multi_state, list(reversed(multi_rows)), quote, "2026-10-26T15:00:00+08:00")["recalculations"] == []
multi_saved = copy.deepcopy(multi_state)
try:
    process(multi_state, [{**base, "year": 2028, "eps": "3"}], quote, "2026-10-26T15:00:00+08:00")
except ValueError:
    assert multi_state == multi_saved
else:
    raise AssertionError("同报告同财年载荷冲突不得发布")
print(json.dumps({"initial": initial, "rating_only": unchanged, "forecast_change": changed,
                  "repeat": duplicate, "late": older, "loss": loss,
                  "multi_year": {str(k[2]): v["forward_pe"] for k, v in multi_state["values"].items()},
                  "states": list(state["values"].values())}, sort_keys=True))