[JRODRIGUEZ] complete dscout interview aide coding round

ai_data_eng_tech_round.py

@@ -22,6 +22,7 @@
 
 import math
 from collections import Counter
+from datetime import datetime
 
 import pandas as pd
 import duckdb
@@ -100,16 +101,73 @@ def _tuples_match(a, b):
 #       valid_events   - clean records ready for Snowflake
 #       rejected_events - records that failed, each with a 'rejection_reason' field
 
+REQUIRED_FIELDS = ("event_id", "participant_id", "mission_id", "study_id", "timestamp")
+
+
 def process_events(raw_events: list[dict]) -> tuple[list[dict], list[dict]]:
     """
     Clean and validate raw mission completion events.
 
+    Each record is checked against every rule and *all* failures are collected,
+    so a record that breaks multiple rules reports all of them (easier to debug
+    than a single first-failure message). Deduplication is by event_id in arrival
+    order: the first sighting of an id is kept, later copies are rejected as
+    duplicates. Records with a null/missing event_id can't be deduped, so they're
+    never treated as duplicates of one another — they fail on the required-field
+    check instead.
+
     Returns:
         valid_events: list of clean records ready for Snowflake
         rejected_events: list of records that failed validation, each with a 'rejection_reason' field
     """
-    # YOUR CODE HERE
-    pass
+    valid_events: list[dict] = []
+    rejected_events: list[dict] = []
+    seen_event_ids: set = set()
+
+    for event in raw_events:
+        reasons: list[str] = []
+
+        # Deduplicate on event_id (only meaningful for non-null ids).
+        event_id = event.get("event_id")
+        if event_id is not None and event_id in seen_event_ids:
+            reasons.append("duplicate event_id")
+
+        # Required fields must be present and non-null.
+        for field in REQUIRED_FIELDS:
+            if event.get(field) is None:
+                reasons.append(f"missing {field}")
+
+        # Timestamp must be a parseable ISO 8601 datetime. Only checked when present
+        # (avoids a redundant reason when it's already flagged as missing above).
+        # datetime.fromisoformat() on Python 3.9/3.10 rejects a trailing 'Z', so we
+        # normalize it to an explicit UTC offset first.
+        ts = event.get("timestamp")
+        if ts is not None:
+            if not isinstance(ts, str):
+                reasons.append("invalid timestamp")
+            else:
+                try:
+                    datetime.fromisoformat(ts.replace("Z", "+00:00"))
+                except ValueError:
+                    reasons.append("invalid timestamp")
+
+        # Duration must be a strictly-positive integer. The bool guard matters
+        # because bool is a subclass of int (True would otherwise pass as 1).
+        duration = event.get("duration_seconds")
+        if not (isinstance(duration, int) and not isinstance(duration, bool) and duration > 0):
+            reasons.append("duration_seconds must be a positive integer")
+
+        # Record the id so later copies dedupe against it, whether or not this
+        # record itself is valid.
+        if event_id is not None:
+            seen_event_ids.add(event_id)
+
+        if reasons:
+            rejected_events.append({**event, "rejection_reason": "; ".join(reasons)})
+        else:
+            valid_events.append(dict(event))
+
+    return valid_events, rejected_events
 
 
 # ─────────────────────────────────────────────
@@ -173,6 +231,65 @@ def process_events(raw_events: list[dict]) -> tuple[list[dict], list[dict]]:
 print("All Part 1 checks passed!" if all(_p1_checks) else "Some Part 1 checks failed — see ✗ above.")
 
 
