#!/usr/bin/env python3"""Aggregates the private swap rows into the small CSVs committed under raw/explorer/.Rows stay local (hundreds of MB, and hashed addresses have no public use). Aggregates are enough to rebuild everychart offline. Anyone with a free Explorer key can regenerate the rows, because the sampling is deterministic."""import csv, gzip, hashlib, io, json, math, pathlib, re, statistics, sysfrom collections import Counter, defaultdictROOT = pathlib.Path(__file__).resolve().parent.parentSRC = ROOT / "raw_private" / "explorer"OUT = ROOT / "raw" / "explorer"sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))NEAR_WALLETS = {"buybacks.multisignature.near", "fefundsadmin.sputnik-dao.near", "1csfundsadmin.sputnik-dao.near"}MANIFEST = []def token_meta():    meta = {}    for r in csv.DictReader((pathlib.Path(__file__).resolve().parent / "sql" / "token_prices.csv").open()):        meta["nep141:" + r["token"]] = r    return metaMETA = token_meta()STABLE = {"USDC", "USDT", "USDT0", "DAI", "USD1", "USDG", "USDC.E", "XDAI", "FRAX", "USDE"}def family(asset):    sym = (META.get(asset) or {}).get("symbol", "").upper()    if sym in STABLE:        return "stable"    if "BTC" in sym:        return "btc"    if sym in ("ETH", "WETH"):        return "eth"    if sym == "ZEC":        return "zec"    if sym in ("SOL", "WSOL"):        return "sol"    if sym in ("WNEAR", "NEAR"):        return "near"    return "other"def chain(asset):    return (META.get(asset) or {}).get("blockchain") or "other"def rows_in(path):    with gzip.open(path, "rt") as fh:        for r in csv.DictReader(fh):            try:                r["in_usd"], r["out_usd"] = float(r["in_usd"] or 0), float(r["out_usd"] or 0)            except ValueError:                continue            r["bps"] = []            for part in (r["fees"] or "").split(";"):                if ":" in part:                    who, fee = part.rsplit(":", 1)                    try:                        r["bps"].append((who, float(fee)))                    except ValueError:                        pass            yield rdef size_bucket(usd):    if usd < 100:        return "under $100"    if usd < 1000:        return "$100-1k"    if usd < 10000:        return "$1k-10k"    if usd < 100000:        return "$10k-100k"    return "over $100k"def write(name, rows):    OUT.mkdir(parents=True, exist_ok=True)    if not rows:        return    buf = io.StringIO()    wr = csv.DictWriter(buf, fieldnames=list(rows[0].keys()), lineterminator="\n")    wr.writeheader()    wr.writerows(rows)    raw = buf.getvalue().encode()    with gzip.GzipFile(OUT / f"{name}.csv.gz", "wb", mtime=0) as gz:        gz.write(raw)    MANIFEST.append({"file": f"{name}.csv.gz", "rows": len(rows), "sha256_csv": hashlib.sha256(raw).hexdigest()})    print(name, len(rows))def panel():    daily, partner, recipient, routes, sizes, cost = [], [], [], [], [], []    files = sorted((SRC / "panel").glob("*.csv.gz"))    for path in files:        day = path.name[:10]        st = Counter()        usd = fee = out = 0.0        p_usd, p_fee, p_n = Counter(), Counter(), Counter()        p_posted = defaultdict(Counter)        r_fee = Counter()        fam, cha = Counter(), Counter()        sz_usd, sz_n = Counter(), Counter()        costs = defaultdict(list)        send = Counter()        blank = 0.0        for r in rows_in(path):            st[r["status"]] += 1            if r["status"] != "SUCCESS" or r["in_usd"] <= 0:                continue            i = r["in_usd"]            usd += i            out += r["out_usd"]            ref = re.sub(r"[^A-Za-z0-9 ._()@:-]", "_", r["referral"] or "(none)")[:60]            f = sum(b for _, b in r["bps"]) / 1e4 * i            fee += f            p_usd[ref] += i            p_fee[ref] += f            p_n[ref] += 1            p_posted[ref][round(sum(b for _, b in r["bps"]), 1)] += 1            for who, b in r["bps"]:                r_fee[who] += b / 1e4 * i            a, b = family(r["origin"]), family(r["dest"])            fam[f"{a}>{b}"] += i            cha[f"{chain(r['origin'])}>{chain(r['dest'])}"] += i            sb = size_bucket(i)            sz_usd[sb] += i            sz_n[sb] += 1            if i >= 10 and r["out_usd"] > 0:                costs[sb].append((i - r["out_usd"]) / i * 1e4)            if r["sender"]:                send[r["sender"]] += i            else:                blank += i        top = send.most_common(100)        daily.append({"day": day, "swaps": st["SUCCESS"], "refunded": st["REFUNDED"], "failed": st["FAILED"], "usd_in": round(usd), "usd_out": round(out), "app_fee_usd": round(fee),                      "near_wallet_fee_usd": round(sum(v for k, v in r_fee.items() if k in NEAR_WALLETS)), "senders": len(send), "blank_sender_usd": round(blank),                      "top1_sender_usd": round(top[0][1]) if top else 0, "top10_sender_usd": round(sum(v for _, v in top[:10])), "top100_sender_usd": round(sum(v for _, v in top))})        for k, v in p_usd.items():            partner.append({"day": day, "partner": k, "usd": round(v), "swaps": p_n[k], "fee_usd": round(p_fee[k], 2), "posted_bps": p_posted[k].most_common(1)[0][0]})        for k, v in r_fee.items():            recipient.append({"day": day, "recipient": k, "fee_usd": round(v, 2)})        for k, v in fam.items():            routes.append({"day": day, "kind": "family", "route": k, "usd": round(v)})        for k, v in cha.items():            routes.append({"day": day, "kind": "chain", "route": k, "usd": round(v)})        for k, v in sz_usd.items():            q = statistics.quantiles(costs[k], n=4) if len(costs[k]) >= 8 else [None, None, None]            sizes.append({"day": day, "bucket": k, "n": sz_n[k], "usd": round(v), "cost_p25": q[0], "cost_p50": q[1], "cost_p75": q[2]})    keep_partner = Counter()    for r in partner:        keep_partner[r["partner"]] += r["usd"]    big = {k for k, _ in keep_partner.most_common(60)}    merged = defaultdict(lambda: [0, 0, 0.0, 0.0])    for r in partner:        k = (r["day"], r["partner"] if r["partner"] in big else "(other)")        merged[k][0] += r["usd"]        merged[k][1] += r["swaps"]        merged[k][2] += r["fee_usd"]        merged[k][3] = r["posted_bps"]    partner = [{"day": k[0], "partner": k[1], "usd": v[0], "swaps": v[1], "fee_usd": round(v[2], 2), "posted_bps": v[3]} for k, v in sorted(merged.items())]    keep_rec = Counter()    for r in recipient:        keep_rec[r["recipient"]] += r["fee_usd"]    # Only large, named fee collectors are published. Hex and small accounts roll into (other).    big = {k for k, _ in keep_rec.most_common(15) if re.fullmatch(r"[a-z0-9_-]+(\.[a-z0-9_-]+)*\.[a-z]+", k) and not re.fullmatch(r"[0-9a-f]{64}", k)}    merged = defaultdict(float)    for r in recipient:        merged[(r["day"], r["recipient"] if r["recipient"] in big else "(other)")] += r["fee_usd"]    recipient = [{"day": k[0], "recipient": k[1], "fee_usd": round(v, 2)} for k, v in sorted(merged.items())]    write("panel_daily", daily)    write("panel_partner", partner)    write("panel_recipient", recipient)    write("panel_routes", sorted(routes, key=lambda r: (r["day"], r["kind"], -r["usd"])))    write("panel_sizes", sizes)def recent():    files = sorted((SRC / "recent").glob("*.csv.gz"))    if not files:        return    weeks = defaultdict(lambda: {"send": Counter(), "recip": Counter(), "swaps": 0, "usd": 0.0})    first_seen = {}    for path in files:        day = path.name[:10]        import datetime        d = datetime.date.fromisoformat(day)        wk = (d - datetime.timedelta(days=d.weekday())).isoformat()        w = weeks[wk]        for r in rows_in(path):            if r["status"] != "SUCCESS" or r["in_usd"] <= 0:                continue            w["swaps"] += 1            w["usd"] += r["in_usd"]            key = r["sender"] or r["recipient"]            if key:                w["send"][key] += r["in_usd"]                first_seen.setdefault(key, wk)            if r["recipient"]:                w["recip"][r["recipient"]] += r["in_usd"]    order = sorted(weeks)    table = []    for i, wk in enumerate(order):        w = weeks[wk]        users = w["send"]        ranked = users.most_common()        total = sum(users.values()) or 1        ret1 = len(set(users) & set(weeks[order[i + 1]]["send"])) / len(users) * 100 if i + 1 < len(order) and users else None        ret4 = len(set(users) & set(weeks[order[i + 4]]["send"])) / len(users) * 100 if i + 4 < len(order) and users else None        table.append({"week": wk, "swaps": w["swaps"], "usd": round(w["usd"]), "users": len(users), "new_users": sum(1 for u in users if first_seen[u] == wk),                      "top1_share": round(ranked[0][1] / total * 100, 2) if ranked else 0, "top10_share": round(sum(v for _, v in ranked[:10]) / total * 100, 2),                      "top100_share": round(sum(v for _, v in ranked[:100]) / total * 100, 2), "top1000_share": round(sum(v for _, v in ranked[:1000]) / total * 100, 2),                      "median_user_usd": round(statistics.median(users.values()), 2) if users else 0, "ret_1w": round(ret1, 1) if ret1 is not None else "", "ret_4w": round(ret4, 1) if ret4 is not None else ""})    write("recent_weekly", table)    all_users = Counter()    for w in weeks.values():        all_users.update(w["send"])    freq = Counter(sum(1 for w in weeks.values() if u in w["send"]) for u in all_users)    write("recent_frequency", [{"weeks_active": k, "users": v} for k, v in sorted(freq.items())])def whales():    files = sorted((SRC / "whales").glob("*.csv.gz")) if (SRC / "whales").exists() else []    rows = []    for path in files:        usd = n = 0        senders = Counter()        fee = 0.0        for r in rows_in(path):            if r["status"] != "SUCCESS":                continue            usd += r["in_usd"]            n += 1            fee += sum(b for _, b in r["bps"]) / 1e4 * r["in_usd"]            senders[r["sender"] or r["recipient"]] += r["in_usd"]        top = senders.most_common(10)        rows.append({"month": path.name[:7], "swaps": n, "usd": round(usd), "fee_usd": round(fee), "users": len(senders), "top1_usd": round(top[0][1]) if top else 0, "top10_usd": round(sum(v for _, v in top))})    write("whales_monthly", rows)def sample_record():    """Records which windows the aggregates were built from, with the row count of each."""    out = {}    for stage in ("panel", "recent", "whales"):        out[stage] = {}        for done in sorted((SRC / stage).glob("*.done")) if (SRC / stage).exists() else []:            out[stage][done.name.split(".")[0]] = json.loads(done.read_text())["rows"]    return outif __name__ == "__main__":    panel()    recent()    whales()    (OUT / "MANIFEST.json").write_text(json.dumps({"built_from": sample_record(), "files": MANIFEST}, indent=0, sort_keys=True))