#!/usr/bin/env python3 """Validate and calculate a fixed fictional campaign experiment; stdlib only.""" import argparse import csv import json import math import sys from collections import defaultdict from datetime import date from pathlib import Path FIELDS = ["period", "start_date", "end_date", "campaign_id", "campaign", "channel", "spend_eur", "impressions", "clicks", "conversions", "revenue_eur"] COUNT_FIELDS = ["impressions", "clicks", "conversions"] MONEY_FIELDS = ["spend_eur", "revenue_eur"] def validate(path): errors, rows, seen, campaigns, periods = [], [], set(), {}, {} with path.open(newline="", encoding="utf-8") as handle: reader = csv.DictReader(handle) if reader.fieldnames != FIELDS: return [], ["CSV columns must exactly match the documented schema and order."] for line, raw in enumerate(reader, 2): row = dict(raw) before = len(errors) if None in raw or any(v is None or not str(v).strip() for v in raw.values()): errors.append(f"Line {line}: missing or extra fields.") continue for field in COUNT_FIELDS + MONEY_FIELDS: try: if field in COUNT_FIELDS: row[field] = int(raw[field]) else: row[field] = float(raw[field]) if not math.isfinite(row[field]) or row[field] < 0: raise ValueError except (ValueError, OverflowError): errors.append(f"Line {line}: {field} must be a finite non-negative {'integer' if field in COUNT_FIELDS else 'number'}.") if len(errors) != before: continue if row["spend_eur"] == 0: errors.append(f"Line {line}: spend_eur must be positive in this demo.") if not row["impressions"] >= row["clicks"] >= row["conversions"]: errors.append(f"Line {line}: counts must satisfy impressions >= clicks >= conversions.") try: start, end = date.fromisoformat(row["start_date"]), date.fromisoformat(row["end_date"]) if start > end: raise ValueError except ValueError: errors.append(f"Line {line}: invalid date range.") continue if row["period"] not in ("baseline", "comparison"): errors.append(f"Line {line}: period must be baseline or comparison.") key = (row["period"], row["campaign_id"]) if key in seen: errors.append(f"Line {line}: duplicate campaign-period pair.") seen.add(key) identity = (row["campaign"], row["channel"]) if row["campaign_id"] in campaigns and campaigns[row["campaign_id"]] != identity: errors.append(f"Line {line}: campaign identity changes across periods.") campaigns[row["campaign_id"]] = identity bounds = (row["start_date"], row["end_date"]) if row["period"] in periods and periods[row["period"]] != bounds: errors.append(f"Line {line}: inconsistent dates within period.") periods[row["period"]] = bounds rows.append(row) if len(rows) != 8 or len(campaigns) != 4 or set(periods) != {"baseline", "comparison"}: errors.append("Expected exactly 8 valid rows: 4 campaigns in each of 2 periods.") for period in ("baseline", "comparison"): if {r["campaign_id"] for r in rows if r["period"] == period} != set(campaigns): errors.append(f"Period {period}: every campaign must appear exactly once.") if set(periods) == {"baseline", "comparison"}: b0, b1 = map(date.fromisoformat, periods["baseline"]) c0, c1 = map(date.fromisoformat, periods["comparison"]) if (b1 - b0) != (c1 - c0) or c0 <= b1: errors.append("Periods must have equal duration, be chronological, and not overlap.") return rows, errors def ratio(numerator, denominator): return round(numerator / denominator, 6) if denominator else None def summarize(rows): totals = {field: sum(row[field] for row in rows) for field in MONEY_FIELDS + COUNT_FIELDS} totals.update( ctr_pct=ratio(totals["clicks"] * 100, totals["impressions"]), conversion_rate_pct=ratio(totals["conversions"] * 100, totals["clicks"]), cpc_eur=ratio(totals["spend_eur"], totals["clicks"]), cpa_eur=ratio(totals["spend_eur"], totals["conversions"]), roas=ratio(totals["revenue_eur"], totals["spend_eur"]), ) return totals def calculate(rows, input_name): by_period = {period: summarize([r for r in rows if r["period"] == period]) for period in ("baseline", "comparison")} changes = {} for metric in by_period["baseline"]: base, comp = by_period["baseline"][metric], by_period["comparison"][metric] changes[metric] = {"absolute": round(comp - base, 6) if base is not None and comp is not None else None, "relative_pct": ratio((comp - base) * 100, base) if base is not None and comp is not None else None} campaigns = [] for campaign_id in sorted({r["campaign_id"] for r in rows}): selected = [r for r in rows if r["campaign_id"] == campaign_id] metrics = {period: summarize([r for r in selected if r["period"] == period]) for period in ("baseline", "comparison")} campaigns.append({"campaign_id": campaign_id, "campaign": selected[0]["campaign"], "channel": selected[0]["channel"], "periods": metrics}) return {"demo": "Fictional data; deterministic Python calculations; no model inference", "input": input_name, "validation": {"status": "passed", "rows": len(rows), "periods": 2, "campaigns": len(campaigns)}, "periods": by_period, "changes": changes, "campaigns": campaigns, "notes": ["Ratios use summed counts, never an average of row ratios.", "Percent metrics are in percentage units; their absolute changes are percentage points.", "Undefined ratios are null; intermediate metrics are rounded to 6 decimals.", "Attributed revenue is fictional gross revenue; ROAS is not profit or causal lift."]} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("input", type=Path) parser.add_argument("--output", type=Path) args = parser.parse_args() try: rows, errors = validate(args.input) except (OSError, csv.Error, UnicodeError) as exc: errors = [str(exc)] if errors: print(json.dumps({"validation": {"status": "failed", "errors": errors}}, indent=2), file=sys.stderr) return 2 result = json.dumps(calculate(rows, args.input.name), indent=2, allow_nan=False) + "\n" if args.output: args.output.write_text(result, encoding="utf-8") else: print(result, end="") return 0 if __name__ == "__main__": raise SystemExit(main())