Implement streaming enrichment aggregator

Problem Statement

You are building a real-time trade enrichment service. You receive two types of events from different upstream sources: OrderEvent and PriceEvent.

• OrderEvent: Contains {order_id, symbol, quantity}.
• PriceEvent: Contains {order_id, price}.

A complete record is formed only when both events for a specific order_id have arrived. Once both are received, you must output an EnrichedTrade object: {order_id, symbol, quantity, price}.

Implement an EnrichmentEngine class that processes these events as they arrive. If an event for an order_id arrives and its counterpart is missing, you must buffer the data in memory. To manage memory, implement a cleanup mechanism: if a partial record has been buffered for longer than T milliseconds, it is considered expired and should be discarded to free up memory.

Interface

class EnrichmentEngine:
    def __init__(self, timeout_ms: int):
        # timeout_ms is the duration a partial record stays in memory
        pass

    def process_order(self, order_id: str, symbol: str, quantity: int, timestamp_ms: int):
        pass

    def process_price(self, order_id: str, price: float, timestamp_ms: int):
        pass

Constraints

• order_id is unique per trade.
• Events can arrive out-of-order.
• The system must be thread-safe.
• Memory management is critical; you should be able to purge expired entries efficiently.

Example

1. process_order("101", "AAPL", 10, 1000) -> No output, buffer: {"101": {symbol: "AAPL", qty: 10, ts: 1000}}.
2. process_price("101", 150.0, 1050) -> Output: {"101", "AAPL", 10, 150.0}.
3. If timeout_ms was 20 and a second OrderEvent arrived at 1040 without a price, the first entry would be evicted.