"""Exact Task 1 case scoring and hierarchical aggregation.""" from __future__ import annotations import json from collections import Counter from dataclasses import dataclass from decimal import Decimal, ROUND_HALF_EVEN, ROUND_HALF_UP, localcontext from fractions import Fraction from typing import Any, Iterable, Mapping, Sequence from .contracts import ( REPORT_SCHEMA_VERSION, ContractError, validate_gold_row, validate_manifest_row, validate_question, validate_submission, ) SCORER_VERSION = "finreason.task1.scorer/2.0.0" RANKING_QUANTUM = Decimal("0.000001") SEEN_DIFFICULTY_WEIGHTS = { "basic": Fraction(2, 5), "intermediate": Fraction(2, 5), "advanced": Fraction(1, 5), } HIDDEN_DIFFICULTY_WEIGHTS = { "intermediate": Fraction(1, 2), "advanced": Fraction(1, 2), } OFFICIAL_SPLIT_COUNTS = { "test_private_seen": 580, "test_private_hidden": 348, } TASK1_V4_DATASET_VERSION = "task1-v4" TASK1_V4_PRIVATE_SCORE_COUNTS = { "cases": 928, "checkpoint_cases": 864, "checkpoint_templates": 374, "total_slots": 1492, } @dataclass(frozen=True) class CaseScore: case_id: str fac: int matched_slots: int total_slots: int trace_accuracy: Fraction | None @dataclass(frozen=True) class AggregateScore: fac: Fraction checkpoint: Fraction | None case_count: int checkpoint_case_count: int checkpoint_template_count: int @dataclass(frozen=True) class ScoreReport: schema_version: str scorer_version: str dataset_version: str combined_fac: Fraction | None seen_fac: Fraction | None hidden_fac: Fraction | None combined_checkpoint: Fraction | None seen_checkpoint: Fraction | None hidden_checkpoint: Fraction | None micro_fac: Fraction micro_checkpoint: Fraction | None ranking_tuple: tuple[str, str, str] | None case_count: int checkpoint_case_count: int checkpoint_template_count: int matched_final_answers: int matched_slots: int total_slots: int def as_dict(self) -> dict[str, Any]: def optional(value: Fraction | None) -> str | None: return None if value is None else render_fraction(value) return { "schema_version": self.schema_version, "scorer_version": self.scorer_version, "dataset_version": self.dataset_version, "scores": { "combined_fac": optional(self.combined_fac), "seen_fac": optional(self.seen_fac), "hidden_fac": optional(self.hidden_fac), "combined_checkpoint": optional(self.combined_checkpoint), "seen_checkpoint": optional(self.seen_checkpoint), "hidden_checkpoint": optional(self.hidden_checkpoint), }, "diagnostics": { "micro_fac": optional(self.micro_fac), "micro_checkpoint": optional(self.micro_checkpoint), }, "ranking_tuple": list(self.ranking_tuple) if self.ranking_tuple else None, "counts": { "cases": self.case_count, "checkpoint_cases": self.checkpoint_case_count, "checkpoint_templates": self.checkpoint_template_count, "matched_final_answers": self.matched_final_answers, "matched_slots": self.matched_slots, "total_slots": self.total_slots, }, } def render_fraction(value: Fraction, *, places: int = 16) -> str: """Serialize an exact score without emitting binary floating point.""" with localcontext() as context: context.prec = max(64, places + 32) decimal_value = Decimal(value.numerator) / Decimal(value.denominator) quantum = Decimal(1).scaleb(-places) rendered = format(decimal_value.quantize(quantum, rounding=ROUND_HALF_EVEN), "f") rendered = rendered.rstrip("0").rstrip(".") return rendered if rendered and rendered != "-0" else "0" def ranking_value(value: Fraction) -> str: with localcontext() as context: context.prec = 64 decimal_value = Decimal(value.numerator) / Decimal(value.denominator) return format(decimal_value.quantize(RANKING_QUANTUM, rounding=ROUND_HALF_EVEN), "f") def _canonical_decimal(text: str, spec: Mapping[str, Any]) -> Decimal: value = Decimal(text) rounding = spec["rounding"] if rounding["mode"] == "half_up": quantum = Decimal(1).scaleb(-rounding["decimal_places"]) with localcontext() as context: context.prec = 256 value = value.quantize(quantum, rounding=ROUND_HALF_UP) if value.is_zero(): value = abs(value) return value def _canonical_exact_number(text: str) -> Fraction: if "/" in text: numerator, denominator = text.split("/", 1) return Fraction(int(numerator), int(denominator)) return Fraction(Decimal(text)) def value_matches( submitted_value: str, gold_value: str, spec: Mapping[str, Any], ) -> bool: if spec["type"] == "enum": return submitted_value == gold_value if spec["type"] == "decimal": if spec["rounding"]["mode"] == "exact": return _canonical_exact_number(submitted_value) == _canonical_exact_number( gold_value ) return _canonical_decimal(submitted_value, spec) == _canonical_decimal( gold_value, spec ) raise ContractError(f"unsupported result type {spec.get('type')!r}") def _gold_values(gold: Mapping[str, Any], question: Mapping[str, Any]) -> tuple[str, dict[str, str]]: validate_gold_row(gold, question) final_value = gold["final_value"] slot_values = gold["slot_values"] expected_slot_ids = [slot["slot_id"] for slot in question["trace_spec"]["slots"]] return final_value, {slot_id: slot_values[slot_id] for slot_id in expected_slot_ids} def score_case( question: Mapping[str, Any], gold: Mapping[str, Any], prediction: Mapping[str, Any], ) -> CaseScore: validate_question(question) validation = validate_submission( [prediction], {question["case_id"]: question}, require_complete=True ) if not validation.valid: messages = "; ".join(issue.message for issue in validation.issues) raise ContractError(f"prediction is invalid: {messages}") final_gold, slot_gold = _gold_values(gold, question) final_answer = prediction["final_answer"] fac = int( final_answer is not None and value_matches( final_answer["value"], final_gold, question["answer_spec"] ) ) step_by_id = {step["slot_id"]: step["value"] for step in prediction["steps"]} matched_slots = 0 slots = question["trace_spec"]["slots"] for slot in slots: slot_id = slot["slot_id"] if slot_id in step_by_id and value_matches( step_by_id[slot_id], slot_gold[slot_id], slot["result_spec"] ): matched_slots += 1 total_slots = len(slots) trace_accuracy = ( None if total_slots == 0 else Fraction(matched_slots, total_slots) ) return CaseScore( case_id=question["case_id"], fac=fac, matched_slots=matched_slots, total_slots=total_slots, trace_accuracy=trace_accuracy, ) def _mean(values: Sequence[Fraction], *, location: str) -> Fraction: if not values: raise ContractError(f"{location}: cannot average an empty cell") return sum(values, Fraction(0)) / len(values) def _validate_manifest_template_identity( manifest_rows: Sequence[Mapping[str, Any]], ) -> None: """Require one global taxonomy identity for every template identifier.""" identities: dict[str, tuple[str, str, str, str]] = {} for index, row in enumerate(manifest_rows, start=1): template_id = row.get("template_id") identity_values = ( row.get("structure_stratum"), row.get("domain_id"), row.get("topic_id"), row.get("difficulty"), ) if not isinstance(template_id, str) or not template_id: raise ContractError(f"manifest row {index} has invalid template_id") if any(not isinstance(value, str) or not value for value in identity_values): raise ContractError( f"manifest template {template_id} has invalid taxonomy identity" ) identity = ( str(identity_values[0]), str(identity_values[1]), str(identity_values[2]), str(identity_values[3]), ) previous = identities.setdefault(template_id, identity) if previous != identity: raise ContractError( f"template {template_id} changes stratum or taxonomy identity" ) def _canonical_scoring_spec(question: Mapping[str, Any]) -> str: return json.dumps( { "answer_spec": question["answer_spec"], "trace_spec": question["trace_spec"], }, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) def _validate_template_scoring_specs( questions: Mapping[str, Mapping[str, Any]], manifest_by_id: Mapping[str, Mapping[str, Any]], ) -> None: """Reject answer or checkpoint contract drift within a template.""" specifications: dict[str, str] = {} for case_id in sorted(questions): template_id = manifest_by_id[case_id]["template_id"] specification = _canonical_scoring_spec(questions[case_id]) previous = specifications.setdefault(template_id, specification) if previous != specification: raise ContractError( f"template {template_id} changes answer_spec or trace_spec" ) def _aggregate_component( case_scores: Mapping[str, CaseScore], manifest_rows: Sequence[Mapping[str, Any]], *, stratum: str, component: str, ) -> Fraction | None: weights = ( SEEN_DIFFICULTY_WEIGHTS if stratum == "seen" else HIDDEN_DIFFICULTY_WEIGHTS ) by_template: dict[tuple[str, str, str, str], list[Fraction]] = {} for row in manifest_rows: if row.get("structure_stratum") != stratum: continue case_id = row["case_id"] if case_id not in case_scores: raise ContractError(f"manifest case {case_id} lacks a score") key = ( row["domain_id"], row["topic_id"], row["difficulty"], row["template_id"], ) if component == "fac": value: Fraction | None = Fraction(case_scores[case_id].fac) elif component == "checkpoint": value = case_scores[case_id].trace_accuracy else: raise ContractError(f"unsupported aggregate component {component!r}") if value is None: continue by_template.setdefault(key, []).append(value) if not by_template: if component == "checkpoint": return None raise ContractError(f"stratum {stratum} is empty") template_presence: dict[tuple[str, str, str, str], set[bool]] = {} for row in manifest_rows: if row.get("structure_stratum") != stratum: continue key = ( row["domain_id"], row["topic_id"], row["difficulty"], row["template_id"], ) template_presence.setdefault(key, set()).add( case_scores[row["case_id"]].trace_accuracy is not None ) if component == "checkpoint": inconsistent = sorted( key for key, states in template_presence.items() if len(states) != 1 ) if inconsistent: raise ContractError( "checkpoint availability changes within template " f"{inconsistent[0]}" ) by_cell: dict[tuple[str, str, str], list[Fraction]] = {} for (domain, topic, difficulty, _template), values in by_template.items(): by_cell.setdefault((domain, topic, difficulty), []).append( _mean(values, location="template") ) topic_keys = sorted( { (row["domain_id"], row["topic_id"]) for row in manifest_rows if row.get("structure_stratum") == stratum } ) topic_values: dict[tuple[str, str], Fraction] = {} for domain, topic in topic_keys: available = { difficulty for cell_domain, cell_topic, difficulty in by_cell if (cell_domain, cell_topic) == (domain, topic) } if component == "fac" and available != set(weights): raise ContractError( f"{stratum} topic {domain}/{topic} has difficulty cells " f"{sorted(available)}, expected {sorted(weights)}" ) if component == "checkpoint" and not available: raise ContractError( f"{stratum} topic {domain}/{topic} has no eligible checkpoints" ) applicable_weight = sum( (weights[difficulty] for difficulty in available), Fraction(0) ) topic_values[(domain, topic)] = sum( weights[difficulty] * _mean( by_cell[(domain, topic, difficulty)], location=f"{domain}/{topic}/{difficulty}", ) for difficulty in available ) / applicable_weight by_domain: dict[str, list[Fraction]] = {} for (domain, _topic), value in topic_values.items(): by_domain.setdefault(domain, []).append(value) if len(by_domain) != 12: raise ContractError( f"stratum {stratum} must cover exactly 12 domains, found {len(by_domain)}" ) return _mean( [_mean(values, location=f"domain {domain}") for domain, values in by_domain.items()], location=f"stratum {stratum}", ) def aggregate_scores( case_scores: Mapping[str, CaseScore], manifest_rows: Iterable[Mapping[str, Any]], ) -> tuple[AggregateScore | None, AggregateScore | None]: rows = list(manifest_rows) if len({row.get("case_id") for row in rows}) != len(rows): raise ContractError("manifest contains duplicate case_id membership") expected_ids = {row.get("case_id") for row in rows} if expected_ids != set(case_scores): raise ContractError("manifest and case-score identifiers differ") _validate_manifest_template_identity(rows) aggregates: dict[str, AggregateScore | None] = {"seen": None, "hidden": None} for stratum in ("seen", "hidden"): count = sum(row.get("structure_stratum") == stratum for row in rows) if not count: continue aggregates[stratum] = AggregateScore( fac=_aggregate_component(case_scores, rows, stratum=stratum, component="fac"), checkpoint=_aggregate_component( case_scores, rows, stratum=stratum, component="checkpoint" ), case_count=count, checkpoint_case_count=sum( row.get("structure_stratum") == stratum and case_scores[row["case_id"]].trace_accuracy is not None for row in rows ), checkpoint_template_count=len( { ( row["domain_id"], row["topic_id"], row["difficulty"], row["template_id"], ) for row in rows if row.get("structure_stratum") == stratum and case_scores[row["case_id"]].trace_accuracy is not None } ), ) return aggregates["seen"], aggregates["hidden"] def validate_official_manifest(manifest_rows: Iterable[Mapping[str, Any]]) -> None: """Reject candidate-pool rows before official private scoring.""" rows = list(manifest_rows) splits: list[str] = [] for index, row in enumerate(rows, start=1): if not isinstance(row, Mapping): raise ContractError(f"official manifest row {index} must be an object") split = row.get("split") if not isinstance(split, str): raise ContractError(f"official manifest row {index} has invalid split") splits.append(split) counts = Counter(splits) if counts != Counter(OFFICIAL_SPLIT_COUNTS): raise ContractError( f"official manifest split counts are {dict(counts)}, " f"expected {OFFICIAL_SPLIT_COUNTS}" ) if len(rows) != sum(OFFICIAL_SPLIT_COUNTS.values()): raise ContractError("official manifest must contain exactly 928 rows") def validate_task1_v4_private_score_report(report: ScoreReport) -> None: """Freeze V4 counts and relational score invariants before publication.""" if report.dataset_version != TASK1_V4_DATASET_VERSION: raise ContractError("private score receipt requires dataset_version task1-v4") if report.schema_version != REPORT_SCHEMA_VERSION: raise ContractError("private score receipt has the wrong report schema version") if report.scorer_version != SCORER_VERSION: raise ContractError("private score receipt has the wrong scorer version") actual = { "cases": report.case_count, "checkpoint_cases": report.checkpoint_case_count, "checkpoint_templates": report.checkpoint_template_count, "total_slots": report.total_slots, } if actual != TASK1_V4_PRIVATE_SCORE_COUNTS: raise ContractError( f"task1-v4 private score counts are {actual}, " f"expected {TASK1_V4_PRIVATE_SCORE_COUNTS}" ) score_values = { "combined_fac": report.combined_fac, "seen_fac": report.seen_fac, "hidden_fac": report.hidden_fac, "combined_checkpoint": report.combined_checkpoint, "seen_checkpoint": report.seen_checkpoint, "hidden_checkpoint": report.hidden_checkpoint, "micro_fac": report.micro_fac, "micro_checkpoint": report.micro_checkpoint, } for name, value in score_values.items(): if value is None: raise ContractError(f"private score receipt requires {name}") if value < 0 or value > 1: raise ContractError(f"private score receipt {name} is outside [0,1]") assert report.seen_fac is not None assert report.hidden_fac is not None assert report.combined_fac is not None assert report.seen_checkpoint is not None assert report.hidden_checkpoint is not None assert report.combined_checkpoint is not None assert report.micro_checkpoint is not None if report.combined_fac != (report.seen_fac + report.hidden_fac) / 2: raise ContractError("private combined FAC is not the exact 50/50 aggregate") if report.combined_checkpoint != ( report.seen_checkpoint + report.hidden_checkpoint ) / 2: raise ContractError( "private combined checkpoint is not the exact 50/50 aggregate" ) expected_ranking = ( ranking_value(report.combined_fac), ranking_value(report.hidden_fac), ranking_value(report.combined_checkpoint), ) if report.ranking_tuple != expected_ranking: raise ContractError("private ranking tuple is inconsistent with score values") if not 0 <= report.matched_final_answers <= report.case_count: raise ContractError("private matched final-answer count is invalid") if not 0 <= report.matched_slots <= report.total_slots: raise ContractError("private matched checkpoint count is invalid") if report.micro_fac != Fraction( report.matched_final_answers, report.case_count ): raise ContractError("private micro FAC is inconsistent with counts") if report.micro_checkpoint != Fraction(report.matched_slots, report.total_slots): raise ContractError("private micro checkpoint is inconsistent with counts") def score_submission( questions: Mapping[str, Mapping[str, Any]], gold_rows: Mapping[str, Mapping[str, Any]], predictions: Iterable[Mapping[str, Any]], manifest_rows: Iterable[Mapping[str, Any]], ) -> ScoreReport: if not questions: raise ContractError("question bundle is empty") manifest = list(manifest_rows) validation = validate_submission(predictions, questions, require_complete=True) if not validation.valid: messages = "; ".join(issue.message for issue in validation.issues) raise ContractError(f"submission rejected before scoring: {messages}") if set(gold_rows) != set(questions): raise ContractError("gold and question identifiers differ") if len(manifest) != len(questions): raise ContractError("manifest and question row counts differ") manifest_by_id: dict[str, Mapping[str, Any]] = {} for row in manifest: case_id = row.get("case_id") if not isinstance(case_id, str) or case_id not in questions: raise ContractError("manifest contains an unknown or invalid case_id") if case_id in manifest_by_id: raise ContractError("manifest contains duplicate case_id membership") validate_manifest_row(row, questions[case_id]) manifest_by_id[case_id] = row if set(manifest_by_id) != set(questions): raise ContractError("manifest and question identifiers differ") _validate_manifest_template_identity(manifest) _validate_template_scoring_specs(questions, manifest_by_id) case_scores: dict[str, CaseScore] = {} for prediction in validation.records: case_id = prediction["case_id"] case_scores[case_id] = score_case( questions[case_id], gold_rows[case_id], prediction ) seen, hidden = aggregate_scores(case_scores, manifest) combined_fac: Fraction | None = None combined_checkpoint: Fraction | None = None ranking_tuple: tuple[str, str, str] | None = None if seen is not None and hidden is not None: combined_fac = (seen.fac + hidden.fac) / 2 if seen.checkpoint is not None and hidden.checkpoint is not None: combined_checkpoint = (seen.checkpoint + hidden.checkpoint) / 2 ranking_tuple = ( ranking_value(combined_fac), ranking_value(hidden.fac), ranking_value(combined_checkpoint), ) dataset_versions = {question["dataset_version"] for question in questions.values()} if len(dataset_versions) != 1: raise ContractError("question bundle contains multiple dataset versions") return ScoreReport( schema_version=REPORT_SCHEMA_VERSION, scorer_version=SCORER_VERSION, dataset_version=next(iter(dataset_versions)), combined_fac=combined_fac, seen_fac=None if seen is None else seen.fac, hidden_fac=None if hidden is None else hidden.fac, combined_checkpoint=combined_checkpoint, seen_checkpoint=None if seen is None else seen.checkpoint, hidden_checkpoint=None if hidden is None else hidden.checkpoint, micro_fac=Fraction( sum(score.fac for score in case_scores.values()), len(case_scores) ), micro_checkpoint=( None if not sum(score.total_slots for score in case_scores.values()) else Fraction( sum(score.matched_slots for score in case_scores.values()), sum(score.total_slots for score in case_scores.values()), ) ), ranking_tuple=ranking_tuple, case_count=len(case_scores), checkpoint_case_count=sum( score.trace_accuracy is not None for score in case_scores.values() ), checkpoint_template_count=len( { manifest_by_id[case_id]["template_id"] for case_id, score in case_scores.items() if score.trace_accuracy is not None } ), matched_final_answers=sum(score.fac for score in case_scores.values()), matched_slots=sum(score.matched_slots for score in case_scores.values()), total_slots=sum(score.total_slots for score in case_scores.values()), )