#!/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()
