"""Read-only check of Gregory Young's Apache-2.0 OpenMarket demo sample.
Tested with Python 3.12 / pyarrow 19.0.1. No network, wallet or trading calls.
Dataset revision: 74502466d1a7cef56395bfd8d0b465fbebc849cf.
Pass the downloaded lag_pairs_ms.parquet path as the only argument.
"""
import hashlib
import sys
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path

import pyarrow.parquet as pq

EXPECTED_SHA256 = "dd5a7b4c1f0ab82fa355a86325459359399f21029ee3a7eb578c0e994e4ad2d4"
path = Path(sys.argv[1])
if path.stat().st_size > 1_048_576:
    raise SystemExit("Sample exceeds the 1 MiB limit")
raw = path.read_bytes()
if hashlib.sha256(raw).hexdigest() != EXPECTED_SHA256:
    raise SystemExit("Sample checksum differs from the inspected revision")
parquet = pq.ParquetFile(path)
if parquet.metadata.num_rows > 10_000:
    raise SystemExit("Sample exceeds the 10,000-row limit")
rows = parquet.read().to_pylist()
required = ["paired_at_ms", "binance_source_ts_ms", "polymarket_source_ts_ms", "lead_lag_ms"]
missing = sum(any(row.get(key) is None for key in required) for row in rows)
if not rows or missing:
    raise SystemExit("Empty sample or missing required timestamps")

def utc(ms):
    return datetime.fromtimestamp(ms / 1000, timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")

def span(key):
    return f"{utc(min(row[key] for row in rows))}..{utc(max(row[key] for row in rows))}"

mismatches = sum(row["lead_lag_ms"] != row["polymarket_source_ts_ms"] - row["binance_source_ts_ms"] for row in rows)
print(f"rows={len(rows)}; bytes={len(raw)}; missing_required={missing}")
print(f"paired_at_utc={span('paired_at_ms')}")
print(f"binance_source_utc={span('binance_source_ts_ms')}; polymarket_source_utc={span('polymarket_source_ts_ms')}")
print(f"lead_lag_ms={min(row['lead_lag_ms'] for row in rows)}..{max(row['lead_lag_ms'] for row in rows)}; formula_mismatches={mismatches}")
print(f"quality_flag:{parquet.schema_arrow.field('quality_flag').type}={dict(sorted(Counter(row['quality_flag'] for row in rows).items()))}; side_label={sorted(set(row['side_label'] for row in rows))}")