scripts/run_fiscalqa_external_validation.py

#!/usr/bin/env python3
"""Evaluate TELEN's immutable version-index contract on FiscalQA Pro.

This is a retrieval-provenance evaluation, not an answer-generation score.  It
uses the released expert-reviewed R3 questions and their gold article-version
identifiers, and deliberately keeps deterministic temporal filtering separate
from dense semantic retrieval.
"""

from __future__ import annotations

import argparse
import ast
import hashlib
import json
import platform
import re
import subprocess
import sys
import unicodedata
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable

import numpy as np
import pandas as pd
import torch

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "src"))

from scripts.run_static_benchmark import load_or_encode, resolve_device  # noqa: E402
from telen.static_benchmark import sha256_file  # noqa: E402


UPSTREAM_REPOSITORY = "https://github.com/rosecymbler/fiscal-fr-bench.git"
UPSTREAM_COMMIT = "6ca3807d46926f6f5b355d2d435b4baa05300bdf"
DATA_LICENSE = "CC-BY-4.0"
CODE_LICENSE = "MIT"
MODEL_NAME = "BAAI/bge-m3"
MODEL_REVISION = "5617a9f61b028005a4858fdac845db406aefb181"
EXPECTED_SCORED_QUESTIONS = 209
EXPECTED_EXCLUDED_QUESTIONS = 12
EXPECTED_R3_QUESTIONS = 221
BOOTSTRAP_SEED = 20260829
CGI_PRINCIPAL_CODE_ID = "LEGITEXT000006069577"


