#!/usr/bin/env python3 """Rebuild the fictional Northstar Dispatch fixtures. No network or real data. Run from anywhere: python tools/generate_data.py The seeds and generating assumptions are fixed in this source; no rerolling to obtain a desired p-value. Use Python 3.12 for the tested reproduction route. """ from __future__ import annotations import csv from collections import Counter import json import math from pathlib import Path import random import re ROOT = Path(__file__).resolve().parents[1] SHIPMENT_SEED = 20261014 TRIAL_SEED = 20261015 SHIPMENT_FIELDS = ["shipment_id", "week", "hub", "distance_km", "service", "promised_minutes", "actual_minutes", "late", "satisfaction"] TRIAL_FIELDS = ["person_id", "group", "duration_minutes"] def generate_shipments() -> list[dict]: """240 observational teaching records; hub is associated with job mix.""" rng = random.Random(SHIPMENT_SEED) rows = [] # Shared conditions create possible within-week dependence. Do not use # an independent-row t test to infer a hub effect from these observations. week_conditions = [-2.0, 1.5, 0.0, 4.0, -1.5, 2.5, 0.5, 3.0] for week in range(1, 9): for hub in ("East", "West"): services = (["priority"] * (5 if hub == "East" else 10) + ["standard"] * (10 if hub == "East" else 5)) rng.shuffle(services) for service in services: distance = round(rng.uniform(20, 55) if hub == "East" else rng.uniform(5, 35), 1) priority = service == "priority" # Neither equation has a direct hub term. Observers would not # normally know the real process or all relevant confounders. promised = round(25 + 1.7 * distance - 8 * priority, 1) duration = (15 + 1.7 * distance - 5 * priority + week_conditions[week - 1] + rng.gauss(0, 7)) if rng.random() < 0.08: duration += rng.uniform(15, 30) actual = round(max(5.0, duration), 1) late = int(actual > promised) # equality is on time latent_rating = round(4.4 - 1.4 * late + rng.gauss(0, 0.7)) rating = min(5, max(1, latent_rating)) # Late deliveries are less likely to receive a response. satisfaction = None if rng.random() < (0.45 if late else 0.10) else rating rows.append(dict(shipment_id=f"S{len(rows) + 1:04d}", week=week, hub=hub, distance_km=distance, service=service, promised_minutes=promised, actual_minutes=actual, late=late, satisfaction=satisfaction)) return rows def generate_trial() -> list[dict]: """Fictional independent-person randomized fixture, unrelated to shipments.""" rng = random.Random(TRIAL_SEED) groups = ["control"] * 20 + ["checklist"] * 20 rng.shuffle(groups) # balanced random assignment before generating outcomes rows = [] for i, group in enumerate(groups, 1): duration = round(24.0 - 3.0 * (group == "checklist") + rng.gauss(0, 4.2), 2) if duration <= 0: raise ValueError("Fixture generated an impossible duration; revise documented model.") rows.append(dict(person_id=f"P{i:03d}", group=group, duration_minutes=duration)) return rows def to_text_row(row: dict) -> dict[str, str]: return {field: "" if value is None else str(value) for field, value in row.items()} def corrupt_export(rows: list[dict]) -> list[dict[str, str]]: """248-row export: 240 recoverable records, 3 duplicate copies, 5 invalid extras.""" raw = [to_text_row(row) for row in rows] # Six formatting-only changes, across three records. Values remain knowable. raw[0]["hub"] = " east " raw[0]["service"] = "PRIORITY" if raw[0]["service"] == "priority" else "STANDARD" raw[1]["shipment_id"] = " " + raw[1]["shipment_id"] + " " raw[1]["distance_km"] = " " + raw[1]["distance_km"] + " " raw[2]["hub"] = "WEST" if raw[2]["hub"] == "West" else "EAST" raw[2]["service"] = " " + raw[2]["service"] + " " # Two stale derived flags; recompute using the documented deadline rule. for index in (9, 19): raw[index]["late"] = str(1 - int(raw[index]["late"])) raw.extend(dict(raw[index]) for index in (39, 79, 119)) # Invalid appended records are not repairable from the export; quarantine. corruptions = [("distance_km", "-4"), ("actual_minutes", "unknown"), ("hub", "North"), ("promised_minutes", ""), ("satisfaction", "7")] for i, (field, value) in enumerate(corruptions, 1): invalid = to_text_row(rows[150 + i]) invalid["shipment_id"] = f"X{i:04d}" invalid[field] = value raw.append(invalid) return raw def clean_shipment_rows(raw_rows: list[dict[str, str]]) -> tuple[list[dict], dict]: """Normalize, validate, deduplicate, derive late, and log every decision. Invalid non-missing satisfaction is quarantined, not silently replaced by a missing rating. Conflicting duplicates quarantine the entire ID; only identical canonical duplicates can be safely collapsed. """ audit = {"input_rows": len(raw_rows), "formatting_cells_normalized": 0, "late_flags_recomputed": 0, "exact_duplicate_rows_removed": 0, "invalid_rows_quarantined": 0, "conflicting_rows_quarantined": 0, "conflicting_ids_quarantined": [], "quarantine": [], "row_dispositions": []} # Delay duplicate decisions until every validated source row for each ID is # known. A later conflict must reclassify the first row AND earlier copies. valid_groups: dict[str, list[tuple[int, dict]]] = {} candidates: dict[str, dict] = {} reasons = Counter() for source_line, source in enumerate(raw_rows, start=2): row = {field: str(source.get(field, "") or "").strip() for field in SHIPMENT_FIELDS} row["hub"] = row["hub"].title() row["service"] = row["service"].lower() audit["formatting_cells_normalized"] += sum(row[f] != str(source.get(f, "") or "") for f in SHIPMENT_FIELDS) reason = None parsed = {} try: if not re.fullmatch(r"[SX]\d{4}", row["shipment_id"]): raise ValueError("invalid shipment_id") parsed["shipment_id"] = row["shipment_id"] if not re.fullmatch(r"[1-8]", row["week"]): raise ValueError("week must be an integer 1–8") parsed["week"] = int(row["week"]) if row["hub"] not in ("East", "West"): raise ValueError("hub outside East/West") parsed["hub"] = row["hub"] for field in ("distance_km", "promised_minutes", "actual_minutes"): try: value = float(row[field]) except ValueError: raise ValueError(f"{field} is not numeric") from None if not math.isfinite(value) or value <= 0: raise ValueError(f"{field} must be finite and positive") parsed[field] = value if row["service"] not in ("standard", "priority"): raise ValueError("service outside standard/priority") parsed["service"] = row["service"] if row["satisfaction"] == "": parsed["satisfaction"] = None elif re.fullmatch(r"[1-5]", row["satisfaction"]): parsed["satisfaction"] = int(row["satisfaction"]) else: raise ValueError("satisfaction must be 1–5 or blank") parsed["late"] = int(parsed["actual_minutes"] > parsed["promised_minutes"]) if row["late"] != str(parsed["late"]): audit["late_flags_recomputed"] += 1 except ValueError as error: reason = str(error) if reason: audit["invalid_rows_quarantined"] += 1 reasons[reason] += 1 audit["quarantine"].append({"csv_line": source_line, "shipment_id": row["shipment_id"], "reason": reason}) audit["row_dispositions"].append({"csv_line": source_line, "shipment_id": row["shipment_id"], "disposition": "invalid_quarantined"}) continue valid_groups.setdefault(parsed["shipment_id"], []).append((source_line, parsed)) for key, records in valid_groups.items(): first = records[0][1] conflicting = any(parsed != first for _, parsed in records[1:]) if conflicting: audit["conflicting_ids_quarantined"].append(key) audit["conflicting_rows_quarantined"] += len(records) else: candidates[key] = first audit["exact_duplicate_rows_removed"] += len(records) - 1 for position, (source_line, _) in enumerate(records): if conflicting: disposition = "conflict_quarantined" audit["quarantine"].append({"csv_line": source_line, "shipment_id": key, "reason": "conflicting duplicate ID; all validated records for ID removed"}) else: disposition = "accepted" if position == 0 else "exact_duplicate_removed" audit["row_dispositions"].append({"csv_line": source_line, "shipment_id": key, "disposition": disposition}) clean = [candidates[key] for key in sorted(candidates)] audit["conflicting_ids_quarantined"].sort() audit["quarantine"].sort(key=lambda row: row["csv_line"]) audit["row_dispositions"].sort(key=lambda row: row["csv_line"]) audit["invalid_reasons"] = dict(sorted(reasons.items())) audit["output_rows"] = len(clean) audit["accepted_rows"] = len(clean) audit["total_rows_quarantined"] = (audit["invalid_rows_quarantined"] + audit["conflicting_rows_quarantined"]) if (audit["input_rows"] != audit["accepted_rows"] + audit["total_rows_quarantined"] + audit["exact_duplicate_rows_removed"]): raise RuntimeError("Cleaning row dispositions do not reconcile.") return clean, audit def write_csv(path: Path, rows: list[dict], fields: list[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fields, lineterminator="\n") writer.writeheader() writer.writerows(rows) def main() -> None: base = generate_shipments() raw = corrupt_export(base) clean, audit = clean_shipment_rows(raw) if clean != base: raise RuntimeError("Cleaning failed to recover the intended observed base.") write_csv(ROOT / "data/shipments_raw.csv", raw, SHIPMENT_FIELDS) write_csv(ROOT / "data/shipments_clean.csv", clean, SHIPMENT_FIELDS) write_csv(ROOT / "data/checklist_trial.csv", generate_trial(), TRIAL_FIELDS) (ROOT / "outputs").mkdir(exist_ok=True) (ROOT / "outputs/cleaning_audit.json").write_text(json.dumps(audit, indent=2) + "\n", encoding="utf-8") print(f"Generated {len(raw)} raw rows → {len(clean)} clean shipments; 40 fictional trial tasks.") if __name__ == "__main__": main()