订单已经发生，数据晚点才到：日报该怎样补算？

"""Offline, single-source order windows; Python 3.10+, standard library only.

Integer minutes share one fixture clock. Amounts are fictional integer units.
The source explicitly supplies watermark estimates; receipt time never advances
them automatically. Every event ID must be unique in this controlled input.
"""
from dataclasses import dataclass


@dataclass(frozen=True)
class Event:
    event_id: str
    occurred: int
    received: int
    amount: int


@dataclass(frozen=True)
class Progress:
    watermark: int
    received: int


@dataclass
class Window:
    total: int = 0
    revision: int = 0
    last_published_status: str | None = None


class Report:
    def __init__(self, size: int = 10, lateness: int = 5):
        if type(size) is not int or size <= 0:
            raise ValueError("size must be a positive integer")
        if type(lateness) is not int or lateness < 0:
            raise ValueError("lateness must be a nonnegative integer")
        self.size, self.lateness = size, lateness
        self.watermark = -1
        self.last_received = -1
        self.windows: dict[int, Window] = {}
        self.seen_ids: set[str] = set()
        self.published: list[dict] = []
        self.side_records: list[dict] = []

    def start_of(self, time: int) -> int:
        # Half-open windows: an event at 10 belongs to [10, 20), not [0, 10).
        return time // self.size * self.size

    def status(self, start: int) -> str:
        end = start + self.size
        if self.watermark >= end + self.lateness:
            return "sealed_by_policy"
        if self.watermark >= end:
            return "revisable"
        return "temporary"

    def publish(self, start: int, received: int) -> None:
        window = self.windows[start]
        window.revision += 1
        window.last_published_status = self.status(start)
        self.published.append({
            "start": start, "end": start + self.size,
            "revision": window.revision, "total": window.total,
            "status": self.status(start), "watermark": self.watermark,
            "received": received,
        })

    def accept(self, item: Event | Progress) -> None:
        if not isinstance(item, (Event, Progress)):
            raise TypeError("expected Event or Progress")
        if type(item.received) is not int or item.received < self.last_received:
            raise ValueError("received times must be nondecreasing integers")
        if item.received < 0:
            raise ValueError("fixture received time must be nonnegative")
        if isinstance(item, Progress):
            if (type(item.watermark) is not int
                    or not self.watermark <= item.watermark <= item.received):
                raise ValueError("fixture watermark must be monotone and <= received")
            self.last_received = item.received
            self.watermark = item.watermark
            for start, window in sorted(self.windows.items()):
                status = self.status(start)
                if status != "temporary" and status != window.last_published_status:
                    self.publish(start, item.received)
            return
        if not isinstance(item.event_id, str) or not item.event_id:
            raise ValueError("event ID must be a nonempty string")
        if item.event_id in self.seen_ids:
            raise ValueError("fixture event IDs must be unique")
        if (type(item.occurred) is not int
                or not 0 <= item.occurred <= item.received):
            raise ValueError("fixture occurrence must be between 0 and receipt")
        if type(item.amount) is not int or item.amount <= 0:
            raise ValueError("fixture amount must be a positive integer")

        self.last_received = item.received
        self.seen_ids.add(item.event_id)
        start = self.start_of(item.occurred)
        if self.status(start) == "sealed_by_policy":
            self.side_records.append({
                "event": item, "start": start, "end": start + self.size,
                "watermark": self.watermark,
                "reason": "received after the policy cutoff",
            })
            return
        window = self.windows.setdefault(start, Window())
        window.total += item.amount
        if self.status(start) == "revisable":
            self.publish(start, item.received)


def raw_totals(events: list[Event], field: str, size: int = 10) -> dict[int, int]:
    totals: dict[int, int] = {}
    for event in events:
        start = getattr(event, field) // size * size
        totals[start] = totals.get(start, 0) + event.amount
    return totals


def must_reject(action) -> None:
    try:
        action()
    except ValueError:
        return
    raise AssertionError("expected rejection")


