"""Build the x402 paid-demand dataset from daily CDP Bazaar snapshots. Input : one JSON snapshot per day (the spark-lab Scout:Bazaar writes these to vault/raw/bazaar/YYYY-MM-DD.json). Only the `cdp` list is used. Output: data/rows.parquet — one row per (anchor snapshot, resource). PRIVATE: the dataset is never published (see docs/PROCESS.md). Label (see docs/PROCESS.md §3 for why it is not "still listed"): paid_{H}d = 1 iff the resource receives >=1 paid call in (ref_t, ref_t + H days], observed as a later snapshot whose quality.lastCalledAt falls in that window. Evicted / never-called-again rows are 0. Features are computed ONLY from snapshots <= t (no look-ahead). """ from __future__ import annotations import argparse import hashlib import json from collections import defaultdict from datetime import datetime, timedelta, timezone from pathlib import Path from urllib.parse import urlparse import numpy as np import pandas as pd DEFAULT_SNAP_DIR = Path.home() / "myapps/spark-lab/vault/raw/bazaar" HORIZONS = (7, 14) def ts(s: str | None) -> datetime | None: if not s: return None try: d = datetime.fromisoformat(str(s).replace("Z", "+00:00")) except ValueError: return None return d if d.tzinfo else d.replace(tzinfo=timezone.utc) def host_of(url: str) -> str: try: return (urlparse(url).hostname or "").lower() except ValueError: return "" def host_bucket(host: str) -> int: """Stable 0-99 bucket per host, used for the host-disjoint split.""" return int(hashlib.sha256(host.encode()).hexdigest(), 16) % 100 def load_snapshots(snap_dir: Path) -> list[tuple[str, dict[str, dict]]]: snaps = [] for f in sorted(snap_dir.glob("20??-??-??.json")): rows = json.loads(f.read_text()).get("cdp") or [] by_res: dict[str, dict] = {} for r in rows: by_res[r["resource"]] = r # duplicates in a page: last wins snaps.append((f.stem, by_res)) return snaps def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--snap-dir", type=Path, default=DEFAULT_SNAP_DIR) ap.add_argument("--out", type=Path, default=Path("data/rows.parquet")) args = ap.parse_args() snaps = load_snapshots(args.snap_dir) n = len(snaps) print(f"{n} snapshots {snaps[0][0]} .. {snaps[-1][0]}") # Reference time of each snapshot = newest lastCalledAt in it (~fetch time). ref: list[datetime] = [] for _, by_res in snaps: ref.append(max(t for r in by_res.values() if (t := ts((r.get("quality") or {}).get("lastCalledAt"))))) # Per-resource timeline of lastCalledAt observations (snapshot idx -> datetime). last_called: dict[str, list[tuple[int, datetime]]] = defaultdict(list) first_seen: dict[str, int] = {} for i, (_, by_res) in enumerate(snaps): for res, r in by_res.items(): first_seen.setdefault(res, i) lc = ts((r.get("quality") or {}).get("lastCalledAt")) if lc: last_called[res].append((i, lc)) out = [] for t, (day, by_res) in enumerate(snaps): rt = ref[t] # Label availability: horizon must be fully observed (one spare snapshot of slack). can = {h: rt + timedelta(days=h + 1) <= ref[-1] for h in HORIZONS} if not any(can.values()): continue # ---- host aggregates at t (only data <= t) ---- host_rows: dict[str, list[dict]] = defaultdict(list) for res, r in by_res.items(): host_rows[host_of(res)].append(r) host_feat = {} for h, rows in host_rows.items(): q = [(r.get("quality") or {}) for r in rows] calls = np.array([x.get("l30DaysTotalCalls") or 0 for x in q], float) payers = np.array([x.get("l30DaysUniquePayers") or 0 for x in q], float) lcs = [ts(x.get("lastCalledAt")) for x in q] recent7 = np.mean([bool(lc and (rt - lc).days < 7) for lc in lcs]) host_feat[h] = dict( host_listings=len(rows), host_active_rate_7d=float(recent7), host_one_call_per_payer_share=float(np.mean((calls > 0) & (calls == payers))), host_median_payers=float(np.median(payers)), host_networks=len({r.get("network") for r in rows}), ) prev7 = snaps[t - 7][1] if t >= 7 else None for res, r in by_res.items(): q = r.get("quality") or {} lc = ts(q.get("lastCalledAt")) lu = ts(r.get("lastUpdated")) calls = q.get("l30DaysTotalCalls") or 0 payers = q.get("l30DaysUniquePayers") or 0 host = host_of(res) path = urlparse(res).path if res.startswith("http") else res try: price_usd = int(r.get("price") or 0) / 1e6 # USDC has 6 decimals except (TypeError, ValueError): price_usd = float("nan") p7 = (prev7 or {}).get(res) row = dict( anchor_idx=t, anchor_day=day, resource=res, host=host, host_bucket=host_bucket(host), path=path, path_depth=path.count("/"), has_path_param=int(":" in path or "{" in path), description=(r.get("description") or "").strip(), desc_len=len(r.get("description") or ""), network=r.get("network") or "", type=r.get("type") or "", price_usd=price_usd, calls_30d=calls, payers_30d=payers, calls_per_payer=calls / payers if payers else 0.0, days_since_last_call=(rt - lc).total_seconds() / 86400 if lc else np.nan, days_since_last_update=(rt - lu).total_seconds() / 86400 if lu else np.nan, days_listed_observed=t - first_seen[res], first_seen_censored=int(first_seen[res] == 0), is_new_listing=int(first_seen[res] == t and t > 0), present_7d_ago=int(p7 is not None) if t >= 7 else -1, payers_delta_7d=(payers - ((p7 or {}).get("quality") or {}).get("l30DaysUniquePayers", 0)) if p7 else np.nan, calls_delta_7d=(calls - ((p7 or {}).get("quality") or {}).get("l30DaysTotalCalls", 0)) if p7 else np.nan, **host_feat[host], ) base = lc or datetime.min.replace(tzinfo=timezone.utc) for h in HORIZONS: if not can[h]: row[f"paid_{h}d"] = -1 # unobserved continue end = rt + timedelta(days=h) hit = any(i > t and base < x <= end and x > rt for i, x in last_called[res]) row[f"paid_{h}d"] = int(hit) out.append(row) df = pd.DataFrame(out) args.out.parent.mkdir(parents=True, exist_ok=True) df.to_parquet(args.out, index=False) print(f"rows {len(df):,} resources {df.resource.nunique():,} hosts {df.host.nunique():,}") for h in HORIZONS: s = df[df[f"paid_{h}d"] >= 0] print(f"paid_{h}d: {len(s):,} labelled rows, anchors {s.anchor_idx.min()}..{s.anchor_idx.max()}, base rate {s[f'paid_{h}d'].mean():.3f}") if __name__ == "__main__": main()