From a474f0986b23d2d7758a9d0fd8d6e8c3d0c37f3b Mon Sep 17 00:00:00 2001 From: Catzalyn Date: Mon, 3 Aug 2026 13:01:22 -0700 Subject: [PATCH] feat: complete dscout interview aide coding round --- ai_data_eng_tech_round.py | 220 ++++++++++++++++++++++++++++++++++++-- 1 file changed, 212 insertions(+), 8 deletions(-) diff --git a/ai_data_eng_tech_round.py b/ai_data_eng_tech_round.py index 259299d..5d9a437 100644 --- a/ai_data_eng_tech_round.py +++ b/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.