def boundary_checks() -> None:
    edge = Report()
    edge.accept(Event("at-10", 10, 10, 1))
    assert 10 in edge.windows and 0 not in edge.windows

    boundary = Report()
    boundary.accept(Progress(14, 14))
    boundary.accept(Event("before-cutoff", 9, 15, 1))
    assert boundary.windows[0].total == 1
    assert boundary.published[-1]["status"] == "revisable"
    boundary.accept(Progress(15, 16))
    boundary.accept(Event("at-cutoff", 9, 17, 2))
    assert boundary.windows[0].total == 1 and len(boundary.side_records) == 1
    count = len(boundary.published)
    boundary.accept(Progress(15, 18))
    assert len(boundary.published) == count
    must_reject(lambda: boundary.accept(Progress(14, 19)))
    assert boundary.watermark == 15 and boundary.last_received == 18
    must_reject(lambda: boundary.accept(Event("at-cutoff", 9, 19, 2)))
    assert len(boundary.side_records) == 1

    zero = Report(lateness=0)
    zero.accept(Event("zero", 2, 3, 1))
    zero.accept(Progress(10, 10))
    assert zero.published[-1]["status"] == "sealed_by_policy"
    assert zero.published[-1]["revision"] == 1
    zero.accept(Event("zero-late", 3, 11, 2))
    assert zero.windows[0].total == 1 and len(zero.side_records) == 1

    idle = Report()
    idle.accept(Event("old", 2, 100, 1))
    assert idle.watermark == -1 and idle.status(0) == "temporary"
    assert idle.published == []
    empty = Report()
    empty.accept(Progress(15, 15))
    empty.accept(Event("unseen-old-window", 2, 16, 1))
    assert empty.windows == {} and len(empty.side_records) == 1
    must_reject(lambda: Report(lateness=-1))
    must_reject(lambda: Report(size=0))


def main() -> None:
    stream: list[Event | Progress] = [
        Event("A", 2, 3, 10),
        Event("B", 12, 12, 20),
        Event("C", 8, 14, 30),
        Progress(10, 15),
        Event("E", 5, 16, 50),
        Event("D", 16, 18, 40),
        Progress(15, 20),
        Event("G", 9, 21, 70),
        Event("F", 21, 22, 60),
    ]
    report = Report()
    for item in stream:
        report.accept(item)
    events = [item for item in stream if isinstance(item, Event)]
    occurred = raw_totals(events, "occurred")
    received = raw_totals(events, "received")
    assert occurred == {0: 160, 10: 60, 20: 60}
    assert received == {0: 10, 10: 140, 20: 130}
    print("Same seven raw orders, grouped two ways:")
    print("window     by_occurrence  by_receipt")
    for start in (0, 10, 20):
        print(f"[{start:02d},{start + 10:02d})    {occurred[start]:>3}            {received[start]:>3}")
    print("Policy publications (replace row by window, never add totals):")
    for row in report.published:
        print(f"r={row['received']:02d} W={row['watermark']:02d} "
              f"[{row['start']:02d},{row['end']:02d}) revision={row['revision']} "
              f"total={row['total']} {row['status']}")
    assert [(row["revision"], row["total"], row["status"]) for row in report.published] == [
        (1, 40, "revisable"), (2, 90, "revisable"), (3, 90, "sealed_by_policy")]
    print(f"At receipt=22: watermark={report.watermark}; current stored windows:")
    for start, window in sorted(report.windows.items()):
        print(f"[{start:02d},{start + report.size:02d}) total={window.total} {report.status(start)}")
    assert report.side_records[0]["event"].event_id == "G"
    side_total = sum(record["event"].amount for record in report.side_records)
    stored_total = sum(window.total for window in report.windows.values())
    assert sum(occurred.values()) == sum(received.values()) == stored_total + side_total == 280
    side = report.side_records[0]
    late = side["event"]
    print(f"Side record: {late.event_id} occurred={late.occurred} "
          f"received={late.received} amount={late.amount} W={side['watermark']}")
    print(f"Reason: {side['reason']}; sealed [00,10) remains 90")
    print(f"Conservation: stored={stored_total} + side={side_total} = raw=280")
    boundary_checks()
    print("PASS: half-open windows, equality cutoffs, revisions, side records, no clock-driven progress")


if __name__ == "__main__":
    main()