Implement deduped CTR/RPM aggregator over event stream

Implement a Python function to compute per-day, per-campaign CTR and RPM from an event stream with possible out-of-order and duplicate click events. Input: An iterator of dicts, each with keys: {"dt": "YYYY-MM-DD", "event_time": ISO8601 string, "ad_request_id": str, "campaign_id": int, "event_type": "impression"|"click", "revenue_cents": int}. Events can arrive out of order; multiple clicks may share the same ad_request_id. Rules:

1. Count exactly one impression per ad_request_id (if multiple impression events appear, count once).
2. Count at most one charged click per ad_request_id using the earliest click’s revenue; ignore later clicks for that ad_request_id.
3. For each (dt, campaign_id): CTR = charged_clicks / impressions; RPM = (sum(revenue_cents)/100) / impressions * 1000.
4. Memory constraints: process in O(U) additional memory where U is the number of active ad_request_ids seen in the current 24-hour window; do not materialize the full stream.
5. Time complexity: O(n) over n events.
6. Handle missing/invalid fields robustly (skip invalid records but continue), and support that an impression could arrive after a click for the same ad_request_id. Output: A mapping {(dt, campaign_id): {"impressions": int, "charged_clicks": int, "ctr": float, "rpm": float}}. Deliverables: the function with type hints and a docstring, plus two unit tests: (a) multiple clicks per request id; (b) out-of-order impression and click spanning two dates.

Compute deduplicated daily CTR and RPM per campaign from ad events

You are given a table ad_events that records impression and click events for ad requests. Each physical ad request is identified by ad_request_id, but due to retries and logging issues the table may contain duplicate impressions and multiple clicks for the same ad_request_id. Some impressions and clicks for the same ad_request_id can appear out of order, and an impression for a given ad_request_id may be on a different date than its click.

Using the ad_events table defined below, write a SQL query to compute, for each (dt, campaign_id), the daily number of impressions, the number of charged clicks, the click-through rate (CTR), and revenue per thousand impressions (RPM), applying these rules:

1. Count at most one impression per ad_request_id overall. If multiple impression rows exist for the same ad_request_id, count exactly one impression for that ad_request_id, using the ad request's date (dt) and campaign_id (they are expected to be consistent).
2. Count at most one charged click per ad_request_id. If there are multiple click events for the same ad_request_id, only the earliest click by event_time is charged; later clicks for that ad_request_id are ignored.
3. Aggregate metrics per (dt, campaign_id) using the dt and campaign_id of each (deduplicated) impression or click event:
   • impressions = count of unique ad_request_ids with at least one impression on that (dt, campaign_id)
   • charged_clicks = count of unique ad_request_ids whose earliest click falls on that (dt, campaign_id)
   • CTR = charged_clicks / impressions (use 0 when impressions = 0)
   • RPM = (sum(revenue_cents) / 100.0) / impressions * 1000 (use 0 when impressions = 0)
4. If a day/campaign has charged clicks but no impressions on that same day (for example, when the impression occurred on a previous date), report impressions = 0, CTR = 0, and RPM = 0 for that (dt, campaign_id).

Return one row per (dt, campaign_id) present in either impressions or clicks, with columns: dt, campaign_id, impressions, charged_clicks, ctr, rpm.

Table ad_events

column	type	constraints
event_id	INT	PRIMARY KEY
dt	DATE	NOT NULL
event_time	TIMESTAMP	NOT NULL
ad_request_id	VARCHAR(50)	NOT NULL
campaign_id	INT	NOT NULL
event_type	VARCHAR(20)	NOT NULL
revenue_cents	INT	NOT NULL

Sample data:

ad_request_id	campaign_id	dt	event_id	event_time	event_type	revenue_cents
req1	101	2025-05-31	1	2025-05-31 10:00:00	impression	0
req1	101	2025-05-31	2	2025-05-31 10:00:05	impression	0
req1	101	2025-05-31	3	2025-05-31 10:00:10	click	50
req1	101	2025-05-31	4	2025-05-31 10:00:20	click	100
req2	101	2025-06-01	5	2025-06-01 00:05:00	click	200
req2	101	2025-05-31	6	2025-05-31 23:55:00	impression	0
req3	102	2025-06-01	7	2025-06-01 09:00:00	impression	0
req3	102	2025-06-01	8	2025-06-01 09:00:30	click	300
req3	102	2025-06-01	9	2025-06-01 09:01:00	impression	0

Expected output

One row per (dt, campaign_id) with deduplicated impressions and earliest-click revenue per ad_request_id. CTR is charged_clicks / impressions (0 when impressions = 0). RPM is ((sum of revenue_cents)/100) / impressions * 1000 (0 when impressions = 0). Based on the sample data, req1 has duplicate impressions and multiple clicks on 2025-05-31; only one impression and the earliest click with 50 cents are counted. req2 has an impression on 2025-05-31 and its (earliest) click on 2025-06-01; the impression contributes to 2025-05-31, while the click contributes to 2025-06-01, producing a day with clicks but zero impressions. req3 has duplicate impressions and a single click on 2025-06-01.

campaign_id	charged_clicks	ctr	dt	impressions	rpm
101	1	0.5	2025-05-31	2	250
101	1	0	2025-06-01	0	0
102	1	1	2025-06-01	1	3000

Hints

1. Use a GROUP BY on ad_request_id to deduplicate impressions, selecting a single dt and campaign_id per ad_request_id.
2. Use ROW_NUMBER() OVER (PARTITION BY ad_request_id ORDER BY event_time) to select the earliest click per ad_request_id, then aggregate clicks by dt and campaign_id and combine with impressions using a FULL OUTER JOIN.