An optimized and corrected version of the complete Python data processing pipeline is provided below.
from collections import deque, defaultdictimport datetimefrom typing import List, Dict, Tuple, Any
# =====================================================================# 1. PURCHASE WINDOW FILTERING# =====================================================================def get_frequent_users(events: List[Tuple[str, str]], n: int) -> List[str]:
"""
Returns users who made more than N purchases in any 30-day window.
events: List of tuples (user_id, timestamp_str) where timestamp_str is 'YYYY-MM-DD'
"""
user_history = defaultdict(list)
for user_id, t_str in events:
dt = datetime.datetime.strptime(t_str, "%Y-%m-%d")
user_history[user_id].append(dt)
frequent_users = []
for user_id, timestamps in user_history.items():
timestamps.sort() # Linearithmic sort per user for chronological order
left = 0
max_in_window = 0
for right in range(len(timestamps)):
# Maintain sliding window boundaries for exactly 30 days
while (timestamps[right] - timestamps[left]).days > 30:
left += 1
current_window_count = right - left + 1
if current_window_count > max_in_window:
max_in_window = current_window_count
if max_in_window > n:
frequent_users.append(user_id)
return frequent_users
# =====================================================================# 2. QUEUEING SYSTEM SIMULATION# =====================================================================def simulate_queue(arrivals: List[float], service_times: List[float]) -> float:
"""
Simulates a single-server First-In, First-Out (FIFO) queueing loop.
Returns the average wait time for all arriving processes.
"""
if not arrivals:
return 0.0
total_wait_time = 0.0
current_time = 0.0
for arrival, service in zip(arrivals, service_times):
# Server starts processing when the job arrives or when the server finishes previous task
start_time = max(arrival, current_time)
wait_time = start_time - arrival
total_wait_time += wait_time
# Advance clock by the duration of active service execution
current_time = start_time + service
return total_wait_time / len(arrivals)
# =====================================================================# 3. ROLLING MEDIAN ANOMALY DETECTION# =====================================================================def detect_anomalies(values: List[float], k: int, threshold: float) -> List[bool]:
"""
Detects anomalies by comparing each value to the median of the previous k values.
threshold: Maximum allowable absolute difference deviation boundary from the rolling median.
"""
anomalies = []
window = deque(maxlen=k)
for val in values:
if len(window) < k:
# Not enough historical lookback context to flags anomalies
anomalies.append(False)
else:
# Compute the rolling median over historical context window
sorted_window = sorted(list(window))
mid = k // 2
if k % 2 == 1:
median = sorted_window[mid]
else:
median = (sorted_window[mid - 1] + sorted_window[mid]) / 2.0
# Flag item if it steps past absolute threshold boundaries
is_anomaly = abs(val - median) > threshold
anomalies.append(is_anomaly)
window.append(val)
return anomalies
# =====================================================================# 4. CATEGORY REVIEW GROUPING# =====================================================================def top_rated_by_category(reviews: List[Dict[str, Any]]) -> Dict[str, List[str]]:
"""
Groups product reviews by category and returns the highest-rated product(s).
reviews: List of dicts, e.g., [{"product": "A", "category": "Tech", "rating": 4.8}]
"""
product_ratings = defaultdict(list)
product_category = {}
# Map raw records to grouped state
for r in reviews:
prod = r["product"]
cat = r["category"]
rating = r["rating"]
product_ratings[prod].append(rating)
product_category[prod] = cat
# Aggregate to calculate mean rating evaluations
avg_ratings = {prod: sum(rat)/len(rat) for prod, rat in product_ratings.items()}
category_groups = defaultdict(list)
for prod, avg_rating in avg_ratings.items():
cat = product_category[prod]
category_groups[cat].append((prod, avg_rating))
result = {}
for cat, prods in category_groups.items():
# Find highest rating score within this specific category subset
max_rating = max(prods, key=lambda x: x[1])[1]
# Select product keys matching maximum bounds to catch duplicates or ties
top_prods = [p[0] for p in prods if p[1] == max_rating]
result[cat] = top_prods
return result
# =====================================================================# DEMO EXECUTION# =====================================================================if __name__ == "__main__":
print("--- 1. Purchase Filter ---")
purchases = [
("user1", "2026-01-01"),
("user1", "2026-01-15"),
("user1", "2026-01-25"),
("user2", "2026-01-01")
]
print("Frequent Users (N=2):", get_frequent_users(purchases, n=2))
print("\n--- 2. Queue Simulation ---")
# Job 1 arrives at 0.0 -> processes immediately (wait=0.0) -> finishes at 5.0
# Job 2 arrives at 2.0 -> waits until 5.0 (wait=3.0) -> finishes at 9.0
print("Avg Wait Time:", simulate_queue(arrivals=[0.0, 2.0], service_times=[5.0, 4.0]))
print("\n--- 3. Anomaly Detection ---")
stream = [10.0, 12.0, 11.0, 13.0, 100.0, 12.0, 11.0]
print("Anomalies (k=4, thresh=15):", detect_anomalies(stream, k=4, threshold=15.0))
print("\n--- 4. Review Grouping ---")
sample_reviews = [
{"product": "Phone X", "category": "Tech", "rating": 5},
{"product": "Phone X", "category": "Tech", "rating": 4}, # Mean = 4.5
{"product": "Laptop Y", "category": "Tech", "rating": 5}, # Mean = 5.0
{"product": "Shirt Z", "category": "Apparel", "rating": 4} # Mean = 4.0
]
print("Top Products by Category:", top_rated_by_category(sample_reviews))
If you are dealing with performance constraints, tell me if you want to optimize the rolling median using dual min/max heaps to scale down execution complexity from $O(k \log k)$ to $O(\log k)$.
Comments