"""Statistical anomaly issues (10 issues).""" import uuid from datetime import timedelta import numpy as np import pandas as pd from .base import Issue, pick_rows def _bot_traffic(df, rate, rng): """Single user with 1000 events in one minute.""" bot_user = str(uuid.uuid4()) bot_session = str(uuid.uuid4()) base_ts = df["timestamp"].dropna().iloc[0] if not df["timestamp"].dropna().empty else pd.Timestamp("2024-01-01") rows = [] for i in range(1000): rows.append({ "event_id": str(uuid.uuid4()), "user_id": bot_user, "session_id": bot_session, "event_type": "click", "timestamp": base_ts + timedelta(seconds=i % 60), "page_url": "http://example.com/bot", "referrer_url": None, "ip_address": "198.51.100.1", "device_type": "desktop", "os": "Linux", "browser": "Chrome", "country_code": "US", "city": "Botville", "duration_seconds": 0, "revenue": 0.0, "product_id": None, "quantity": 0, "search_query": None, "user_agent": "python-requests/2.28", "is_authenticated": False, "user_age": 0, "email": "bot@spam.com", "user_registration_date": None, }) return pd.concat([df, pd.DataFrame(rows)], ignore_index=True) def _single_ip_all_events(df, rate, rng): df = df.copy() mask = pick_rows(df, rate * 10, rng) df.loc[mask, "ip_address"] = "203.0.113.42" return df def _all_same_timestamp(df, rate, rng): df = df.copy() mask = pick_rows(df, rate, rng) shared_ts = pd.Timestamp("2024-01-15 00:00:00") df.loc[mask, "timestamp"] = shared_ts return df def _zero_variance_column(df, rate, rng): """Set country_code to a single value for a large chunk of rows.""" df = df.copy() mask = pick_rows(df, 0.9, rng) df.loc[mask, "country_code"] = "US" return df def _sudden_spike(df, rate, rng): """Add a burst of 500 identical events in a narrow time window.""" base_ts = pd.Timestamp("2024-02-01 12:00:00") rows = [] for _ in range(500): rows.append({ "event_id": str(uuid.uuid4()), "user_id": str(uuid.uuid4()), "session_id": str(uuid.uuid4()), "event_type": "page_view", "timestamp": base_ts + timedelta(seconds=int(rng.integers(0, 60))), "page_url": "http://example.com/viral", "referrer_url": None, "ip_address": "198.51.100.0", "device_type": "mobile", "os": "Android", "browser": "Chrome", "country_code": "US", "city": "New York", "duration_seconds": 1, "revenue": 0.0, "product_id": None, "quantity": 0, "search_query": None, "user_agent": "Mozilla/5.0", "is_authenticated": False, "user_age": 25, "email": "anon@example.com", "user_registration_date": None, }) return pd.concat([df, pd.DataFrame(rows)], ignore_index=True) def _missing_time_range(df, rate, rng): """Delete all events in a 6-hour window to simulate a data gap.""" df = df.copy() gap_start = pd.Timestamp("2024-02-10 02:00:00") gap_end = pd.Timestamp("2024-02-10 08:00:00") def in_gap(t): try: ts = pd.Timestamp(t) return gap_start <= ts <= gap_end except Exception: return False keep = ~df["timestamp"].apply(in_gap) return df[keep].reset_index(drop=True) def _all_events_same_user(df, rate, rng): """Skew 90% of events to one user.""" df = df.copy() top_user = df["user_id"].value_counts().index[0] mask = pick_rows(df, 0.9, rng) df.loc[mask, "user_id"] = top_user return df def _impossible_geography(df, rate, rng): """Assign events to same user from two distant locations within 1 minute.""" df = df.copy() if len(df) < 2: return df idx1, idx2 = rng.choice(len(df), size=2, replace=False) shared_user = str(uuid.uuid4()) shared_ts = df["timestamp"].dropna().iloc[0] df.loc[idx1, "user_id"] = shared_user df.loc[idx1, "country_code"] = "US" df.loc[idx1, "timestamp"] = shared_ts df.loc[idx2, "user_id"] = shared_user df.loc[idx2, "country_code"] = "AU" df.loc[idx2, "timestamp"] = shared_ts + timedelta(seconds=30) return df def _duplicate_sessions_same_time(df, rate, rng): """Same session_id appears with two different device_types at the same time.""" df = df.copy() if df["session_id"].dropna().empty: return df shared_session = df["session_id"].dropna().iloc[0] ts = df["timestamp"].dropna().iloc[0] # Find two rows in this session and give them different devices at same time session_rows = df[df["session_id"] == shared_session].index[:2] if len(session_rows) >= 2: df.loc[session_rows[0], "device_type"] = "desktop" df.loc[session_rows[1], "device_type"] = "mobile" df.loc[session_rows[0], "timestamp"] = ts df.loc[session_rows[1], "timestamp"] = ts return df ISSUES = [ Issue("bot_traffic", "Bot traffic (1000 events/minute from one user)", "statistical", "Single user generates 1000 click events in 60 seconds", "critical", 0.02, _bot_traffic), Issue("single_ip_all_events", "Many events from single IP address", "statistical", "10%+ of events share the exact same IP address", "high", 0.02, _single_ip_all_events), Issue("all_same_timestamp", "Many rows with identical timestamp", "statistical", "Large group of events all have exact same timestamp", "high", 0.02, _all_same_timestamp), Issue("zero_variance_column", "Zero-variance country_code (90% single value)", "statistical", "country_code is the same value for 90% of rows", "medium", 0.02, _zero_variance_column), Issue("sudden_spike", "Sudden event volume spike (500 events in 60s)", "statistical", "500 page_view events in a narrow 60-second window", "high", 0.02, _sudden_spike), Issue("missing_time_range", "6-hour data gap (events deleted)", "statistical", "All events in a 6-hour window are removed", "critical", 0.02, _missing_time_range), Issue("all_events_same_user", "90% of events from one user", "statistical", "Single user_id dominates 90% of event volume", "high", 0.02, _all_events_same_user), Issue("impossible_geography", "Same user in US and AU within 30 seconds", "statistical", "User generates events from two distant countries in rapid succession", "critical", 0.02, _impossible_geography), Issue("duplicate_sessions_same_time", "Same session on two devices simultaneously", "statistical", "Same session_id active on desktop and mobile at the same time", "high", 0.02, _duplicate_sessions_same_time), ]