"""按证券日收盘截止执行as-of连接；精确同刻发布按政策延至下次会话。"""
import json
from bisect import bisect_right
from datetime import datetime

days = ["2026-10-19", "2026-10-20", "2026-10-21", "2026-10-22", "2026-10-23", "2026-10-26"]
cutoffs = [datetime.fromisoformat(d+"T15:00:00+08:00") for d in days]
events = [{"id": "a-original", "symbol": "AAA", "published": "2026-10-20T16:00:00+08:00", "eps": "1.0"},
          {"id": "a-revision", "symbol": "AAA", "published": "2026-10-22T14:00:00+08:00", "eps": "1.1"},
          {"id": "a-revision", "symbol": "AAA", "published": "2026-10-22T14:00:00+08:00", "eps": "1.1"},
          {"id": "a-at-close", "symbol": "AAA", "published": "2026-10-23T15:00:00+08:00", "eps": "1.2"},
          {"id": "a-future", "symbol": "AAA", "published": "2026-10-26T16:00:00+08:00", "eps": "9"},
          {"id": "b-original", "symbol": "BBB", "published": "2026-10-19T10:00:00+08:00", "eps": "2"},
          {"id": "c-no-time", "symbol": "CCC", "published": None, "eps": "3"}]
prices = [{"symbol": "AAA", "date": d, "close": 10+i} for i, d in enumerate(days)]
prices += [{"symbol": "BBB", "date": days[0], "close": 20}, {"symbol": "CCC", "date": days[0], "close": 30}]

def join(records):
    unique, excluded, pending, prepared = {}, [], [], {}
    for row in records:
        identity = (row["symbol"], row["id"])
        if identity in unique and unique[identity] != row:
            raise ValueError("conflicting_event_id")
        unique[identity] = row
    for row in unique.values():
        if row["published"] is None:
            excluded.append(row["id"]); continue
        published = datetime.fromisoformat(row["published"])
        if published.tzinfo is None:
            raise ValueError("timezone_required")
        eligible = [c for c in cutoffs if published < c]
        if not eligible:
            pending.append(row["id"]); continue
        prepared.setdefault(row["symbol"], []).append((min(eligible), published, row["id"], row["eps"]))
    for symbol, values in prepared.items():
        if len({(v[0], v[1]) for v in values}) != len(values):
            raise ValueError("ambiguous_same_time_revision")
        values.sort()
    panel = []
    for price in prices:
        values = prepared.get(price["symbol"], [])
        decision = datetime.fromisoformat(price["date"]+"T15:00:00+08:00")
        i = bisect_right([v[0] for v in values], decision)-1
        selected = values[i] if i >= 0 else None
        panel.append({**price, "eps": selected[3] if selected else None,
                      "event_id": selected[2] if selected else None,
                      "published_at": selected[1].isoformat() if selected else None,
                      "effective_at": selected[0].isoformat() if selected else None})
    return panel, sorted(excluded), sorted(pending)

panel, excluded, pending = join(events)
assert [r["eps"] for r in panel if r["symbol"] == "AAA"] == [None, None, "1.0", "1.1", "1.1", "1.2"]
assert panel[6]["eps"] == "2" and panel[7]["eps"] is None
assert excluded == ["c-no-time"] and pending == ["a-future"]
assert join(list(reversed(events))) == (panel, excluded, pending)
cross_security_events = [{**r, "id": "filing-1"} if r["id"] in {"a-original", "b-original"} else r for r in events]
cross_panel, _, _ = join(cross_security_events)
assert cross_panel[2]["eps"] == "1.0" and cross_panel[6]["eps"] == "2"
assert join(list(reversed(cross_security_events)))[0] == cross_panel
try:
    join(events + [{**events[1], "eps": "99"}])
except ValueError as exc:
    assert str(exc) == "conflicting_event_id"
else:
    raise AssertionError("冲突修订编号必须拒绝")
print(json.dumps({"panel": panel, "excluded_missing_time": excluded, "pending_beyond_horizon": pending,
                  "policy": "published strictly before close cutoff; exact-close release becomes next session"}, sort_keys=True))