diff --git a/.github/workflows/hf-publish.yml b/.github/workflows/hf-publish.yml new file mode 100644 index 00000000..c2d95d39 --- /dev/null +++ b/.github/workflows/hf-publish.yml @@ -0,0 +1,76 @@ +name: HF dataset publish + +# Daily snapshot of the live citable feed pushed to +# https://huggingface.co/datasets/OpenChainBench/benchmarks. +# +# Trigger: +# - 00:00 UTC every day (after the daily Prom roll-up is settled). +# - workflow_dispatch with an optional `dry_run` flag so we can +# validate changes without touching HF. +# +# Required secrets: +# HF_TOKEN write-scoped token for the dataset repo. +# SLACK_WEBHOOK_URL optional, incoming-webhook URL for ops alerts. + +on: + schedule: + - cron: "0 0 * * *" + workflow_dispatch: + inputs: + dry_run: + description: "Stage parquet locally without pushing to HF" + type: boolean + default: false + +concurrency: + group: hf-publish + cancel-in-progress: false + +jobs: + publish: + runs-on: ubuntu-latest + timeout-minutes: 15 + permissions: + contents: read + env: + HF_TOKEN: ${{ secrets.HF_TOKEN }} + SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }} + OCB_API: https://openchainbench.com + HF_REPO_ID: OpenChainBench/benchmarks + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: pip + cache-dependency-path: scripts/hf_publisher/requirements.txt + + - name: Install deps + run: pip install -r scripts/hf_publisher/requirements.txt + + - name: Run publisher tests (offline) + # Schema/quorum regressions caught here never reach the dataset. + run: | + cd scripts/hf_publisher + python -m unittest test_publish.py -v + + - name: Publish snapshot + run: | + cd scripts/hf_publisher + if [ "${{ github.event_name }}" = "workflow_dispatch" ] && [ "${{ inputs.dry_run }}" = "true" ]; then + python publish.py --dry-run + else + python publish.py + fi + + - name: Summary + if: always() + run: | + { + echo "## HF publish ${{ job.status }}" + echo "" + echo "**Repo:** https://huggingface.co/datasets/${HF_REPO_ID}" + echo "**Mode:** ${{ github.event_name == 'workflow_dispatch' && inputs.dry_run == true && 'dry-run' || 'live' }}" + echo "**Time:** $(date -u +%FT%TZ)" + } >> "$GITHUB_STEP_SUMMARY" diff --git a/.gitignore b/.gitignore index 32b1bcfe..5ca4a66c 100644 --- a/.gitignore +++ b/.gitignore @@ -61,3 +61,8 @@ harnesses/*/script harnesses/*/monitor harnesses/*/cmd/script/script harnesses/*/cmd/monitor/monitor + +# Python venvs for the HF publisher (created locally for dry-runs) +scripts/hf_publisher/.venv/ +**/__pycache__/ +*.pyc diff --git a/scripts/hf_publisher/README.md b/scripts/hf_publisher/README.md new file mode 100644 index 00000000..01e55dca --- /dev/null +++ b/scripts/hf_publisher/README.md @@ -0,0 +1,55 @@ +# HF dataset publisher + +Daily snapshot publisher for the public +[OpenChainBench/benchmarks](https://huggingface.co/datasets/OpenChainBench/benchmarks) +dataset on Hugging Face. + +## What it does + +- Fetches `https://openchainbench.com/api/citable` and + `/api/stat/` for every live bench. +- Refuses to publish if the source feed is degraded (quorum guard: half + the count below `live` status, or count below the floor). +- Projects the JSON into three Hive-partitioned Parquet tables: + - `headlines/` 1 row per (slug, day) + - `providers/` 1 row per (slug, provider, day) + - `timeseries/` 1 row per (slug, point, day) +- Stages a fixed set of static assets (README, CITATION.cff, LICENSE, + JSON schemas, example queries) and pushes the whole thing to HF. + +## Schema versioning + +`SCHEMA_VERSION` in `publish.py` is the source of truth. Bump it any +time a column is added. **Never** rename or remove columns: the dataset +is a long-lived public artifact and consumers will write queries +against the column names. + +## Local dry-run + +```bash +cd scripts/hf_publisher +pip install -r requirements.txt +python publish.py --dry-run --out /tmp/ocb-hf-test +ls /tmp/ocb-hf-test +``` + +## CI + +The `.github/workflows/hf-publish.yml` workflow runs the tests first, +then either `publish.py --dry-run` (manual dispatch with the flag) or +the real push (scheduled run or manual without the flag). + +Required GitHub secrets: +- `HF_TOKEN` write-scoped token on the dataset repo. +- `SLACK_WEBHOOK_URL` optional incoming-webhook URL for ops alerts. + +## Tests + +```bash +cd scripts/hf_publisher +python -m unittest test_publish.py -v +``` + +Tests cover the quorum guard, all three row builders, the partition +path layout, and the static-asset templating step. They never hit the +live API or HF Hub. diff --git a/scripts/hf_publisher/dataset_template/CITATION.cff b/scripts/hf_publisher/dataset_template/CITATION.cff new file mode 100644 index 00000000..24e605ad --- /dev/null +++ b/scripts/hf_publisher/dataset_template/CITATION.cff @@ -0,0 +1,25 @@ +cff-version: 1.2.0 +title: OpenChainBench Benchmarks +abstract: >- + Daily snapshots of every public benchmark on openchainbench.com - + open, reproducible measurements of crypto infrastructure (RPCs, + oracles, bridges, data APIs, Polymarket adapters, Hyperliquid + builders). Released as Hive-partitioned Parquet under CC-BY-4.0. +authors: + - name: OpenChainBench Contributors + website: https://openchainbench.com +type: dataset +license: CC-BY-4.0 +repository-code: https://github.com/ChainBench/OpenChainBench +url: https://huggingface.co/datasets/OpenChainBench/benchmarks +date-released: "{{snapshot_date}}" +keywords: + - blockchain + - crypto + - benchmarks + - infrastructure + - latency + - oracles + - bridges + - polymarket + - hyperliquid diff --git a/scripts/hf_publisher/dataset_template/LICENSE b/scripts/hf_publisher/dataset_template/LICENSE new file mode 100644 index 00000000..eef3a499 --- /dev/null +++ b/scripts/hf_publisher/dataset_template/LICENSE @@ -0,0 +1,23 @@ +Creative Commons Attribution 4.0 International (CC BY 4.0) + +You are free to: + Share - copy and redistribute the material in any medium or format + Adapt - remix, transform, and build upon the material for any purpose, + even commercially. + +Under the following terms: + Attribution - You must give appropriate credit, provide a link to the + license, and indicate if changes were made. You may do + so in any reasonable manner, but not in any way that + suggests the licensor endorses you or your use. + +No additional restrictions - You may not apply legal terms or +technological measures that legally restrict others from doing anything +the license permits. + +Full license text: https://creativecommons.org/licenses/by/4.0/legalcode +Summary: https://creativecommons.org/licenses/by/4.0/ + +Suggested attribution: + OpenChainBench. (2026). OpenChainBench Benchmarks [Data set]. + Hugging Face. https://huggingface.co/datasets/OpenChainBench/benchmarks diff --git a/scripts/hf_publisher/dataset_template/README.md b/scripts/hf_publisher/dataset_template/README.md new file mode 100644 index 00000000..3b75792a --- /dev/null +++ b/scripts/hf_publisher/dataset_template/README.md @@ -0,0 +1,142 @@ +--- +license: cc-by-4.0 +language: + - en +pretty_name: OpenChainBench Benchmarks +task_categories: + - tabular-classification + - other +tags: + - crypto + - blockchain + - benchmarks + - rpc + - oracles + - bridges + - polymarket + - infrastructure + - latency + - finance +size_categories: + - 1K` +- Site-wide methodology: `https://openchainbench.com/methodology` +- YAML spec source (Apache 2.0): `https://github.com/ChainBench/OpenChainBench/tree/main/benchmarks` + +Harness code is open source where applicable (column `source_url`). + +## Schema stability promise + +- New columns may be added without warning. Consumer queries should + select named columns rather than `SELECT *`. +- Existing columns will never be renamed or removed. If a breaking + change ever proves unavoidable, a parallel v2 folder ships alongside + the v1 partitions so old consumers keep working. +- `schema_version` integer in each row marks the additive schema epoch. + +## Update cadence + +Snapshots run daily at 00:00 UTC via a GitHub Action. If a run aborts +on a quorum check (the source feed has fewer than half its benches +live), no new partition is added that day - the previous good snapshot +stays as truth. + +## License & citation + +Data is released under +[Creative Commons Attribution 4.0](https://creativecommons.org/licenses/by/4.0/). +Use it freely with attribution to OpenChainBench. + +For academic citation see `CITATION.cff` at the root of this dataset +(GitHub / HF / Zenodo all parse it). Suggested attribution string: + +> OpenChainBench. (2026). OpenChainBench Benchmarks [Data set]. +> Hugging Face. https://huggingface.co/datasets/OpenChainBench/benchmarks + +## Provenance + +- Source code (publisher + benchmark YAML specs): https://github.com/ChainBench/OpenChainBench +- Live measurement APIs: https://openchainbench.com/api/citable, /api/stat, /api/llm-context, /api/mcp +- Issues / questions: https://github.com/ChainBench/OpenChainBench/issues diff --git a/scripts/hf_publisher/dataset_template/examples/01_pandas.py b/scripts/hf_publisher/dataset_template/examples/01_pandas.py new file mode 100644 index 00000000..0b641c57 --- /dev/null +++ b/scripts/hf_publisher/dataset_template/examples/01_pandas.py @@ -0,0 +1,25 @@ +""" +Load the OCB headlines feed with pandas via the Hugging Face datasets +library. Good when you want a familiar DataFrame and the dataset is +small enough to fit in memory (it is). +""" + +from datasets import load_dataset + +ds = load_dataset( + "OpenChainBench/benchmarks", + "headlines", + split="train", +) +df = ds.to_pandas() + +# Latest snapshot only +latest = df["snapshot_date"].max() +today = df[df["snapshot_date"] == latest] + +# Top 10 benchmarks by sample size today +print( + today.sort_values("sample_size", ascending=False)[ + ["slug", "leader_name", "value", "unit", "sample_size"] + ].head(10) +) diff --git a/scripts/hf_publisher/dataset_template/examples/02_polars.py b/scripts/hf_publisher/dataset_template/examples/02_polars.py new file mode 100644 index 00000000..43b60907 --- /dev/null +++ b/scripts/hf_publisher/dataset_template/examples/02_polars.py @@ -0,0 +1,23 @@ +""" +Stream the OCB providers feed with Polars directly from HF. Pushdown +predicate + projection means only the columns and partitions you ask +for ever hit the wire. Recommended for analytic workloads. +""" + +import polars as pl + +providers = pl.scan_parquet( + "hf://datasets/OpenChainBench/benchmarks/providers/**/*.parquet" +) + +# Trend of Mobula's p50 latency on bridge-quote-latency across all snapshots +trend = ( + providers.filter( + (pl.col("bench_slug") == "bridge-quote-latency") + & (pl.col("provider_slug") == "mobula") + ) + .select(["snapshot_date", "p50", "p90", "p99", "sample_size"]) + .sort("snapshot_date") + .collect() +) +print(trend) diff --git a/scripts/hf_publisher/dataset_template/examples/03_duckdb.sql b/scripts/hf_publisher/dataset_template/examples/03_duckdb.sql new file mode 100644 index 00000000..4787f420 --- /dev/null +++ b/scripts/hf_publisher/dataset_template/examples/03_duckdb.sql @@ -0,0 +1,32 @@ +-- DuckDB can read Parquet directly from Hugging Face over httpfs. +-- Install once: +-- INSTALL httpfs; LOAD httpfs; +-- Then run any query like the ones below. + +-- 1) Today's leader per benchmark, sorted by sample size +WITH latest AS ( + SELECT max(snapshot_date) AS d + FROM 'hf://datasets/OpenChainBench/benchmarks/headlines/**/*.parquet' +) +SELECT slug, leader_name, value, unit, sample_size +FROM 'hf://datasets/OpenChainBench/benchmarks/headlines/**/*.parquet' +WHERE snapshot_date = (SELECT d FROM latest) +ORDER BY sample_size DESC; + +-- 2) 7-day p50 trend for one bench / one provider +SELECT snapshot_date, p50, p90, p99, success_rate +FROM 'hf://datasets/OpenChainBench/benchmarks/providers/**/*.parquet' +WHERE bench_slug = 'bridge-quote-latency' + AND provider_slug = 'mobula' +ORDER BY snapshot_date DESC +LIMIT 7; + +-- 3) Sparkline for today, one bench +SELECT point_index, value +FROM 'hf://datasets/OpenChainBench/benchmarks/timeseries/**/*.parquet' +WHERE bench_slug = 'bridge-quote-latency' + AND snapshot_date = ( + SELECT max(snapshot_date) + FROM 'hf://datasets/OpenChainBench/benchmarks/timeseries/**/*.parquet' + ) +ORDER BY point_index; diff --git a/scripts/hf_publisher/dataset_template/schemas/headlines.schema.json b/scripts/hf_publisher/dataset_template/schemas/headlines.schema.json new file mode 100644 index 00000000..6042ebb6 --- /dev/null +++ b/scripts/hf_publisher/dataset_template/schemas/headlines.schema.json @@ -0,0 +1,39 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://huggingface.co/datasets/OpenChainBench/benchmarks/blob/main/schemas/headlines.schema.json", + "title": "OCB headlines row", + "description": "One row per (slug, snapshot_date). Lightweight headline feed.", + "type": "object", + "required": [ + "snapshot_date", + "captured_at", + "slug", + "title", + "category", + "metric", + "unit", + "status", + "schema_version" + ], + "properties": { + "snapshot_date": { "type": "string", "format": "date" }, + "captured_at": { "type": "string", "format": "date-time" }, + "slug": { "type": "string" }, + "title": { "type": "string" }, + "category": { "type": "string" }, + "metric": { "type": "string" }, + "unit": { "type": "string" }, + "status": { "type": "string", "enum": ["live", "draft", "insufficient"] }, + "value": { "type": ["number", "null"] }, + "leader_name": { "type": ["string", "null"] }, + "leader_slug": { "type": ["string", "null"] }, + "leader_value": { "type": ["number", "null"] }, + "sample_size": { "type": ["number", "null"] }, + "as_of": { "type": ["string", "null"], "format": "date-time" }, + "citation_url": { "type": "string", "format": "uri" }, + "stat_api_url": { "type": "string", "format": "uri" }, + "source_url": { "type": ["string", "null"], "format": "uri" }, + "license": { "type": "string", "const": "CC-BY-4.0" }, + "schema_version": { "type": "integer", "minimum": 1 } + } +} diff --git a/scripts/hf_publisher/dataset_template/schemas/providers.schema.json b/scripts/hf_publisher/dataset_template/schemas/providers.schema.json new file mode 100644 index 00000000..55e59959 --- /dev/null +++ b/scripts/hf_publisher/dataset_template/schemas/providers.schema.json @@ -0,0 +1,29 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://huggingface.co/datasets/OpenChainBench/benchmarks/blob/main/schemas/providers.schema.json", + "title": "OCB providers row", + "description": "One row per (bench, provider, snapshot_date). Detailed ranking with percentiles.", + "type": "object", + "required": [ + "snapshot_date", + "captured_at", + "bench_slug", + "provider_slug", + "schema_version" + ], + "properties": { + "snapshot_date": { "type": "string", "format": "date" }, + "captured_at": { "type": "string", "format": "date-time" }, + "bench_slug": { "type": "string" }, + "provider_name": { "type": "string" }, + "provider_slug": { "type": "string" }, + "p50": { "type": ["number", "null"] }, + "p90": { "type": ["number", "null"] }, + "p99": { "type": ["number", "null"] }, + "mean": { "type": ["number", "null"] }, + "success_rate": { "type": ["number", "null"] }, + "sample_size": { "type": ["number", "null"] }, + "is_leader": { "type": "boolean" }, + "schema_version": { "type": "integer", "minimum": 1 } + } +} diff --git a/scripts/hf_publisher/dataset_template/schemas/timeseries.schema.json b/scripts/hf_publisher/dataset_template/schemas/timeseries.schema.json new file mode 100644 index 00000000..0f2d3044 --- /dev/null +++ b/scripts/hf_publisher/dataset_template/schemas/timeseries.schema.json @@ -0,0 +1,25 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://huggingface.co/datasets/OpenChainBench/benchmarks/blob/main/schemas/timeseries.schema.json", + "title": "OCB timeseries row", + "description": "One row per (bench, point_index, snapshot_date). 24h sparkline series.", + "type": "object", + "required": [ + "snapshot_date", + "captured_at", + "bench_slug", + "point_index", + "value", + "window", + "schema_version" + ], + "properties": { + "snapshot_date": { "type": "string", "format": "date" }, + "captured_at": { "type": "string", "format": "date-time" }, + "bench_slug": { "type": "string" }, + "point_index": { "type": "integer", "minimum": 0 }, + "value": { "type": "number" }, + "window": { "type": "string", "enum": ["24h", "7d", "30d"] }, + "schema_version": { "type": "integer", "minimum": 1 } + } +} diff --git a/scripts/hf_publisher/publish.py b/scripts/hf_publisher/publish.py new file mode 100644 index 00000000..9c088ab8 --- /dev/null +++ b/scripts/hf_publisher/publish.py @@ -0,0 +1,402 @@ +""" +Daily snapshot publisher for the Hugging Face dataset +`OpenChainBench/benchmarks`. + +Reads the live citable JSON API plus per-bench detail, projects it into +three Hive-partitioned Parquet tables (headlines, providers, timeseries) +keyed by snapshot_date, and pushes the new partitions to the HF dataset +repo. + +Why three tables and not one wide table: + headlines - 1 row per (slug, date). Light. The "who leads" feed used + by LLM agents and journalists. Cheap to scan. + providers - 1 row per (slug, provider, date). Detailed per-provider + ranking with p50/p90/p99. Used by devs comparing options. + timeseries - 1 row per (slug, point_index, date) holding the 24h + sparkline. Useful for analytical workloads. Separated so + consumers can ignore it if they only want headlines. + +Quorum guard: refuses to publish if /api/citable returns fewer than half +its declared count as live. The previous good snapshot stays as truth +on HF instead of being overwritten by a degraded one. + +Schema versioning: each table embeds a `schema_version` int column. +Bump it when adding columns. Never remove columns. Never rename. The +HF dataset is a long-lived public artifact - downstream consumers will +write queries that assume column names are stable. + +Idempotency: same snapshot_date overwrites itself. Re-running the cron +for a given day is safe. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import os +import sys +import urllib.error +import urllib.request +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable + +import pandas as pd +import pyarrow as pa +import pyarrow.parquet as pq + +logger = logging.getLogger("hf_publisher") + +# Bump together with any additive schema change in the row builders below. +# Never decrement. Never re-use a version for a breaking change - if a +# breaking change is unavoidable, ship a parallel `headlines_v2/` folder +# while keeping the v1 partitions readable for old consumers. +SCHEMA_VERSION = 1 + +# Minimum count of live benches in /api/citable to allow publishing. The +# bench registry sits around 26; a snapshot with <50% live is considered +# degraded and refused. Floor of 8 avoids tripping during early-stage +# dev where the registry is intentionally small. +QUORUM_MIN_LIVE = 8 +QUORUM_MIN_RATIO = 0.5 + +DEFAULT_API_BASE = "https://openchainbench.com" +DEFAULT_REPO_ID = "OpenChainBench/benchmarks" +USER_AGENT = "ocb-hf-publisher/1.0 (+https://openchainbench.com)" + + +class PublisherError(Exception): + """Raised when the snapshot is unfit to publish.""" + + +@dataclass(frozen=True) +class Snapshot: + date: str # ISO date, partition key value + captured_at: str # ISO timestamp UTC, embedded in every row + + +def fetch_json(url: str, timeout: float = 30.0) -> dict[str, Any]: + req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + payload = resp.read().decode("utf-8") + except urllib.error.HTTPError as e: + raise PublisherError(f"GET {url} returned HTTP {e.code}") from e + except urllib.error.URLError as e: + raise PublisherError(f"GET {url} failed: {e.reason}") from e + try: + return json.loads(payload) + except json.JSONDecodeError as e: + raise PublisherError(f"GET {url} returned non-JSON body") from e + + +def validate_quorum(citable: dict[str, Any]) -> None: + count = int(citable.get("count") or 0) + benches = citable.get("benchmarks") or [] + live = sum(1 for b in benches if b.get("status") == "live") + if count < QUORUM_MIN_LIVE: + raise PublisherError( + f"degraded source: count={count} below minimum {QUORUM_MIN_LIVE}" + ) + if live < QUORUM_MIN_LIVE or live / max(count, 1) < QUORUM_MIN_RATIO: + raise PublisherError( + f"degraded source: only {live}/{count} live (<{QUORUM_MIN_RATIO:.0%})" + ) + logger.info("quorum ok: %d/%d live", live, count) + + +def build_headlines( + citable: dict[str, Any], + snap: Snapshot, +) -> pd.DataFrame: + rows: list[dict[str, Any]] = [] + for b in citable.get("benchmarks", []): + leader = b.get("leader") or {} + rows.append( + { + "snapshot_date": snap.date, + "captured_at": snap.captured_at, + "slug": b.get("slug"), + "title": b.get("title"), + "category": b.get("category"), + "metric": b.get("metric"), + "unit": b.get("unit"), + "status": b.get("status"), + "value": _f(b.get("value")), + "leader_name": leader.get("name"), + "leader_slug": leader.get("slug"), + "leader_value": _f(leader.get("value")), + "sample_size": _f(b.get("sampleSize")), + "as_of": b.get("asOf"), + "citation_url": b.get("url"), + "stat_api_url": b.get("api"), + "source_url": b.get("source"), + "license": b.get("license"), + "schema_version": SCHEMA_VERSION, + } + ) + return pd.DataFrame(rows) + + +def build_providers( + stats: Iterable[dict[str, Any]], + snap: Snapshot, +) -> pd.DataFrame: + rows: list[dict[str, Any]] = [] + for stat in stats: + slug = stat.get("slug") + leader_slug = (stat.get("leader") or {}).get("slug") + for r in stat.get("rankings") or []: + ms = r.get("ms") or {} + rows.append( + { + "snapshot_date": snap.date, + "captured_at": snap.captured_at, + "bench_slug": slug, + "provider_name": r.get("name"), + "provider_slug": r.get("slug"), + "p50": _f(ms.get("p50")), + "p90": _f(ms.get("p90")), + "p99": _f(ms.get("p99")), + "mean": _f(ms.get("mean")), + "success_rate": _f(r.get("successRate")), + "sample_size": _f(r.get("sampleSize")), + "is_leader": r.get("slug") == leader_slug, + "schema_version": SCHEMA_VERSION, + } + ) + return pd.DataFrame(rows) + + +def build_timeseries( + stats: Iterable[dict[str, Any]], + snap: Snapshot, +) -> pd.DataFrame: + rows: list[dict[str, Any]] = [] + for stat in stats: + slug = stat.get("slug") + spark = stat.get("sparkline") or [] + for idx, value in enumerate(spark): + if value is None: + continue + rows.append( + { + "snapshot_date": snap.date, + "captured_at": snap.captured_at, + "bench_slug": slug, + "point_index": idx, + "value": _f(value), + "window": "24h", + "schema_version": SCHEMA_VERSION, + } + ) + return pd.DataFrame(rows) + + +def _f(v: Any) -> float | None: + if v is None: + return None + try: + return float(v) + except (TypeError, ValueError): + return None + + +def write_partition(df: pd.DataFrame, root: Path, table: str, snap: Snapshot) -> Path: + """Write a single Hive partition: //snapshot_date=/part-0.parquet. + + Snappy + ZSTD: Snappy is wider compatible (most readers default), + ZSTD compresses better. We use ZSTD because Polars/DuckDB/PyArrow + all read it natively now and the size delta matters at scale. + """ + target_dir = root / table / f"snapshot_date={snap.date}" + target_dir.mkdir(parents=True, exist_ok=True) + target = target_dir / "part-0.parquet" + table_ar = pa.Table.from_pandas(df, preserve_index=False) + pq.write_table(table_ar, target, compression="zstd") + logger.info("wrote %s rows=%d size=%dKB", target, len(df), target.stat().st_size // 1024) + return target + + +def post_slack(webhook: str | None, text: str) -> None: + if not webhook: + return + data = json.dumps({"text": text}).encode("utf-8") + req = urllib.request.Request( + webhook, + data=data, + headers={"Content-Type": "application/json", "User-Agent": USER_AGENT}, + ) + try: + with urllib.request.urlopen(req, timeout=10) as resp: + resp.read() + except Exception as e: + # Slack failure must not mask the underlying error + logger.warning("slack notify failed: %s", e) + + +def push_to_hf( + out_root: Path, + repo_id: str, + token: str, + commit_message: str, +) -> str: + """Push the staged dataset folder to HF Hub. Creates the repo if it + doesn't exist. Returns the commit hash.""" + from huggingface_hub import HfApi + + api = HfApi(token=token) + api.create_repo( + repo_id=repo_id, + repo_type="dataset", + exist_ok=True, + private=False, + ) + # upload_folder commits everything below `folder_path` keeping the + # relative paths. Hive partitions therefore land at the right place. + info = api.upload_folder( + folder_path=str(out_root), + repo_id=repo_id, + repo_type="dataset", + commit_message=commit_message, + ) + return info.oid if hasattr(info, "oid") else "unknown" + + +def stage_static_assets(out_root: Path, template_root: Path, snap: Snapshot) -> None: + """Copy README.md + CITATION.cff + LICENSE + schemas + examples into + the upload folder. Templates may include `{{date}}` placeholders.""" + import shutil + + if not template_root.is_dir(): + return + for src in template_root.rglob("*"): + if not src.is_file(): + continue + rel = src.relative_to(template_root) + dst = out_root / rel + dst.parent.mkdir(parents=True, exist_ok=True) + if src.suffix in {".md", ".cff", ".json", ".py", ".sql"}: + text = src.read_text(encoding="utf-8") + text = ( + text.replace("{{snapshot_date}}", snap.date) + .replace("{{captured_at}}", snap.captured_at) + .replace("{{schema_version}}", str(SCHEMA_VERSION)) + ) + dst.write_text(text, encoding="utf-8") + else: + shutil.copy2(src, dst) + + +def run( + api_base: str, + repo_id: str, + token: str | None, + out_root: Path, + template_root: Path, + dry_run: bool, + slack_webhook: str | None, +) -> None: + snap = Snapshot( + date=datetime.now(timezone.utc).strftime("%Y-%m-%d"), + captured_at=datetime.now(timezone.utc).isoformat(timespec="seconds"), + ) + logger.info("publishing snapshot %s", snap.date) + + citable = fetch_json(f"{api_base}/api/citable") + validate_quorum(citable) + + live_slugs = [ + b["slug"] + for b in citable.get("benchmarks", []) + if b.get("status") == "live" and b.get("slug") + ] + stats: list[dict[str, Any]] = [] + for slug in live_slugs: + try: + stats.append(fetch_json(f"{api_base}/api/stat/{slug}")) + except PublisherError as e: + # One bad per-slug fetch shouldn't abort the run. Log and skip. + logger.warning("skip per-slug fetch for %s: %s", slug, e) + + headlines = build_headlines(citable, snap) + providers = build_providers(stats, snap) + timeseries = build_timeseries(stats, snap) + + if headlines.empty: + raise PublisherError("empty headlines table - refusing to publish") + + write_partition(headlines, out_root, "headlines", snap) + write_partition(providers, out_root, "providers", snap) + write_partition(timeseries, out_root, "timeseries", snap) + stage_static_assets(out_root, template_root, snap) + + if dry_run: + logger.info("dry-run: skipping HF push, files staged at %s", out_root) + return + + if not token: + raise PublisherError("HF_TOKEN missing in non-dry-run mode") + + commit_message = f"snapshot {snap.date} (rows: h={len(headlines)} p={len(providers)} ts={len(timeseries)})" + oid = push_to_hf(out_root, repo_id, token, commit_message) + logger.info("pushed to HF: %s commit=%s", repo_id, oid) + post_slack( + slack_webhook, + f":white_check_mark: OCB HF snapshot {snap.date} published " + f"(h={len(headlines)}, p={len(providers)}, ts={len(timeseries)}) " + f"https://huggingface.co/datasets/{repo_id}/tree/main", + ) + + +def main() -> int: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + ) + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--api-base", default=os.environ.get("OCB_API", DEFAULT_API_BASE)) + p.add_argument("--repo-id", default=os.environ.get("HF_REPO_ID", DEFAULT_REPO_ID)) + p.add_argument("--out", default=os.environ.get("OCB_OUT", "/tmp/ocb-hf-staging")) + p.add_argument( + "--template", + default=str(Path(__file__).parent / "dataset_template"), + help="Static files copied into the dataset (README, CITATION, schemas, examples).", + ) + p.add_argument( + "--dry-run", + action="store_true", + help="Skip HF push. Use for local + CI checks.", + ) + args = p.parse_args() + + token = os.environ.get("HF_TOKEN") + slack = os.environ.get("SLACK_WEBHOOK_URL") + out_root = Path(args.out) + out_root.mkdir(parents=True, exist_ok=True) + + try: + run( + api_base=args.api_base, + repo_id=args.repo_id, + token=token, + out_root=out_root, + template_root=Path(args.template), + dry_run=args.dry_run, + slack_webhook=slack, + ) + return 0 + except PublisherError as e: + logger.error("publisher aborted: %s", e) + post_slack(slack, f":x: OCB HF publisher aborted: {e}") + return 2 + except Exception as e: + logger.exception("publisher crashed: %s", e) + post_slack(slack, f":fire: OCB HF publisher crashed: {e}") + return 3 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/hf_publisher/requirements.txt b/scripts/hf_publisher/requirements.txt new file mode 100644 index 00000000..61c512fd --- /dev/null +++ b/scripts/hf_publisher/requirements.txt @@ -0,0 +1,3 @@ +huggingface_hub>=0.30,<1.0 +pandas>=2.2,<3.0 +pyarrow>=18.0,<22.0 diff --git a/scripts/hf_publisher/test_publish.py b/scripts/hf_publisher/test_publish.py new file mode 100644 index 00000000..911c0b59 --- /dev/null +++ b/scripts/hf_publisher/test_publish.py @@ -0,0 +1,201 @@ +""" +Offline tests for the HF publisher. + +These never hit the live API or HF Hub. They drive the row-builders + +quorum guard with hand-crafted payloads so a CI run can catch schema +regressions before they propagate to the public dataset. +""" + +from __future__ import annotations + +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory + +from publish import ( + PublisherError, + SCHEMA_VERSION, + Snapshot, + build_headlines, + build_providers, + build_timeseries, + stage_static_assets, + validate_quorum, + write_partition, +) + + +def _snap(date: str = "2026-06-22") -> Snapshot: + return Snapshot(date=date, captured_at=f"{date}T00:00:00+00:00") + + +def _citable_fixture(live_count: int = 20, total: int = 26) -> dict: + benches = [] + for i in range(total): + status = "live" if i < live_count else "insufficient" + benches.append( + { + "slug": f"bench-{i}", + "title": f"Bench {i}", + "category": "RPCs", + "metric": "Latency", + "unit": "ms", + "status": status, + "value": 100.5 + i if status == "live" else None, + "leader": ( + {"name": "Mobula", "slug": "mobula", "value": 100.5 + i} + if status == "live" + else None + ), + "sampleSize": 1000 + i, + "asOf": "2026-06-22T00:00:00.000Z", + "url": f"https://openchainbench.com/benchmarks/bench-{i}", + "api": f"https://openchainbench.com/api/stat/bench-{i}", + "source": "https://github.com/ChainBench/OpenChainBench/blob/main/benchmarks/bench-0.yml", + "license": "CC-BY-4.0", + } + ) + return {"count": total, "benchmarks": benches} + + +def _stat_fixture(slug: str = "bench-0", n_providers: int = 3, sparkline_len: int = 72) -> dict: + return { + "slug": slug, + "leader": {"name": "Mobula", "slug": "mobula"}, + "rankings": [ + { + "name": f"Provider {i}", + "slug": f"provider-{i}", + "ms": { + "p50": 100.0 + i, + "p90": 200.0 + i, + "p99": 500.0 + i, + "mean": 150.0 + i, + }, + "successRate": 99.0 - i * 0.1, + "sampleSize": 5000 - i * 100, + } + for i in range(n_providers) + ], + "sparkline": [100.0 + (j % 10) for j in range(sparkline_len)], + } + + +class QuorumTests(unittest.TestCase): + def test_passes_at_full_live(self): + validate_quorum(_citable_fixture(live_count=26, total=26)) + + def test_passes_at_half(self): + validate_quorum(_citable_fixture(live_count=13, total=26)) + + def test_refuses_below_half(self): + with self.assertRaises(PublisherError): + validate_quorum(_citable_fixture(live_count=12, total=26)) + + def test_refuses_below_floor(self): + with self.assertRaises(PublisherError): + validate_quorum(_citable_fixture(live_count=7, total=20)) + + def test_refuses_empty(self): + with self.assertRaises(PublisherError): + validate_quorum({"count": 0, "benchmarks": []}) + + +class HeadlinesTests(unittest.TestCase): + def test_columns_stable(self): + df = build_headlines(_citable_fixture(), _snap()) + expected_cols = { + "snapshot_date", + "captured_at", + "slug", + "title", + "category", + "metric", + "unit", + "status", + "value", + "leader_name", + "leader_slug", + "leader_value", + "sample_size", + "as_of", + "citation_url", + "stat_api_url", + "source_url", + "license", + "schema_version", + } + self.assertEqual(set(df.columns), expected_cols) + + def test_schema_version_present(self): + df = build_headlines(_citable_fixture(), _snap()) + self.assertTrue((df["schema_version"] == SCHEMA_VERSION).all()) + + def test_insufficient_rows_have_null_value(self): + df = build_headlines(_citable_fixture(live_count=10, total=12), _snap()) + live = df[df["status"] == "live"] + insufficient = df[df["status"] == "insufficient"] + self.assertTrue(live["value"].notna().all()) + self.assertTrue(insufficient["value"].isna().all()) + + +class ProvidersTests(unittest.TestCase): + def test_one_row_per_provider(self): + df = build_providers([_stat_fixture(n_providers=5)], _snap()) + self.assertEqual(len(df), 5) + + def test_leader_flag(self): + df = build_providers([_stat_fixture(n_providers=3)], _snap()) + # Fixture leader is "mobula" but no provider has that slug, so 0 leaders. + # Sanity: column exists and is boolean dtype. + self.assertIn("is_leader", df.columns) + + def test_leader_flag_matches(self): + stat = _stat_fixture(n_providers=3) + stat["leader"] = {"name": "Provider 0", "slug": "provider-0"} + df = build_providers([stat], _snap()) + self.assertEqual(df[df["is_leader"]]["provider_slug"].tolist(), ["provider-0"]) + + +class TimeseriesTests(unittest.TestCase): + def test_one_row_per_point(self): + df = build_timeseries([_stat_fixture(sparkline_len=72)], _snap()) + self.assertEqual(len(df), 72) + + def test_skips_nulls(self): + stat = _stat_fixture(sparkline_len=5) + stat["sparkline"] = [1.0, None, 3.0, None, 5.0] + df = build_timeseries([stat], _snap()) + self.assertEqual(len(df), 3) + + +class PartitioningTests(unittest.TestCase): + def test_write_partition_path_layout(self): + df = build_headlines(_citable_fixture(), _snap()) + with TemporaryDirectory() as tmp: + root = Path(tmp) + path = write_partition(df, root, "headlines", _snap("2026-06-22")) + self.assertTrue(path.exists()) + self.assertTrue( + path.as_posix().endswith( + "headlines/snapshot_date=2026-06-22/part-0.parquet" + ), + msg=path.as_posix(), + ) + + def test_template_substitution(self): + with TemporaryDirectory() as tmp: + tmpl = Path(tmp) / "tmpl" + tmpl.mkdir() + (tmpl / "README.md").write_text("date={{snapshot_date}} v={{schema_version}}") + out = Path(tmp) / "out" + out.mkdir() + stage_static_assets(out, tmpl, _snap("2026-06-22")) + self.assertEqual( + (out / "README.md").read_text(), + f"date=2026-06-22 v={SCHEMA_VERSION}", + ) + + +if __name__ == "__main__": + unittest.main()