def _sha256_text(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def _git_head(path: Path) -> str:
    completed = subprocess.run(
        ["git", "-C", str(path), "rev-parse", "HEAD"],
        check=True,
        capture_output=True,
        text=True,
    )
    return completed.stdout.strip()


def _read_id_file(path: Path) -> set[str]:
    return {
        line.strip()
        for line in path.read_text(encoding="utf-8").splitlines()
        if line.strip() and not line.lstrip().startswith("#")
    }


def _gold_dict(value: object) -> dict:
    if isinstance(value, dict):
        return value
    if isinstance(value, str):
        parsed = ast.literal_eval(value)
        if isinstance(parsed, dict):
            return parsed
    raise ValueError("FiscalQA gold must be a dictionary or dictionary literal")


def _load_scored_questions(root: Path) -> list[dict]:
    benchmark = root / "data/benchmark"
    raw = json.loads((benchmark / "questions.json").read_text(encoding="utf-8"))
    # Never expose or propagate the upstream contamination canary.
    r3 = [
        row
        for row in raw
        if row.get("qid") != "_CANARY_" and row.get("regime") == "R3"
    ]
    scored = _read_id_file(benchmark / "scored209_qids.txt")
    excluded = _read_id_file(benchmark / "excluded_qids_scope.txt")
    observed_ids = {str(row["qid"]) for row in r3}
    if len(r3) != EXPECTED_R3_QUESTIONS:
        raise ValueError(f"expected {EXPECTED_R3_QUESTIONS} R3 questions")
    if len(scored) != EXPECTED_SCORED_QUESTIONS:
        raise ValueError(f"expected {EXPECTED_SCORED_QUESTIONS} scored qids")
    if len(excluded) != EXPECTED_EXCLUDED_QUESTIONS:
        raise ValueError(f"expected {EXPECTED_EXCLUDED_QUESTIONS} excluded qids")
    if scored != observed_ids.difference(excluded):
        raise ValueError("scored qids are not exactly R3 minus scope exclusions")

    result = []
    for row in r3:
        if row["qid"] not in scored:
            continue
        gold = _gold_dict(row["gold"])
        required = {"version_id", "article_cid", "canonical_answer", "code"}
        if not required.issubset(gold):
            raise ValueError(f"question {row['qid']} has incomplete gold metadata")
        result.append(
            {
                "qid": str(row["qid"]),
                "prompt": str(row["prompt"]),
                "date_anchor": pd.Timestamp(row["date_anchor"]),
                "gold_version_id": str(gold["version_id"]),
                "gold_cid": str(gold["article_cid"]),
                "gold_code": str(gold["code"]),
            }
        )
    if len(result) != EXPECTED_SCORED_QUESTIONS:
        raise ValueError("scored question load is incomplete")
    return result


def _load_corpus(path: Path) -> pd.DataFrame:
    corpus = pd.read_parquet(path).copy()
    required = {
        "id",
        "cid",
        "num",
        "texte_clean",
        "date_debut",
        "date_fin",
        "code_id",
    }
    missing = required.difference(corpus.columns)
    if missing:
        raise ValueError(f"FiscalQA corpus is missing columns: {sorted(missing)}")
    if corpus["id"].duplicated().any():
        raise ValueError("FiscalQA corpus has duplicate version ids")
    if corpus[["id", "cid", "num", "date_debut", "date_fin"]].isna().any().any():
        raise ValueError("FiscalQA corpus has null version identity or interval fields")
    corpus["id"] = corpus["id"].astype(str)
    corpus["cid"] = corpus["cid"].astype(str)
    corpus["num"] = corpus["num"].astype(str)
    corpus["code_id"] = corpus["code_id"].astype(str)
    corpus["texte_clean"] = corpus["texte_clean"].fillna("").astype(str)
    # Légifrance contains MODIFIE_MORT_NE/ABROGE records whose end is not after
    # their start.  Preserve them in the faithful all-version corpus but
    # quarantine them from every snapshot rather than inventing a validity span.
    corpus["snapshot_eligible"] = corpus["date_fin"] > corpus["date_debut"]
    corpus = corpus.sort_values("id", kind="mergesort").reset_index(drop=True)
    return corpus


def _resolved_snapshot_mask(corpus: pd.DataFrame, as_of: pd.Timestamp) -> np.ndarray:
    """Choose the latest-starting valid version for every article identity.

    Three FiscalQA scored dates overlap two upstream intervals for their gold
    article.  Choosing the latest valid start is deterministic and recovers the
    upstream gold without reading a question label.
    """
    active = (
        corpus["snapshot_eligible"]
        & (corpus["date_debut"] <= as_of)
        & (as_of < corpus["date_fin"])
    )
    eligible = corpus.loc[active, ["cid", "date_debut", "id"]].copy()
    if eligible.empty:
        return np.zeros(len(corpus), dtype=bool)
    eligible["position"] = eligible.index
    selected = (
        eligible.sort_values(
            ["cid", "date_debut", "id"],
            ascending=[True, True, True],
            kind="mergesort",
        )
        .groupby("cid", sort=False, as_index=False)
        .tail(1)["position"]
        .to_numpy(dtype=np.int64)
    )
    mask = np.zeros(len(corpus), dtype=bool)
    mask[selected] = True
    return mask


def _interval_mask(corpus: pd.DataFrame, as_of: pd.Timestamp) -> np.ndarray:
    return np.asarray(
        corpus["snapshot_eligible"]
        & (corpus["date_debut"] <= as_of)
        & (as_of < corpus["date_fin"]),
        dtype=bool,
    )


def _normalize_tokens(value: str) -> tuple[str, ...]:
    normalized = unicodedata.normalize("NFKC", value).casefold()
    return tuple(re.findall(r"[0-9]+|[^\W\d_]+", normalized, flags=re.UNICODE))


def _cited_article_numbers(
    prompt: str,
    known_numbers: Iterable[str],
    *,
    max_number_tokens: int = 6,
) -> tuple[str, ...]:
    """Extract explicit ``article``/``art.`` citations without gold metadata."""
    lookup: dict[tuple[str, ...], set[str]] = defaultdict(set)
    for number in known_numbers:
        tokens = _normalize_tokens(number)
        if tokens:
            lookup[tokens].add(number)
    tokens = _normalize_tokens(prompt)
    candidates: set[str] = set()
    for position, token in enumerate(tokens):
        if token not in {"article", "articles", "art"}:
            continue
        longest = 0
        matches: set[str] = set()
        for width in range(1, max_number_tokens + 1):
            chunk = tokens[position + 1 : position + 1 + width]
            if len(chunk) != width:
                break
            observed = lookup.get(chunk, set())
            if observed:
                longest = width
                matches = set(observed)
        if longest:
            candidates.update(matches)
    return tuple(sorted(candidates))


def _rank_observation(
    scores: np.ndarray,
    candidate_mask: np.ndarray,
    version_ids: np.ndarray,
    cids: np.ndarray,
    *,
    gold_version_id: str,
    gold_cid: str,
) -> dict:
    positions = np.flatnonzero(candidate_mask)
    if not len(positions):
        return {
            "gold_rank": None,
            "gold_article_rank": None,
            "top1_version_id": None,
            "top1_cid": None,
            "ndcg_at_10": 0.0,
            "mrr": 0.0,
            "hit_at_1": 0.0,
            "hit_at_5": 0.0,
            "hit_at_10": 0.0,
            "article_hit_at_1": 0.0,
            "article_hit_at_5": 0.0,
            "article_hit_at_10": 0.0,
        }
    order_local = np.lexsort(
        (version_ids[positions], -np.asarray(scores[positions], dtype=np.float64))
    )
    ranking = positions[order_local]
    ranked_versions = version_ids[ranking]
    ranked_cids = cids[ranking]
    gold_positions = np.flatnonzero(ranked_versions == gold_version_id)
    article_positions = np.flatnonzero(ranked_cids == gold_cid)
    gold_rank = int(gold_positions[0]) + 1 if len(gold_positions) else None
    article_rank = int(article_positions[0]) + 1 if len(article_positions) else None
    return {
        "gold_rank": gold_rank,
        "gold_article_rank": article_rank,
        "top1_version_id": str(ranked_versions[0]),
        "top1_cid": str(ranked_cids[0]),
        "ndcg_at_10": (
            float(1.0 / np.log2(gold_rank + 1))
            if gold_rank is not None and gold_rank <= 10
            else 0.0
        ),
        "mrr": float(1.0 / gold_rank) if gold_rank is not None else 0.0,
        "hit_at_1": float(gold_rank == 1),
        "hit_at_5": float(gold_rank is not None and gold_rank <= 5),
        "hit_at_10": float(gold_rank is not None and gold_rank <= 10),
        "article_hit_at_1": float(article_rank == 1),
        "article_hit_at_5": float(article_rank is not None and article_rank <= 5),
        "article_hit_at_10": float(article_rank is not None and article_rank <= 10),
    }


METRIC_FIELDS = (
    "ndcg_at_10",
    "mrr",
    "hit_at_1",
    "hit_at_5",
    "hit_at_10",
    "article_hit_at_1",
    "article_hit_at_5",
    "article_hit_at_10",
)


def _aggregate(rows: list[dict]) -> dict:
    by_cid: dict[str, list[dict]] = defaultdict(list)
    for row in rows:
        by_cid[row["gold_cid"]].append(row)
    aggregate = {field: float(np.mean([row[field] for row in rows])) for field in METRIC_FIELDS}
    aggregate.update(
        {
            "query_count": len(rows),
            "article_cluster_count": len(by_cid),
            "article_macro_ndcg_at_10": float(
                np.mean(
                    [
                        np.mean([row["ndcg_at_10"] for row in cluster])
                        for cluster in by_cid.values()
                    ]
                )
            ),
            "gold_missing_from_candidates": int(
                sum(row["gold_rank"] is None for row in rows)
            ),
            "top1_temporal_misgrounding": float(
                np.mean(
                    [
                        row["top1_cid"] == row["gold_cid"]
                        and row["top1_version_id"] != row["gold_version_id"]
                        for row in rows
                    ]
                )
            ),
        }
    )
    return aggregate


def _clustered_difference(
    candidate_rows: list[dict],
    baseline_rows: list[dict],
    *,
    field: str = "ndcg_at_10",
    samples: int = 10_000,
) -> dict:
    if [row["qid"] for row in candidate_rows] != [row["qid"] for row in baseline_rows]:
        raise ValueError("paired rows are not aligned")
    grouped: dict[str, list[float]] = defaultdict(list)
    for candidate, baseline in zip(candidate_rows, baseline_rows, strict=True):
        if candidate["gold_cid"] != baseline["gold_cid"]:
            raise ValueError("paired rows have different article clusters")
        grouped[candidate["gold_cid"]].append(candidate[field] - baseline[field])
    cluster_values = np.asarray(
        [np.mean(grouped[cid]) for cid in sorted(grouped)], dtype=np.float64
    )
    rng = np.random.default_rng(BOOTSTRAP_SEED)
    draws = rng.choice(
        cluster_values,
        size=(samples, len(cluster_values)),
        replace=True,
    ).mean(axis=1)
    return {
        "metric": field,
        "estimand": "paired article-macro candidate minus baseline",
        "article_cluster_count": int(len(cluster_values)),
        "observed_difference": float(cluster_values.mean()),
        "bootstrap_samples": samples,
        "bootstrap_seed": BOOTSTRAP_SEED,
        "ci_95": [float(np.quantile(draws, 0.025)), float(np.quantile(draws, 0.975))],
    }


def run(args: argparse.Namespace) -> dict:
    upstream_head = _git_head(args.upstream_root)
    if upstream_head != UPSTREAM_COMMIT:
        raise ValueError(
            f"FiscalQA upstream commit drifted: expected {UPSTREAM_COMMIT}, got {upstream_head}"
        )
    corpus_path = args.upstream_root / "data/articles_cgi_versioned.parquet"
    question_path = args.upstream_root / "data/benchmark/questions.json"
    scored_path = args.upstream_root / "data/benchmark/scored209_qids.txt"
    excluded_path = args.upstream_root / "data/benchmark/excluded_qids_scope.txt"
    license_path = args.upstream_root / "LICENSE-DATA"
    corpus = _load_corpus(corpus_path)
    questions = _load_scored_questions(args.upstream_root)

    version_ids = corpus["id"].to_numpy(dtype=str)
    cids = corpus["cid"].to_numpy(dtype=str)
    numbers = corpus["num"].to_numpy(dtype=str)
    code_ids = corpus["code_id"].to_numpy(dtype=str)
    id_to_position = {version_id: position for position, version_id in enumerate(version_ids)}
    if len(id_to_position) != len(version_ids):
        raise ValueError("corpus version ids are not unique after normalization")
    for question in questions:
        position = id_to_position.get(question["gold_version_id"])
        if position is None:
            raise ValueError(f"missing gold version {question['gold_version_id']}")
        if cids[position] != question["gold_cid"]:
            raise ValueError(f"gold CID mismatch for {question['qid']}")
        if not _interval_mask(corpus, question["date_anchor"])[position]:
            raise ValueError(f"gold version is not date-valid for {question['qid']}")
    if {question["gold_code"] for question in questions} != {"CGI"}:
        raise ValueError("the frozen R3 scored slice is no longer CGI-principal only")

    version_texts = tuple(
        f"Article {row.num}\n{row.texte_clean}" for row in corpus.itertuples(index=False)
    )
    query_ids = tuple(question["qid"] for question in questions)
    query_texts = tuple(question["prompt"] for question in questions)
    device = resolve_device(args.device)
    version_embeddings, version_cache = load_or_encode(
        version_texts,
        tuple(version_ids),
        role="fiscalqa-pro-article-versions",
        cache_dir=args.cache_dir,
        model_name=args.model_name,
        revision=args.model_revision,
        device=device,
        max_length=args.max_length,
        batch_size=args.corpus_batch_size,
    )
    query_embeddings, query_cache = load_or_encode(
        query_texts,
        query_ids,
        role="fiscalqa-pro-r3-scored-queries",
        cache_dir=args.cache_dir,
        model_name=args.model_name,
        revision=args.model_revision,
        device=device,
        max_length=args.max_length,
        batch_size=args.query_batch_size,
    )
    scores = np.asarray(query_embeddings, dtype=np.float32) @ np.asarray(
        version_embeddings, dtype=np.float32
    ).T

    current_mask = _resolved_snapshot_mask(corpus, pd.Timestamp(args.current_snapshot))
    known_numbers = tuple(sorted(set(numbers)))
    conditions: dict[str, list[dict]] = defaultdict(list)
    citation_audit = []
    all_mask = np.ones(len(corpus), dtype=bool)
    for row_index, question in enumerate(questions):
        date = question["date_anchor"]
        interval_mask = _interval_mask(corpus, date)
        resolved_mask = _resolved_snapshot_mask(corpus, date)
        cited_numbers = _cited_article_numbers(question["prompt"], known_numbers)
        cited_mask = resolved_mask & np.isin(numbers, np.asarray(cited_numbers, dtype=str))
        declared_cgi_mask = resolved_mask & (code_ids == CGI_PRINCIPAL_CODE_ID)
        cited_cgi_mask = declared_cgi_mask & np.isin(
            numbers, np.asarray(cited_numbers, dtype=str)
        )
        citation_or_dense_cgi_mask = (
            cited_cgi_mask if cited_numbers else declared_cgi_mask
        )
        oracle_mask = resolved_mask & (cids == question["gold_cid"])
        masks = {
            "all_versions_dense": all_mask,
            "current_snapshot_dense": current_mask,
            "date_interval_dense": interval_mask,
            "date_resolved_dense": resolved_mask,
            "explicit_citation_date_resolved_dense": cited_mask,
            "declared_r3_cgi_scope_date_resolved_dense": declared_cgi_mask,
            "citation_or_dense_cgi_date_resolved": citation_or_dense_cgi_mask,
            "oracle_cid_date_resolved_dense": oracle_mask,
        }
        citation_audit.append(
            {
                "qid": question["qid"],
                "cited_number_count": len(cited_numbers),
                "gold_number_detected": bool(
                    numbers[id_to_position[question["gold_version_id"]]] in cited_numbers
                ),
            }
        )
        for condition, mask in masks.items():
            observation = _rank_observation(
                scores[row_index],
                mask,
                version_ids,
                cids,
                gold_version_id=question["gold_version_id"],
                gold_cid=question["gold_cid"],
            )
            observation.update(
                {
                    "qid": question["qid"],
                    "gold_cid": question["gold_cid"],
                    "gold_version_id": question["gold_version_id"],
                    "date_anchor": date.date().isoformat(),
                }
            )
            conditions[condition].append(observation)

    aggregates = {name: _aggregate(rows) for name, rows in conditions.items()}
    comparisons = {
        "date_resolved_vs_current": _clustered_difference(
            conditions["date_resolved_dense"],
            conditions["current_snapshot_dense"],
        ),
        "date_resolved_vs_all_versions": _clustered_difference(
            conditions["date_resolved_dense"],
            conditions["all_versions_dense"],
        ),
        "citation_resolver_vs_date_resolved": _clustered_difference(
            conditions["explicit_citation_date_resolved_dense"],
            conditions["date_resolved_dense"],
        ),
        "declared_cgi_scope_vs_date_resolved": _clustered_difference(
            conditions["declared_r3_cgi_scope_date_resolved_dense"],
            conditions["date_resolved_dense"],
        ),
        "citation_or_dense_cgi_vs_declared_cgi_scope": _clustered_difference(
            conditions["citation_or_dense_cgi_date_resolved"],
            conditions["declared_r3_cgi_scope_date_resolved_dense"],
        ),
        "citation_or_dense_cgi_vs_date_resolved": _clustered_difference(
            conditions["citation_or_dense_cgi_date_resolved"],
            conditions["date_resolved_dense"],
        ),
        "citation_or_dense_cgi_vs_current": _clustered_difference(
            conditions["citation_or_dense_cgi_date_resolved"],
            conditions["current_snapshot_dense"],
        ),
        "citation_or_dense_cgi_vs_all_versions": _clustered_difference(
            conditions["citation_or_dense_cgi_date_resolved"],
            conditions["all_versions_dense"],
        ),
    }
    receipt = {
        "schema_version": 1,
        "artifact_type": "telen_fiscalqa_external_retrieval_validation",
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "claim_scope": {
            "retrieval_provenance_evaluated": True,
            "answer_generation_evaluated": False,
            "learned_graph_residual_evaluated": False,
            "vietnamese_external_validity_evaluated": False,
            "interpretation": (
                "External expert-reviewed French temporal retrieval diagnostic; "
                "it validates version resolution but cannot replace Vietnamese gold qrels."
            ),
        },
        "upstream": {
            "repository": UPSTREAM_REPOSITORY,
            "commit": upstream_head,
            "data_license": DATA_LICENSE,
            "code_license": CODE_LICENSE,
            "contamination_canary_policy": "header skipped and never copied to outputs",
            "files": {
                "corpus": {"path": str(corpus_path), "sha256": sha256_file(corpus_path)},
                "questions": {"path": str(question_path), "sha256": sha256_file(question_path)},
                "scored_qids": {"path": str(scored_path), "sha256": sha256_file(scored_path)},
                "excluded_qids": {"path": str(excluded_path), "sha256": sha256_file(excluded_path)},
                "data_license": {"path": str(license_path), "sha256": sha256_file(license_path)},
            },
        },
        "dataset": {
            "corpus_version_count": len(corpus),
            "article_identity_count": int(corpus["cid"].nunique()),
            "snapshot_ineligible_interval_count": int(
                (~corpus["snapshot_eligible"]).sum()
            ),
            "scored_question_count": len(questions),
            "scored_article_cluster_count": len({q["gold_cid"] for q in questions}),
            "date_range": [
                min(q["date_anchor"] for q in questions).date().isoformat(),
                max(q["date_anchor"] for q in questions).date().isoformat(),
            ],
            "gold_interval_checks_passed": len(questions),
            "explicit_citation_gold_number_coverage": float(
                np.mean([row["gold_number_detected"] for row in citation_audit])
            ),
        },
        "model": {
            "name": args.model_name,
            "revision": args.model_revision,
            "device": str(device),
            "max_length": args.max_length,
            "version_embedding_cache": str(version_cache),
            "version_embedding_cache_sha256": sha256_file(version_cache),
            "query_embedding_cache": str(query_cache),
            "query_embedding_cache_sha256": sha256_file(query_cache),
            "version_text_order_sha256": _sha256_text("\n".join(version_ids)),
            "query_order_sha256": _sha256_text("\n".join(query_ids)),
        },
        "protocol": {
            "current_snapshot": args.current_snapshot,
            "declared_r3_code_scope": {
                "label": "CGI",
                "code_id": CGI_PRINCIPAL_CODE_ID,
                "source": "upstream R3 benchmark definition",
            },
            "validity_interval": "date_debut <= date_anchor < date_fin",
            "overlap_resolution": "latest date_debut, then stable version id",
            "ranking_ties": "stable version id ascending",
            "cluster_unit": "gold article CID",
            "bootstrap_samples": 10_000,
            "bootstrap_seed": BOOTSTRAP_SEED,
            "analysis_status": "external_post_hoc_diagnostic_not_confirmatory",
            "protocol_amendment": (
                "Declared-CGI-scope and citation-or-dense baselines were added "
                "after the first date-resolved result was visible, but before "
                "testing any learned external residual; they are reported as "
                "post-hoc transparent strong baselines."
            ),
        },
        "results": aggregates,
        "comparisons": comparisons,
        "environment": {
            "python": platform.python_version(),
            "numpy": np.__version__,
            "pandas": pd.__version__,
            "torch": torch.__version__,
            "cuda_available": torch.cuda.is_available(),
        },
        "limitations": [
            "Retrieval provenance is not FiscalQA strict answer accuracy.",
            "Most R3 prompts explicitly cite the target article, so the citation resolver is a necessary strong rule baseline.",
            "The benchmark is French tax law and does not establish Vietnamese legal relevance.",
            "This run does not train or credit the learned local graph residual.",
            "Validity intervals are supplied by the benchmark; uncertainty in event extraction is not tested.",
        ],
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(receipt, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    return receipt


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--upstream-root",
        type=Path,
        default=REPO_ROOT / "tmp/deep-research/fiscal-fr-bench",
    )
    parser.add_argument(
        "--cache-dir",
        type=Path,
        default=REPO_ROOT / "tmp/telen/static-benchmarks/cache/fiscalqa",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=REPO_ROOT
        / "research/results/telen_fiscalqa_external_validation_2026-08-29.json",
    )
    parser.add_argument("--model-name", default=MODEL_NAME)
    parser.add_argument("--model-revision", default=MODEL_REVISION)
    parser.add_argument("--device", default="auto")
    parser.add_argument("--max-length", type=int, default=2048)
    parser.add_argument("--corpus-batch-size", type=int, default=8)
    parser.add_argument("--query-batch-size", type=int, default=32)
    parser.add_argument("--current-snapshot", default="2026-08-05")
    return parser.parse_args()


def main() -> None:
    receipt = run(parse_args())
    print(json.dumps(receipt, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()