#!/usr/bin/env python3 """Cost per accepted record for an enrichment waterfall, from a run ledger. Usage: python3 enrichment_cost.py LEDGER.csv [--ceiling USD] [--warn-ratio 3] [--min-sample 200] [--accepted-status valid,deliverable,verified] [--fixed-usd 0] [--review-minutes 0] [--review-rate 0] [--billed BILLED.csv] [--variance 0.10] [--json] LEDGER.csv has one row per provider attempt, with the columns of assets/enrichment-ledger-template.csv. references/enrichment-and-cost.md explains each column and outcome value. Stage 0 rows mark records excluded (or reused) before any paid step. BILLED.csv has provider, unit_type and billed_units, copied from each vendor's usage export or invoice for the run. Validator usage goes on a row with provider "validation", whatever the vendor, because the ledger models all validation units as one line. For each data point it reports eligible records, then per stage: records entered, candidates returned, outcomes, cost, cost per candidate, marginal cost per accepted record and cumulative coverage. Then the blended cost per accepted record, money spent on excluded records or after acceptance, and a fully loaded run total. Checks W04, W06, W07, W08 and W09 (references/rules.md). The sample check counts, per segment, the records entering each stage, because later stages see only what earlier stages left unresolved. --fixed-usd is the run's share of subscriptions and platform meters not in the ledger. --review-minutes is reviewer time per accepted record, priced at --review-rate USD per hour. Standard library only. Reads files and prints a report; writes nothing. Exit status: 0 no FAIL, 1 at least one FAIL, 2 bad input. """ import argparse import csv import json import sys from collections import Counter, OrderedDict OUTCOMES = ("excluded", "reused", "no_result", "candidate_accepted", "candidate_rejected", "candidate_held", "error", "pending") CANDIDATES = {"candidate_accepted", "candidate_rejected", "candidate_held"} SHORT = {"candidate_accepted": "accepted", "candidate_held": "held", "candidate_rejected": "rejected", "no_result": "no_result", "error": "error", "pending": "pending", "reused": "reused"} FINAL_ORDER = ("held", "rejected", "no_result", "error", "pending") REQUIRED = ("record_id", "data_point", "stage", "provider", "attempt", "outcome", "units", "unit_price_usd") CHECKS = OrderedDict([("W04", "exclusion-before-spend"), ("W06", "waterfall-stop-and-states"), ("W07", "validation-policy"), ("W08", "marginal-cost"), ("W09", "caps-and-reconciliation")]) SEVERITY_ORDER = {"FAIL": 0, "WARN": 1, "INFO": 2} class InputError(Exception): pass def text(value): return str(value).strip() if value is not None else "" def usd(value): return "%.4f" % value def as_float(value, line, column): try: return float(text(value) or 0) except ValueError: raise InputError("row %d: %s must be a number, got %r" % (line, column, value)) def load_ledger(path): try: with open(path, newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) columns = [text(c) for c in (reader.fieldnames or [])] missing = [c for c in REQUIRED if c not in columns] if missing: raise InputError("%s is missing columns: %s" % (path, ", ".join(missing))) rows = [] for line, raw in enumerate(reader, 2): row = {text(k): text(v) for k, v in raw.items() if k is not None} row["outcome"] = row["outcome"].lower() if row["outcome"] not in OUTCOMES: raise InputError("row %d: outcome %r is not one of %s" % (line, row["outcome"], ", ".join(OUTCOMES))) if not row["record_id"]: raise InputError("row %d: record_id is empty" % line) row["_line"] = line row["_stage"] = int(as_float(row["stage"], line, "stage")) row["_attempt"] = int(as_float(row["attempt"] or 1, line, "attempt")) row["_cost"] = (as_float(row["units"], line, "units") * as_float(row["unit_price_usd"], line, "unit_price_usd") + as_float(row.get("validation_units"), line, "validation_units") * as_float(row.get("validation_price_usd"), line, "validation_price_usd")) row["_units"] = as_float(row["units"], line, "units") row["_vunits"] = as_float(row.get("validation_units"), line, "validation_units") # A blank price would make a charged attempt free and hide a stage above the ceiling. for used, units, price in ((row["_units"], "units", "unit_price_usd"), (row["_vunits"], "validation_units", "validation_price_usd")): if used and not text(row.get(price)): raise InputError("row %d: %s is set but %s is blank; enter the price, 0 when the unit is free" % (line, units, price)) rows.append(row) except (OSError, UnicodeDecodeError, csv.Error) as exc: raise InputError("cannot read %s as a UTF-8 CSV: %s" % (path, exc)) if not rows: raise InputError("%s has no rows" % path) return rows def load_billed(path): try: with open(path, newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) columns = [text(c) for c in (reader.fieldnames or [])] if not {"provider", "billed_units"} <= set(columns): raise InputError("%s needs provider, unit_type and billed_units columns" % path) billed = OrderedDict() for line, raw in enumerate(reader, 2): row = {text(k): text(v) for k, v in raw.items() if k is not None} key = (row["provider"], row.get("unit_type", "")) billed[key] = billed.get(key, 0.0) + as_float(row["billed_units"], line, "billed_units") except (OSError, UnicodeDecodeError, csv.Error) as exc: raise InputError("cannot read %s as a UTF-8 CSV: %s" % (path, exc)) return billed def stage_label(rows): providers = [] for row in rows: if row["provider"] and row["provider"] not in providers: providers.append(row["provider"]) return "+".join(providers) or "unnamed provider" def analyse_point(point, rows, args, findings): records = OrderedDict() for row in rows: records.setdefault(row["record_id"], []).append(row) for rid in records: records[rid].sort(key=lambda r: (r["_stage"], r["_attempt"], r["_line"])) segment_of = {rid: records[rid][0].get("segment") or "(no segment)" for rid in records} entered_by_segment = {} excluded = [rid for rid, rs in records.items() if any(r["outcome"] == "excluded" for r in rs)] excluded_set = set(excluded) eligible = [rid for rid in records if rid not in excluded_set] accepted_at = {} for rid in records: resolved = None for r in records[rid]: r["_excluded"] = rid in excluded_set r["_after"] = resolved is not None and not r["_excluded"] if resolved is None and r["outcome"] in ("candidate_accepted", "reused"): resolved = r accepted_at[rid] = r stage_numbers = sorted({r["_stage"] for r in rows if r["_stage"] > 0}) stages = [] cumulative = 0 earlier_cost = 0.0 earlier_accepted = 0 for number in stage_numbers: srows = [r for rid in records for r in records[rid] if r["_stage"] == number] live = [r for r in srows if not r["_excluded"] and not r["_after"]] last = OrderedDict() for r in live: last[r["record_id"]] = r outcomes = Counter(SHORT[r["outcome"]] for r in last.values()) entered_by_segment[number] = Counter(segment_of[rid] for rid in last) cost = sum(r["_cost"] for r in srows) candidates = sum(1 for r in srows if r["outcome"] in CANDIDATES) accepted = outcomes["accepted"] cumulative += accepted stage = OrderedDict([ ("stage", number), ("provider", stage_label(srows)), ("entered", len(last)), ("after_acceptance", len({r["record_id"] for r in srows if r["_after"]})), ("candidates", candidates), ("candidates_on_excluded", sum(1 for r in srows if r["outcome"] in CANDIDATES and r["_excluded"])), ("candidates_after_acceptance", sum(1 for r in srows if r["outcome"] in CANDIDATES and r["_after"])), ("accepted", accepted), ("held", outcomes["held"]), ("rejected", outcomes["rejected"]), ("no_result", outcomes["no_result"]), ("error", outcomes["error"]), ("pending", outcomes["pending"]), ("cost_usd", round(cost, 6)), ("cost_per_candidate_usd", round(cost / candidates, 6) if candidates else None), ("marginal_usd", round(cost / accepted, 6) if accepted else None), ("earlier_blended_usd", round(earlier_cost / earlier_accepted, 6) if earlier_accepted else None), ("coverage", round(cumulative / len(eligible), 6) if eligible else None), ]) stages.append(stage) if accepted and stage["earlier_blended_usd"]: ratio = stage["marginal_usd"] / stage["earlier_blended_usd"] stage["ratio_to_earlier"] = round(ratio, 2) if args.ceiling is not None and accepted and stage["marginal_usd"] > args.ceiling: findings.append(("FAIL", "W08", "%s stage %d (%s) costs %s USD per accepted record, above the %s USD ceiling" % (point, number, stage["provider"], usd(stage["marginal_usd"]), "%.2f" % args.ceiling))) elif args.ceiling is None and accepted and stage.get("ratio_to_earlier", 0) > args.warn_ratio: findings.append(("WARN", "W08", "%s stage %d (%s) costs %s USD per accepted record, %.1f times the %s USD " "blended cost of earlier stages (limit %gx); set a ceiling or drop the stage" % (point, number, stage["provider"], usd(stage["marginal_usd"]), stage["ratio_to_earlier"], usd(stage["earlier_blended_usd"]), args.warn_ratio))) if cost > 0 and not accepted: findings.append(("WARN", "W08", "%s stage %d (%s) spent %s USD and accepted no record" % (point, number, stage["provider"], usd(cost)))) earlier_cost += cost earlier_accepted += accepted finals = Counter() for rid in eligible: rs = [r for r in records[rid] if not r["_after"]] outs = [r["outcome"] for r in rs] paid = [r for r in rs if r["_stage"] > 0] if "candidate_accepted" in outs: finals["accepted"] += 1 elif "reused" in outs: finals["reused"] += 1 elif not paid: finals["not_attempted"] += 1 elif paid[-1]["outcome"] == "pending": finals["pending"] += 1 else: seen = {SHORT[r["outcome"]] for r in paid} finals[next(state for state in FINAL_ORDER if state in seen)] += 1 for rid in excluded: paid = [r for r in records[rid] if r["_stage"] > 0] spent = sum(r["_cost"] for r in paid) reason = next((r.get("excluded_reason") for r in records[rid] if r["outcome"] == "excluded"), "") or "no reason" if spent > 0: findings.append(("FAIL", "W04", "%s %s was excluded at stage 0 (%s) but has %d paid attempt%s at stage %s " "costing %s USD" % (rid, point, reason, len(paid), "" if len(paid) == 1 else "s", ", ".join(str(s) for s in sorted({r["_stage"] for r in paid})), usd(spent)))) elif paid: findings.append(("WARN", "W04", "%s %s was excluded at stage 0 (%s) but was still attempted %d time%s " "(no cost recorded)" % (rid, point, reason, len(paid), "" if len(paid) == 1 else "s"))) for rid in eligible: after = [r for r in records[rid] if r["_after"] and r["_stage"] > 0] accepted_rows = [r for r in records[rid] if r["outcome"] == "candidate_accepted"] if after: first = accepted_at[rid] message = "%s %s was attempted again at stage %d (%s) after it was %s at stage %d" % ( rid, point, after[0]["_stage"], after[0]["provider"] or "unnamed provider", "accepted" if first["outcome"] == "candidate_accepted" else "reused", first["_stage"]) if len(accepted_rows) > 1: message += ", and has %d accepted candidates" % len(accepted_rows) findings.append(("FAIL", "W06", message)) live = [r for r in records[rid] if not r["_after"] and r["_stage"] > 0] for number in sorted({r["_stage"] for r in live}): at_stage = [r for r in live if r["_stage"] == number] later = [r for r in live if r["_stage"] > number] if at_stage[-1]["outcome"] == "error" and later: attempts = len(at_stage) findings.append(("WARN", "W06", "%s %s ended stage %d (%s) in error after %d attempt%s and moved to stage " "%d without a successful retry" % (rid, point, number, at_stage[-1]["provider"], attempts, "" if attempts == 1 else "s", later[0]["_stage"]))) elif at_stage[-1]["outcome"] == "pending" and later: # Pending is not failed: Apollo's waterfall, for one, delivers results later by webhook. findings.append(("WARN", "W06", "%s %s was still pending at stage %d (%s) and moved to stage %d; wait for " "the webhook or poll before buying the next source" % (rid, point, number, at_stage[-1]["provider"] or "unnamed provider", later[0]["_stage"]))) if finals["pending"]: findings.append(("INFO", "W06", "%s: %d record%s still pending; not counted as no result" % (point, finals["pending"], "" if finals["pending"] == 1 else "s"))) accepted_status = {s.strip().lower() for s in args.accepted_status.split(",") if s.strip()} checked_validation = 0 for rid in eligible: row = accepted_at.get(rid) if not row or row["outcome"] != "candidate_accepted": continue status = text(row.get("validation_status")).lower() if "email" not in point.lower() and not status: continue checked_validation += 1 if status not in accepted_status: findings.append(("FAIL", "W07", "%s %s was accepted at stage %d with validation status %s; automated " "sends need %s" % (rid, point, row["_stage"], status or "blank", ", ".join(sorted(accepted_status))))) segments = OrderedDict((segment_of[rid], True) for rid in eligible) for segment in segments: entered = [(number, entered_by_segment[number][segment]) for number in stage_numbers] if any(count < args.min_sample for _, count in entered): findings.append(("WARN", "W08", "segment %s (%s): %s, below the %d records a provider ordering decision " "needs at each stage" % (segment, point, ", ".join("stage %d entered %d" % pair for pair in entered), args.min_sample))) variable = sum(r["_cost"] for r in rows) waste_excluded = OrderedDict((rid, round(sum(r["_cost"] for r in records[rid] if r["_stage"] > 0), 6)) for rid in excluded) waste_after = OrderedDict() for rid in eligible: spent = sum(r["_cost"] for r in records[rid] if r["_after"]) if spent > 0: waste_after[rid] = round(spent, 6) return OrderedDict([ ("data_point", point), ("records", len(records)), ("excluded", len(excluded)), ("eligible", len(eligible)), ("stages", stages), ("final", OrderedDict((k, finals[k]) for k in ("accepted", "reused", "held", "rejected", "no_result", "pending", "error", "not_attempted"))), ("variable_usd", round(variable, 6)), ("blended_usd", round(variable / finals["accepted"], 6) if finals["accepted"] else None), ("waste_excluded_usd", {k: v for k, v in waste_excluded.items() if v > 0}), ("waste_after_acceptance_usd", dict(waste_after)), ("validated_accepted", checked_validation), ]) def reconcile(rows, billed, args, findings): modeled = OrderedDict() for row in rows: if row["provider"] and row["_units"]: key = (row["provider"], row.get("unit_type", "")) modeled[key] = modeled.get(key, 0.0) + row["_units"] validation = sum(r["_vunits"] for r in rows) if validation: modeled[("validation", "")] = validation lines = [] for key, units in modeled.items(): match = billed.get(key) if match is None: same = [v for (p, u), v in billed.items() if p == key[0]] match = same[0] if len(same) == 1 else None name = "%s %s" % (key[0], key[1]) if key[1] else key[0] if match is None: lines.append(OrderedDict([("provider", key[0]), ("unit_type", key[1]), ("modeled", units), ("billed", None)])) hint = "; put validator usage on a billed row with provider validation" if key[0] == "validation" else "" findings.append(("INFO", "W09", "%s: modeled %g units, no billed figure supplied%s" % (name, units, hint))) continue variance = (match - units) / units if units else None lines.append(OrderedDict([("provider", key[0]), ("unit_type", key[1]), ("modeled", units), ("billed", match), ("variance", round(variance, 4) if variance is not None else None)])) if variance is not None and abs(variance) > args.variance: findings.append(("WARN", "W09", "%s billed %g vs modeled %g (%+.1f%%), above the %g%% variance; find the " "reruns, charged misses or retries before scaling" % (name, match, units, variance * 100, args.variance * 100))) for (provider, unit_type), units in billed.items(): if (provider, unit_type) not in modeled and not any(p == provider for p, _ in modeled): hint = "; if this is the validator, name its billed row validation" if validation else "" findings.append(("WARN", "W09", "%s %s billed %g units that the ledger does not model%s" % (provider, unit_type, units, hint))) return lines def main(): parser = argparse.ArgumentParser(description="Cost per accepted record from an enrichment run ledger.") parser.add_argument("ledger") parser.add_argument("--ceiling", type=float) parser.add_argument("--warn-ratio", type=float, default=3.0) parser.add_argument("--min-sample", type=int, default=200) parser.add_argument("--accepted-status", default="valid,deliverable,verified") parser.add_argument("--fixed-usd", type=float, default=0.0) parser.add_argument("--review-minutes", type=float, default=0.0) parser.add_argument("--review-rate", type=float, default=0.0) parser.add_argument("--billed") parser.add_argument("--variance", type=float, default=0.10) parser.add_argument("--json", action="store_true") args = parser.parse_args() try: rows = load_ledger(args.ledger) billed = load_billed(args.billed) if args.billed else None except InputError as exc: print("error: %s" % exc, file=sys.stderr) return 2 findings = [] points = OrderedDict() for row in rows: points.setdefault(row["data_point"] or "(no data point)", []).append(row) results = [analyse_point(point, prow, args, findings) for point, prow in points.items()] reconciliation = reconcile(rows, billed, args, findings) if billed is not None else None accepted_total = sum(p["final"]["accepted"] for p in results) variable_total = sum(p["variable_usd"] for p in results) review_total = args.review_minutes * accepted_total * args.review_rate / 60.0 loaded_total = variable_total + args.fixed_usd + review_total validated = sum(p["validated_accepted"] for p in results) flagged = {c for s, c, m in findings if s in ("FAIL", "WARN")} passes = [] excluded_total = sum(p["excluded"] for p in results) paid_stages = sum(len(p["stages"]) for p in results) if "W04" not in flagged: if excluded_total: passes.append(("PASS", "W04", "no paid attempt on the %d excluded records" % excluded_total)) else: findings.append(("INFO", "W04", "not checked: the ledger has no stage 0 excluded rows, so nothing shows the " "exclusions ran before spend; add them from the exclusion step and rerun")) if "W06" not in flagged: if paid_stages: passes.append(("PASS", "W06", "no attempt after acceptance, and no error or pending result fell through to " "the next provider")) else: findings.append(("INFO", "W06", "not checked: the ledger has no paid stage")) if "W07" not in flagged: if validated: passes.append(("PASS", "W07", "%d accepted candidates checked, all with an accepted validation status (%s)" % (validated, args.accepted_status))) else: findings.append(("INFO", "W07", "not checked: no accepted candidate carries a validation status to read")) if "W08" not in flagged: if paid_stages: passes.append(("PASS", "W08", "every stage within the ceiling or ratio, and at least %d records entering " "each stage in every segment" % args.min_sample)) else: findings.append(("INFO", "W08", "not checked: the ledger has no paid stage")) if billed is None: findings.append(("INFO", "W09", "billed usage not checked; pass --billed with each vendor's usage for this run")) elif "W09" not in flagged: passes.append(("PASS", "W09", "billed usage within %g%% of modeled for every provider" % (args.variance * 100))) ordered = sorted(findings, key=lambda f: (SEVERITY_ORDER[f[0]], f[1])) fails = sum(1 for f in findings if f[0] == "FAIL") warns = sum(1 for f in findings if f[0] == "WARN") if args.json: print(json.dumps({ "ledger": args.ledger, "runs": sorted({r.get("run_id", "") for r in rows}), "data_points": results, "run_total": {"accepted": accepted_total, "variable_usd": round(variable_total, 6), "fixed_usd": args.fixed_usd, "review_usd": round(review_total, 6), "fully_loaded_usd": round(loaded_total, 6), "fully_loaded_per_accepted_usd": round(loaded_total / accepted_total, 6) if accepted_total else None}, "reconciliation": reconciliation, "findings": [{"severity": s, "check": c, "name": CHECKS[c], "message": m} for s, c, m in ordered], "passes": [{"severity": s, "check": c, "name": CHECKS[c], "message": m} for s, c, m in passes], "fail": fails, "warn": warns, }, indent=2)) return 1 if fails else 0 runs = sorted({r.get("run_id", "") for r in rows if r.get("run_id")}) print("Enrichment cost: %s (%d rows; run %s)" % (args.ledger, len(rows), ", ".join(runs) or "not named")) for p in results: print("") print("%s" % p["data_point"]) print(" Records: %d; excluded at stage 0: %d; eligible: %d" % (p["records"], p["excluded"], p["eligible"])) for s in p["stages"]: extra = [] if s["candidates_on_excluded"]: extra.append("%d on excluded records" % s["candidates_on_excluded"]) if s["candidates_after_acceptance"]: extra.append("%d after acceptance" % s["candidates_after_acceptance"]) candidates = "%d%s" % (s["candidates"], " (%s)" % ", ".join(extra) if extra else "") marginal = usd(s["marginal_usd"]) + " USD per accepted" if s["marginal_usd"] is not None else "no accepted record" if s.get("ratio_to_earlier"): marginal += " (%.1fx earlier stages)" % s["ratio_to_earlier"] per_candidate = usd(s["cost_per_candidate_usd"]) if s["cost_per_candidate_usd"] is not None else "n/a" coverage = "%.1f%%" % (s["coverage"] * 100) if s["coverage"] is not None else "n/a" print(" Stage %d %s: entered %d; candidates %s; accepted %d, held %d, rejected %d, no result %d, " "error %d, pending %d" % (s["stage"], s["provider"], s["entered"], candidates, s["accepted"], s["held"], s["rejected"], s["no_result"], s["error"], s["pending"])) print(" cost %s USD (%s per candidate); marginal %s; coverage %s" % (usd(s["cost_usd"]), per_candidate, marginal, coverage)) f = p["final"] unresolved = p["eligible"] - f["accepted"] - f["reused"] share = "%.1f%%" % (100.0 * f["accepted"] / p["eligible"]) if p["eligible"] else "n/a" print(" Final: accepted %d of %d (%s); reused %d; unresolved %d (held %d, rejected %d, no result %d, " "pending %d, error %d, not attempted %d)" % (f["accepted"], p["eligible"], share, f["reused"], unresolved, f["held"], f["rejected"], f["no_result"], f["pending"], f["error"], f["not_attempted"])) blended = usd(p["blended_usd"]) + " USD per accepted" if p["blended_usd"] is not None else "no accepted record" print(" Cost: variable %s USD; blended %s" % (usd(p["variable_usd"]), blended)) waste = [] if p["waste_excluded_usd"]: waste.append("%s USD on excluded records (%s)" % (usd(sum(p["waste_excluded_usd"].values())), ", ".join(p["waste_excluded_usd"]))) if p["waste_after_acceptance_usd"]: waste.append("%s USD after acceptance (%s)" % (usd(sum(p["waste_after_acceptance_usd"].values())), ", ".join(p["waste_after_acceptance_usd"]))) print(" Waste: %s" % ("; ".join(waste) if waste else "none")) print("") per = usd(loaded_total / accepted_total) + " USD per accepted record" if accepted_total else "no accepted record" print("Run total: variable %s + fixed %s + review %s = %s USD; %d accepted; %s" % (usd(variable_total), usd(args.fixed_usd), usd(review_total), usd(loaded_total), accepted_total, per)) if reconciliation is not None: print("") print("Billed versus modeled units:") for line in reconciliation: name = "%s %s" % (line["provider"], line["unit_type"]) if line["unit_type"] else line["provider"] if line["billed"] is None: print(" %s: modeled %g, billed not supplied" % (name, line["modeled"])) else: variance = "%+.1f%%" % (line["variance"] * 100) if line.get("variance") is not None else "n/a" print(" %s: modeled %g, billed %g (%s)" % (name, line["modeled"], line["billed"], variance)) print("") print("Findings:") for severity, check, message in ordered: print("%s %s %s: %s" % (severity, check, CHECKS[check], message)) for severity, check, message in passes: print("%s %s %s: %s" % (severity, check, CHECKS[check], message)) print("") print("%d FAIL, %d WARN. Nothing was changed." % (fails, warns)) return 1 if fails else 0 if __name__ == "__main__": sys.exit(main())