code/build_dataset.py

"""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()