-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathmain.py
More file actions
28 lines (25 loc) · 1.02 KB
/
main.py
File metadata and controls
28 lines (25 loc) · 1.02 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from datetime import datetime, timezone, timedelta
from typing import Iterable, List, Dict, Any
from collections import defaultdict
class Main:
#
# ---------------------------------------------------------------------------
# Implement your solution here
#
# ---------------------------------------------------------------------------
def aggregate_events_by_window(self,
events: Iterable[Dict[str, Any]],
window_minutes: int,
) -> List[Dict[str, Any]]:
"""
Group events into tumbling time windows and compute per-tenant
aggregates.
Args:
events: Iterable of event dictionaries with keys:
tenant_id, user_id, event_time, amount
window_minutes: Size of tumbling window in minutes
Returns:
List of aggregate dictionaries sorted by (tenant_id, window_start)
"""
# TODO: Implement your solution here
raise NotImplementedError("Implement aggregate_events_by_window")