+# ─────────────────────────────────────────────
+# UNIT TESTS: process_events() edge cases
+# ─────────────────────────────────────────────
+# The self-checks above only exercise the one fixed raw_events dataset. These
+# assert-based tests cover cases that dataset doesn't reach: empty input, an
+# all-valid batch, multi-reason records, dedupe independent of validity, and the
+# bool-vs-int and null-event_id edges. Plain asserts keep this dependency-free
+# (no pytest needed — the exercise only installs duckdb + pandas).
+
+def run_unit_tests() -> None:
+    # Empty input -> two empty lists.
+    assert process_events([]) == ([], [])
+
+    # A fully clean batch rejects nothing.
+    clean = [
+        {"event_id": "e1", "participant_id": "p1", "mission_id": "m1", "study_id": "s1",
+         "event_type": "mission_complete", "timestamp": "2024-01-15T10:30:00Z", "duration_seconds": 100},
+    ]
+    valid, rejected = process_events(clean)
+    assert len(valid) == 1 and rejected == []
+
+    # A record failing multiple checks reports ALL reasons (bad timestamp + bad duration).
+    valid, rejected = process_events([
+        {"event_id": "e2", "participant_id": "p1", "mission_id": "m1", "study_id": "s1",
+         "event_type": "mission_complete", "timestamp": "not-a-timestamp", "duration_seconds": -5},
+    ])
+    assert valid == [] and len(rejected) == 1
+    reason = rejected[0]["rejection_reason"]
+    assert "invalid timestamp" in reason and "positive integer" in reason, reason
+
+    # Dedupe is by arrival order and independent of validity: first copy kept, second rejected.
+    dupe = clean[0]
+    valid, rejected = process_events([dupe, dict(dupe)])
+    assert len(valid) == 1 and len(rejected) == 1
+    assert "duplicate event_id" in rejected[0]["rejection_reason"]
+
+    # Two records with a null event_id are NOT duplicates of each other — each fails
+    # only on the missing-field check, not on dedupe.
+    null_id = {"event_id": None, "participant_id": "p1", "mission_id": "m1", "study_id": "s1",
+               "event_type": "mission_complete", "timestamp": "2024-01-15T10:30:00Z", "duration_seconds": 100}
+    valid, rejected = process_events([dict(null_id), dict(null_id)])
+    assert valid == [] and len(rejected) == 2
+    assert all("duplicate event_id" not in r["rejection_reason"] for r in rejected)
+    assert all("missing event_id" in r["rejection_reason"] for r in rejected)
+
+    # Boolean duration is rejected (bool is a subclass of int, so True must not pass as 1).
+    valid, rejected = process_events([{**clean[0], "event_id": "e3", "duration_seconds": True}])
+    assert valid == [] and len(rejected) == 1
+    assert "positive integer" in rejected[0]["rejection_reason"]
+
+
+print("\n── Unit tests: process_events() ──")
+try:
+    run_unit_tests()
+    print("  ✓ All process_events() unit tests passed!")
+except AssertionError as _e:
+    print(f"  ✗ Unit test failed: {_e}")
+
+
 # ─────────────────────────────────────────────
 # PART 2: SQL
 # ─────────────────────────────────────────────
@@ -197,19 +314,73 @@ def process_events(raw_events: list[dict]) -> tuple[list[dict], list[dict]]:
 con = duckdb.connect()
 con.register("mission_events", valid_events_df)
 
-# Q1
+# Q1 — Joins & Aggregations
+# Each row is one mission_complete event, so COUNT(*) = number of completions per
+# study (a single mission_id can be completed by many participants, which is why we
+# count events, not DISTINCT mission_id). HAVING drops studies with 2 or fewer.
 q1 = """
--- YOUR QUERY HERE
+SELECT
+    study_id,
+    COUNT(*)              AS completed_missions,
+    AVG(duration_seconds) AS avg_duration_seconds
+FROM mission_events
+GROUP BY study_id
+HAVING COUNT(*) > 2
+ORDER BY study_id
 """
 
-# Q2
+# Q2 — Window Functions
+# Rank each participant's missions by duration (longest first) within each study,
+# so the partition is (participant_id, study_id). RANK() leaves gaps on ties; the
+# data has no ties, so RANK/DENSE_RANK/ROW_NUMBER coincide here.
 q2 = """
--- YOUR QUERY HERE
+SELECT
+    participant_id,
+    study_id,
+    mission_id,
+    duration_seconds,
+    RANK() OVER (
+        PARTITION BY participant_id, study_id
+        ORDER BY duration_seconds DESC
+    ) AS duration_rank
+FROM mission_events
+ORDER BY participant_id, study_id, duration_rank
 """
 
