Your question is Usage Metrics Aggregation Under Constraints. Start with the requirements on the right.
Run and submit as often as you like. When you're ready, talk me through your approach or go straight to the code.
Descript receives a stream of usage events such as edit, export, and transcribe. Implement a bounded-memory aggregator that identifies frequently occurring event types without storing every distinct event.
Use the space-saving algorithm. Maintain at most capacity event types. For each event, increment its tracked estimate if present. Otherwise, replace the currently least frequent tracked event with the new event, assigning the replacement an estimate of minimum_count + 1 and an error bound of minimum_count.
Implement aggregate_usage(events, capacity), where events is a list of strings representing an event stream and capacity is the maximum number of tracked event types. Return a dictionary mapping each retained event type to [estimated_count, error_bound]. The estimate must be no smaller than the event's true count, and the error bound must represent the maximum possible overestimate introduced by replacement.
The implementation must process events in one pass and use O(capacity) memory. Use a min-heap or an equivalent indexed heap so updates remain efficient. When counts tie, the lexicographically smaller event is considered the minimum.
def aggregate_usage(events, capacity):