#!/usr/bin/env python3 """Offline synthetic cash/cohort report. Python standard library only; no network. Run: python3 calculate.py --check Writes deterministic result/report.json and result/channel-report.csv. Money is integer RUB; rates retain exact numerator/denominator and display floats. """ import argparse import csv import json import re from datetime import datetime from pathlib import Path ROOT = Path(__file__).resolve().parent def flag(value): if value not in (True, False, "true", "false"): raise ValueError("Invalid boolean") return value in (True, "true") def load_rows(name): with (ROOT / "input" / name).open(encoding="utf-8", newline="") as stream: return list(csv.DictReader(stream)) def unique(rows, key): """Identical replays collapse; conflicting versions stop the calculation.""" result = {} duplicate_count = 0 for row in rows: row_id = row[key] if not row_id: raise ValueError("Missing " + key) if row_id in result: if row != result[row_id]: raise ValueError("Conflicting duplicate " + key + ": " + row_id) duplicate_count += 1 else: result[row_id] = row return result, duplicate_count def ratio(numerator, denominator): return { "numerator": numerator, "denominator": denominator, "value": None if numerator is None or not denominator else numerator / denominator, } def compute(cost_rows, lead_rows, order_rows, payment_rows, form_rows, contract): start = datetime.fromisoformat(contract["period_start"]) end = datetime.fromisoformat(contract["period_end_exclusive"]) def in_period(value): timestamp = datetime.fromisoformat(value) if timestamp.tzinfo is None: raise ValueError("Naive timestamp") return start <= timestamp < end def eligible(row, date_field): if not flag(row["is_synthetic"]): raise ValueError("This demonstration accepts synthetic data only") return not flag(row["is_test"]) and in_period(row[date_field]) costs, _ = unique(cost_rows, "channel") for cost in costs.values(): if not flag(cost["is_synthetic"]) or any(int(cost[key]) < 0 for key in ("paid_spend_rub", "other_cost_rub")): raise ValueError("Invalid synthetic acquisition cost") raw_leads, lead_replays = unique(lead_rows, "lead_id") leads = {key: row for key, row in raw_leads.items() if eligible(row, "created_at")} orders, order_replays = unique(order_rows, "order_id") payments, payment_replays = unique(payment_rows, "transaction_id") forms, form_replays = unique(form_rows, "event_id") channels = list(costs) for row in leads.values(): channel = row["channel"] or "unattributed" if channel not in costs and channel != "unattributed": raise ValueError("Lead channel has no cost coverage") if channel not in channels: channels.append(channel) paid_by_order, refunded_by_order, eligible_transactions = {}, {}, [] for txn in payments.values(): if not eligible(txn, "occurred_at") or txn["status"] != "succeeded": continue order_id = txn["order_id"] if order_id not in orders: raise ValueError("Transaction references unknown order") order = orders[order_id] if flag(order["is_test"]): raise ValueError("Non-test transaction references test order") if not eligible(order, "created_at") or order["lead_id"] not in leads: raise ValueError("Transaction is outside the supported lead/order cohort") if datetime.fromisoformat(order["created_at"]) < datetime.fromisoformat(leads[order["lead_id"]]["created_at"]): raise ValueError("Order precedes its lead") if datetime.fromisoformat(txn["occurred_at"]) < datetime.fromisoformat(order["created_at"]): raise ValueError("Transaction precedes its order") amount = int(txn["amount_rub"]) if amount <= 0: raise ValueError("Non-positive transaction amount") if txn["kind"] == "payment": if order_id in paid_by_order: raise ValueError("Multiple payments for one order: manual reconciliation required") if amount != int(order["amount_due_rub"]): raise ValueError("Payment does not match full order amount") paid_by_order[order_id] = amount elif txn["kind"] == "refund": refunded_by_order[order_id] = refunded_by_order.get(order_id, 0) + amount else: raise ValueError("Unsupported transaction kind") eligible_transactions.append(txn) for order_id, amount in refunded_by_order.items(): if order_id not in paid_by_order or amount > paid_by_order[order_id]: raise ValueError("Refund has no reconciled payment or exceeds payment") customers_seen = set() for order_id in paid_by_order: order = orders[order_id] if order["customer_id"] in customers_seen: raise ValueError("Returning customer needs a defined channel allocation policy") customers_seen.add(order["customer_id"]) def section(selection, is_total=False): selected_leads = [row for row in leads.values() if (row["channel"] or "unattributed") in selection] selected_ids = {row["lead_id"] for row in selected_leads} selected_orders = [key for key in paid_by_order if orders[key]["lead_id"] in selected_ids] selected_customer_ids = {orders[key]["customer_id"] for key in selected_orders} selected_form_events = [row for row in forms.values() if eligible(row, "occurred_at") and row["lead_id"] in selected_ids] cost_known = is_total or all(channel in costs for channel in selection) paid_cost = sum(int(costs[channel]["paid_spend_rub"]) for channel in selection if channel in costs) if cost_known else None other_cost = sum(int(costs[channel]["other_cost_rub"]) for channel in selection if channel in costs) if cost_known else None total_cost = paid_cost + other_cost if cost_known else None qualified = sum(flag(row["qualified"]) for row in selected_leads) gross = sum(paid_by_order[key] for key in selected_orders) refunds = sum(refunded_by_order.get(key, 0) for key in selected_orders) return { "channel": "all" if is_total else selection[0], "lead_ids": sorted(selected_ids), "paid_order_ids": sorted(selected_orders), "customer_ids": sorted(selected_customer_ids), "form_event_ids": sorted(row["event_id"] for row in selected_form_events), "leads": len(selected_leads), "qualified": qualified, "paid_orders": len(selected_orders), "paid_customers": len(selected_customer_ids), "form_events": len(selected_form_events), "paid_spend_rub": paid_cost, "other_cost_rub": other_cost, "total_cost_rub": total_cost, "cost_coverage": "known_global_pool_not_allocated_to_unknown_source" if is_total else ("known" if cost_known else "unknown"), "cpl": ratio(total_cost, len(selected_leads)), "cpql": ratio(total_cost, qualified), "cac": ratio(total_cost, len(selected_customer_ids)), "qualified_rate": ratio(qualified, len(selected_leads)), "paid_rate": ratio(len(selected_customer_ids), len(selected_leads)), "gross_revenue_rub": gross, "refunds_rub": refunds, "net_revenue_rub": gross - refunds, "net_revenue_minus_acquisition_rub": None if total_cost is None else gross - refunds - total_cost, "profit_rub": None, } by_channel = [section([channel]) for channel in channels] total = section(channels, True) attributed = sum(row["paid_orders"] for row in by_channel if row["channel"] != "unattributed") return { "schema_version": 1, "classification": "synthetic_teaching_derivative", "period": {"start": contract["period_start"], "end_exclusive": contract["period_end_exclusive"], "timezone": "Europe/Moscow"}, "currency": "RUB", "total": total, "channels": by_channel, "attribution_coverage_paid_orders": ratio(attributed, total["paid_orders"]), "reconciliation": { "raw_lead_rows": len(lead_rows), "lead_row_replays": lead_replays, "eligible_distinct_leads": len(leads), "raw_order_rows": len(order_rows), "order_row_replays": order_replays, "raw_payment_event_rows": len(payment_rows), "payment_transaction_replays_removed": payment_replays, "distinct_transaction_ids": len(payments), "eligible_succeeded_payment_transactions": sum(txn["kind"] == "payment" for txn in eligible_transactions), "eligible_succeeded_refund_transactions": sum(txn["kind"] == "refund" for txn in eligible_transactions), "raw_form_event_rows": len(form_rows), "form_event_id_replays_removed": form_replays, "eligible_distinct_form_events": total["form_events"], "unique_paid_orders": len(paid_by_order), "unique_paid_customers": len(customers_seen), "known_channel_cost_rub": total["total_cost_rub"], "channel_leads_reconcile": sum(row["leads"] for row in by_channel) == total["leads"], "channel_orders_reconcile": sum(row["paid_orders"] for row in by_channel) == total["paid_orders"], "channel_net_revenue_reconciles": sum(row["net_revenue_rub"] for row in by_channel) == total["net_revenue_rub"], }, "limitations": contract["limitations"], } def load_inputs(): return [load_rows(name) for name in ("channel-costs.csv", "leads.csv", "orders.csv", "payment-events.csv", "form-events.csv")] + [json.loads((ROOT / "input" / "measurement-contract.json").read_text(encoding="utf-8"))] def checks(inputs, report): """Independent expected arithmetic plus meaningful mutation/boundary checks.""" import copy checked = [] def verify(name, condition): if not condition: raise AssertionError(name) checked.append(name) total = report["total"] verify("27 distinct leads; duplicate form event is not a lead", total["leads"] == 27 and total["form_events"] == 28) verify("16 qualified leads", total["qualified"] == 16) verify("250000 paid + 50000 other costs", total["paid_spend_rub"] == 250000 and total["other_cost_rub"] == 50000) verify("7 unique full paid orders and customers", total["paid_orders"] == total["paid_customers"] == 7) verify("7 x 48000 - 48000 refund = 288000", total["gross_revenue_rub"] == 336000 and total["refunds_rub"] == 48000 and total["net_revenue_rub"] == 288000) verify("CPL recomputed, not averaged", abs(total["cpl"]["value"] - 300000 / 27) < 1e-8) verify("CPQL 18750", total["cpql"]["value"] == 18750) verify("CAC uses unique customers", abs(total["cac"]["value"] - 300000 / 7) < 1e-8) unknown = next(row for row in report["channels"] if row["channel"] == "unattributed") verify("unknown channel cost is null, not free", unknown["leads"] == 1 and unknown["net_revenue_rub"] == 48000 and unknown["cac"]["value"] is None) direct = next(row for row in report["channels"] if row["channel"] == "direct_search") verify("Direct 8/5/2 and source-backed ratios", [direct["leads"], direct["qualified"], direct["paid_orders"]] == [8, 5, 2] and direct["cpl"]["value"] == 15000 and direct["cpql"]["value"] == 24000 and direct["cac"]["value"] == 60000) verify("all channel additive totals reconcile", all(report["reconciliation"][key] for key in ("channel_leads_reconcile", "channel_orders_reconcile", "channel_net_revenue_reconciles"))) verify("one replayed payment removed", report["reconciliation"]["payment_transaction_replays_removed"] == 1) mutated = copy.deepcopy(inputs) mutated[3].append(copy.deepcopy(mutated[3][0])) verify("another exact payment replay does not change cash/orders", compute(*mutated)["total"] == total) mutated = copy.deepcopy(inputs) bad = copy.deepcopy(mutated[3][0]); bad["amount_rub"] = "999999"; mutated[3].append(bad) try: compute(*mutated) except ValueError: checked.append("conflicting transaction ID stops calculation") else: raise AssertionError("conflicting transaction accepted") mutated = copy.deepcopy(inputs) refund = next(row for row in mutated[3] if row["kind"] == "refund"); refund["amount_rub"] = "48001" try: compute(*mutated) except ValueError: checked.append("over-refund stops calculation") else: raise AssertionError("over-refund accepted") verify("zero and missing denominator return null", ratio(120000, 0)["value"] is None and ratio(None, 1)["value"] is None) mutated = copy.deepcopy(inputs) for row in mutated[1]: if row["channel"] == "outreach": row["qualified"] = "false" out = next(row for row in compute(*mutated)["channels"] if row["channel"] == "outreach") verify("no qualifications gives CPQL null", out["cpql"]["value"] is None) verify("cash after acquisition is not profit", total["net_revenue_minus_acquisition_rub"] == -12000 and total["profit_rub"] is None) return {"passed": True, "checks": checked, "count": len(checked), "scope": "Arithmetic and stated input invariants; visual/browser acceptance is separate."} def main(): parser = argparse.ArgumentParser() parser.add_argument("--check", action="store_true") args = parser.parse_args() inputs = load_inputs() report = compute(*inputs) receipt = checks(inputs, report) if args.check else None destination = ROOT / "result" destination.mkdir(exist_ok=True) (destination / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") columns = ["channel", "leads", "qualified", "paid_orders", "paid_customers", "form_events", "total_cost_rub", "gross_revenue_rub", "refunds_rub", "net_revenue_rub", "cpl_rub", "cpql_rub", "cac_rub"] with (destination / "channel-report.csv").open("w", encoding="utf-8", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=columns); writer.writeheader() for row in report["channels"] + [report["total"]]: flat = {key: row.get(key) for key in columns if not key.endswith(("cpl_rub", "cpql_rub", "cac_rub"))} for key in ("cpl", "cpql", "cac"): flat[key + "_rub"] = row[key]["value"] writer.writerow(flat) # Keep the offline dashboard bound to this exact calculation/input snapshot. # No fetch, network, server process, external path, or hosted write is used. index = ROOT / "index.html" if index.is_file(): html = index.read_text(encoding="utf-8") snapshots = {"report-data": report, "input-data": {"leads": inputs[1], "orders": inputs[2], "transactions": inputs[3]}} for element_id, data in snapshots.items(): pattern = r'()' encoded = json.dumps(data, ensure_ascii=False).replace("