-# Q3
+# Q3 — Anomaly Detection
+# Compare each participant's average duration to their study's mean, measured in
+# study standard deviations. STDDEV() is sample stddev in DuckDB; NULLIF guards
+# against single-row studies where stddev is 0/NULL (avoids divide-by-zero). Only
+# rows more than 2 stddev out are returned.
 q3 = """
--- YOUR QUERY HERE
+WITH participant_avg AS (
+    SELECT
+        participant_id,
+        study_id,
+        AVG(duration_seconds) AS avg_duration
+    FROM mission_events
+    GROUP BY participant_id, study_id
+),
+study_stats AS (
+    SELECT
+        study_id,
+        AVG(duration_seconds)    AS study_mean,
+        STDDEV(duration_seconds) AS study_stddev
+    FROM mission_events
+    GROUP BY study_id
+)
+SELECT
+    pa.participant_id,
+    pa.study_id,
+    pa.avg_duration,
+    ss.study_mean,
+    ss.study_stddev,
+    ABS(pa.avg_duration - ss.study_mean) / NULLIF(ss.study_stddev, 0) AS stddevs_away
+FROM participant_avg pa
+JOIN study_stats ss USING (study_id)
+WHERE ss.study_stddev > 0
+  AND ABS(pa.avg_duration - ss.study_mean) / ss.study_stddev > 2
+ORDER BY stddevs_away DESC
 """
 
 q1_df = con.execute(q1).df()
@@ -258,3 +429,36 @@ def process_events(raw_events: list[dict]) -> tuple[list[dict], list[dict]]:
 )
 
 print("All Part 2 checks passed!" if _q1_ok and _q2_ok and _q3_ok else "Some Part 2 checks failed — see ✗ above.")
+
+
+# ─────────────────────────────────────────────
+# NOTES: tradeoffs & assumptions
+# ─────────────────────────────────────────────
+# Part 1 — process_events()
+#   - Reject reasons are COLLECTED, not short-circuited: a record is run through
+#     every check and all failures are joined into one rejection_reason (e.g.
+#     ghi789 -> "invalid timestamp; duration_seconds must be a positive integer").
+#     Costs a little extra work per bad record but makes pipeline failures fully
+#     visible/debuggable instead of surfacing one problem at a time.
+#   - Dedupe is first-wins in arrival order, keyed on event_id, and independent of
+#     validity. Null/missing event_ids are never deduped against each other — they
+#     fail the required-field check instead, so two null-id records don't collapse.
+#   - Timestamps: 'Z' is normalized to '+00:00' before datetime.fromisoformat(),
+#     which on Python 3.9/3.10 rejects a bare trailing 'Z'. Only ISO 8601 is
+#     accepted — no attempt to coerce other formats (fail loudly rather than guess).
+#   - duration_seconds must be a strictly-positive int; bool is explicitly excluded
+#     since it subclasses int. 0 and floats are rejected (tighten/loosen if the
+#     real source emits fractional seconds).
+#   - Original records are shallow-copied into the output lists so the returned
+#     data doesn't alias the raw input.
+#
+# Part 2 — SQL
+#   - Q1 counts completions with COUNT(*), not COUNT(DISTINCT mission_id): each row
+#     is one participant completing a mission, and the same mission_id can be
+#     completed by several participants (s2.m3 x3). Counting distinct missions would
+#     answer a different question and, on this data, drop every study below the
+#     >2 threshold.
+#   - Q3 uses DuckDB's STDDEV (sample stddev). With this tiny dataset the result is
+#     empty under both sample and population stddev (every z-score < 1.5). NULLIF
+#     guards single-row studies (stddev 0/NULL) against divide-by-zero. At real
+#     scale, decide deliberately between sample and population stddev.