| |
| """Score a Task 3 submission with the official FinMR A/S/E/C evaluator. |
| |
| Participants (local development): |
| |
| python scripts/score_submission.py \ |
| --predictions predictions.jsonl \ |
| --gold data/public_dev.jsonl |
| |
| Organizers (official run against a frozen config): |
| |
| python scripts/score_submission.py \ |
| --predictions submissions/team_a.jsonl \ |
| --gold /path/to/private_test_gold.jsonl \ |
| --config configs/official_evaluation.yaml \ |
| --output-dir results/team_a |
| |
| The judge needs OPENAI_API_KEY in the environment (see .env.example). |
| Use --judge deterministic for an API-free rule-based run - useful for smoke |
| tests, but NOT the official metric. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import logging |
| import sys |
| from pathlib import Path |
|
|
| import _bootstrap |
|
|
| from evaluation.config import EvaluationConfig |
| from evaluation.evaluator import IncompleteSubmissionError, evaluate |
| from task3_core.io import DataError, load_gold, load_predictions |
| from task3_core.validation import validate_predictions |
|
|
| ROOT = Path(__file__).resolve().parent.parent |
|
|
|
|
| def build_parser() -> argparse.ArgumentParser: |
| parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) |
| parser.add_argument("--predictions", required=True) |
| parser.add_argument("--gold", required=True) |
| parser.add_argument("--config", default=None, help="YAML config (default: built-in defaults)") |
| parser.add_argument("--output-dir", default=None, help="Write case_results/summary/metadata here") |
| parser.add_argument("--format", dest="fmt", default="native", choices=("native", "lm-eval")) |
| parser.add_argument("--submission-name", default=None, help="Label used in the report") |
|
|
| parser.add_argument("--judge", default=None, help="Override judge provider (openai/deterministic/mock)") |
| parser.add_argument("--model", default=None, help="Override judge model") |
| parser.add_argument("--prompt-version", default=None) |
| parser.add_argument("--workers", type=int, default=None) |
| parser.add_argument("--max-retries", type=int, default=None) |
| parser.add_argument( |
| "--missing-policy", default=None, |
| choices=("invalid_submission", "structural_error", "skip"), |
| help="Override missing_prediction_policy", |
| ) |
|
|
| cache = parser.add_mutually_exclusive_group() |
| cache.add_argument("--cache", dest="cache", action="store_true", default=None) |
| cache.add_argument("--no-cache", dest="cache", action="store_false") |
| parser.add_argument("--cache-path", default=None) |
| parser.add_argument( |
| "--resume", action="store_true", |
| help="Reuse cached judgements (equivalent to --cache; kept for discoverability)", |
| ) |
| parser.add_argument("--limit", type=int, default=None, help="Score only the first N cases (debugging)") |
| parser.add_argument("--skip-validation", action="store_true") |
| parser.add_argument("--quiet", action="store_true") |
| parser.add_argument("--verbose", action="store_true") |
| return parser |
|
|
|
|
| def main() -> int: |
| args = build_parser().parse_args() |
| logging.basicConfig( |
| level=logging.DEBUG if args.verbose else logging.INFO, |
| format="%(levelname)s %(name)s: %(message)s", |
| ) |
|
|
| config = EvaluationConfig.load(args.config) |
| if args.judge: |
| config.judge.provider = args.judge |
| if args.model: |
| config.judge.model = args.model |
| if args.prompt_version: |
| config.prompt_version = args.prompt_version |
| if args.workers is not None: |
| config.workers = args.workers |
| if args.max_retries is not None: |
| config.max_retries = args.max_retries |
| if args.missing_policy: |
| config.missing_prediction_policy = args.missing_policy |
| if args.cache is not None: |
| config.cache = args.cache |
| if args.resume: |
| config.cache = True |
| if args.cache_path: |
| config.cache_path = args.cache_path |
| config.__post_init__() |
|
|
| try: |
| gold = load_gold(args.gold) |
| predictions, load_errors = load_predictions(args.predictions, fmt=args.fmt) |
| except DataError as exc: |
| print(f"ERROR: {exc}", file=sys.stderr) |
| return 2 |
|
|
| case_ids = list(gold) |
| if args.limit: |
| |
| |
| |
| |
| case_ids = case_ids[: args.limit] |
| predictions = {k: v for k, v in predictions.items() if k in set(case_ids)} |
|
|
| if not args.skip_validation: |
| |
| |
| lenient = config.missing_prediction_policy != "invalid_submission" |
| report = validate_predictions( |
| predictions, |
| case_ids, |
| load_errors=load_errors, |
| predictions_path=str(args.predictions), |
| reference_path=str(args.gold), |
| allow_empty_values=lenient, |
| allow_missing=config.missing_prediction_policy == "skip", |
| ) |
| if not report.ok: |
| print(report.render(), file=sys.stderr) |
| print( |
| "\nSubmission did not pass validation. Fix the errors above, or re-run with " |
| "--missing-policy structural_error to score an incomplete submission " |
| "(never the official rule for the final leaderboard).", |
| file=sys.stderr, |
| ) |
| return 2 |
| for warning in report.warnings: |
| print(f"WARNING: {warning}", file=sys.stderr) |
| elif load_errors: |
| for error in load_errors: |
| print(f"WARNING: {error}", file=sys.stderr) |
|
|
| submission_name = args.submission_name or Path(args.predictions).stem |
|
|
| try: |
| run = evaluate( |
| gold, |
| predictions, |
| config, |
| case_ids=case_ids, |
| submission_name=submission_name, |
| predictions_path=args.predictions, |
| gold_path=args.gold, |
| cache_dir=ROOT, |
| show_progress=not args.quiet, |
| ) |
| except IncompleteSubmissionError as exc: |
| print(f"ERROR: {exc}", file=sys.stderr) |
| return 2 |
|
|
| print() |
| print(run.summary.render(submission_name)) |
|
|
| if run.summary.invalid_judge_outputs: |
| print( |
| f"\nWARNING: {run.summary.invalid_judge_outputs} case(s) received no valid judge " |
| "label after retries; see case_results.jsonl for the raw responses.", |
| file=sys.stderr, |
| ) |
|
|
| if args.output_dir: |
| out = run.write(args.output_dir, submission_name) |
| print(f"\nWrote results to {out}/") |
| for name in ("evaluation_report.txt", "evaluation_summary.json", |
| "evaluation_metadata.json", "case_results.jsonl"): |
| print(f" {name}") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
