#!/usr/bin/env python3 """Descriptive rate standardization for two periods with the same sources. Run: python3 traffic-mix-lab.py [CSV] [--self-test] The included CSV is invented teaching data, not Gumi analytics. A visit has at most one success: whether it clicked an article. No causal estimate, confidence interval, or significance test is provided. """ import argparse import csv import io import platform from fractions import Fraction from pathlib import Path HEADER = ["period", "source", "visits", "clicked_visits"] EXAMPLE = """period,source,visits,clicked_visits before,newsletter,100,50 before,search,400,40 after,newsletter,400,160 after,search,100,5 """ def parse_csv(stream): reader = csv.DictReader(stream) if reader.fieldnames != HEADER: raise ValueError("CSV columns must be: " + ",".join(HEADER)) periods = {"before": {}, "after": {}} for line, row in enumerate(reader, 2): period, source = row["period"], row["source"] if period not in periods or not source or None in row: raise ValueError(f"line {line}: invalid period, source, or column count") if source in periods[period]: raise ValueError(f"line {line}: duplicate period/source") try: visits, clicks = int(row["visits"]), int(row["clicked_visits"]) except (TypeError, ValueError) as exc: raise ValueError(f"line {line}: counts must be integers") from exc if visits <= 0 or not 0 <= clicks <= visits: raise ValueError(f"line {line}: visits must be positive; 0 <= clicks <= visits") periods[period][source] = (visits, clicks) if not periods["before"] or periods["before"].keys() != periods["after"].keys(): raise ValueError("both periods must have the same non-empty source set") return periods def analyze(periods): totals = {p: sum(v for v, c in sources.values()) for p, sources in periods.items()} rates = {p: {s: Fraction(c, v) for s, (v, c) in sources.items()} for p, sources in periods.items()} weights = {p: {s: Fraction(v, totals[p]) for s, (v, c) in sources.items()} for p, sources in periods.items()} overall = {p: Fraction(sum(c for v, c in sources.values()), totals[p]) for p, sources in periods.items()} sources = sorted(periods["before"]) fixed_after = sum(weights["before"][s] * rates["after"][s] for s in sources) within = sum(weights["before"][s] * (rates["after"][s] - rates["before"][s]) for s in sources) composition = sum((weights["after"][s] - weights["before"][s]) * rates["after"][s] for s in sources) if overall["after"] - overall["before"] != within + composition: raise AssertionError("decomposition does not reconcile") return rates, weights, overall, fixed_after, within, composition def self_test(): example = parse_csv(io.StringIO(EXAMPLE)) rates, weights, overall, fixed, within, composition = analyze(example) assert all(rates["after"][s] < rates["before"][s] for s in rates["before"]) assert overall == {"before": Fraction(18, 100), "after": Fraction(33, 100)} assert (fixed, within, composition) == (Fraction(12, 100), Fraction(-6, 100), Fraction(21, 100)) same_mix = EXAMPLE.replace("after,newsletter,400,160", "after,newsletter,100,40").replace( "after,search,100,5", "after,search,400,20") r = analyze(parse_csv(io.StringIO(same_mix))) assert r[2]["after"] == r[3] == Fraction(12, 100) and r[5] == 0 same_rates = EXAMPLE.replace("after,newsletter,400,160", "after,newsletter,400,200").replace( "after,search,100,5", "after,search,100,10") r = analyze(parse_csv(io.StringIO(same_rates))) assert r[2]["after"] == Fraction(42, 100) and r[3] == Fraction(18, 100) and r[4] == 0 scaled = {p: {s: (v * 10, c * 10) for s, (v, c) in rows.items()} for p, rows in example.items()} assert analyze(scaled) == analyze(example) invalid = { "zero denominator": EXAMPLE.replace("before,newsletter,100,50", "before,newsletter,0,0"), "negative count": EXAMPLE.replace("before,newsletter,100,50", "before,newsletter,100,-1"), "clicks exceed visits": EXAMPLE.replace("before,newsletter,100,50", "before,newsletter,100,101"), "missing source": EXAMPLE.replace("after,search,100,5\n", ""), "duplicate row": EXAMPLE + "after,search,100,5\n", "non-integer": EXAMPLE.replace("before,newsletter,100,50", "before,newsletter,100,2.5"), } for label, data in invalid.items(): try: parse_csv(io.StringIO(data)) except ValueError: print("rejected:", label) else: raise AssertionError("accepted invalid case: " + label) print("Checks passed: reversal, unchanged mix, unchanged rates, scaling, and invalid inputs.") def percent(value, signed=False): return format(float(value) * 100, "+.2f" if signed else ".2f") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("csv_path", nargs="?", type=Path, default=Path(__file__).with_name("traffic-mix-example.csv")) parser.add_argument("--self-test", action="store_true") options = parser.parse_args() try: with options.csv_path.open(encoding="utf-8", newline="") as stream: periods = parse_csv(stream) except (OSError, ValueError) as exc: parser.error(str(exc)) print(f"Python {platform.python_version()} on {platform.system()}") rates, weights, overall, fixed, within, composition = analyze(periods) print("period source visits clicked rate share") for period in ("before", "after"): for source in sorted(periods[period]): visits, clicks = periods[period][source] print(f"{period:<7} {source:<11} {visits:>6} {clicks:>8}" f" {percent(rates[period][source]):>7}% {percent(weights[period][source]):>7}%") print("overall: " + percent(overall["before"]) + "% -> " + percent(overall["after"]) + "%") print("fixed before mix: " + percent(overall["before"]) + "% -> " + percent(fixed) + "%") print("within at before weights: " + percent(within, True) + " pp") print("composition at after rates: " + percent(composition, True) + " pp") print("total change: " + percent(overall["after"] - overall["before"], True) + " pp") if options.self_test: self_test() if __name__ == "__main__": main()