"""Illustrative local ledger. No network calls or product SDK dependency.""" from dataclasses import dataclass from decimal import Decimal from typing import Iterable @dataclass(frozen=True) class Charge: event_id: str work_id: str attempt_id: str dollars: Decimal def summarize(charges: Iterable[Charge], outcomes: dict[str, bool | None]) -> dict: """An outcome is True/False only once its predeclared window has closed.""" unique: dict[str, Charge] = {} for charge in charges: if not charge.dollars.is_finite() or charge.dollars < 0: raise ValueError('Require a finite nonnegative charge') if not all((charge.event_id, charge.work_id, charge.attempt_id)): raise ValueError('All identifiers are required') if charge.event_id in unique and unique[charge.event_id] != charge: raise ValueError('Conflicting duplicate event; reconcile before reporting') unique[charge.event_id] = charge if any(c.work_id not in outcomes for c in unique.values()): raise ValueError('A charge has no declared cohort work item') if any(value is not None and type(value) is not bool for value in outcomes.values()): raise ValueError('Outcome must be true, false, or unknown') total = sum((c.dollars for c in unique.values()), Decimal('0')) successes = sum(value is True for value in outcomes.values()) unknown = sum(value is None for value in outcomes.values()) return {'total_dollars': total, 'successful_items': successes, 'unknown_items': unknown, 'dollars_per_success': total / successes if successes and not unknown else None} if __name__ == '__main__': events = [Charge('e1','ticket-1','a1',Decimal('0.03')), Charge('e1','ticket-1','a1',Decimal('0.03')), Charge('e2','ticket-1','a2',Decimal('0.02')), Charge('e3','ticket-2','a3',Decimal('0.05'))] print(summarize(events, {'ticket-1': True, 'ticket-2': False}))