File size: 7,295 Bytes
8bf7463 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | """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()
|