Jesse_Chen
a0ce55f066
Showing a choice card is no longer treated as completion. Distinguish probes require real candidate groups, holdout stays out of scoring, and ordinary sessions can finish with a credible range instead of an exact-minute gate. Co-authored-by: Cursor <cursoragent@cursor.com>
115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
"""Per-case holdout reservation for birth-time rectification.
|
|
|
|
Reserve at least one dated event as soon as collection has two dated
|
|
events. Holdout evidence must not enter screening totals, candidate
|
|
scoring, or discriminator probe selection. The public AA sealed holdout
|
|
contract is a different file (`sealed_holdout.py`).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from typing import Any
|
|
|
|
from scripts.rectification.candidate_contrast import event_year
|
|
|
|
MIN_EVENTS_TO_RESERVE_HOLDOUT = 2
|
|
_MONTH_OR_BETTER = frozenset({"day", "month"})
|
|
|
|
|
|
def _event_id(event: dict[str, Any]) -> str:
|
|
return str(event.get("id") or "").strip()
|
|
|
|
|
|
def _domain(event: dict[str, Any]) -> str:
|
|
return str(event.get("domain") or "").strip()
|
|
|
|
|
|
def _precision(event: dict[str, Any]) -> str:
|
|
return str(event.get("precision") or event.get("date_precision") or "year").strip() or "year"
|
|
|
|
|
|
def dated_events(events: Sequence[dict[str, Any]] | None) -> list[dict[str, Any]]:
|
|
dated: list[dict[str, Any]] = []
|
|
for event in events or []:
|
|
if not isinstance(event, dict):
|
|
continue
|
|
if not _event_id(event) or not _domain(event):
|
|
continue
|
|
if event_year(event) is None:
|
|
continue
|
|
if _precision(event) == "unknown":
|
|
continue
|
|
dated.append(event)
|
|
return dated
|
|
|
|
|
|
def pick_holdout_id(events: Sequence[dict[str, Any]] | None) -> str | None:
|
|
dated = dated_events(events)
|
|
if len(dated) < MIN_EVENTS_TO_RESERVE_HOLDOUT:
|
|
return None
|
|
by_domain: dict[str, list[dict[str, Any]]] = {}
|
|
for event in dated:
|
|
by_domain.setdefault(_domain(event), []).append(event)
|
|
month_or_better = [event for event in dated if _precision(event) in _MONTH_OR_BETTER]
|
|
singleton = next((item for item in by_domain.items() if len(item[1]) == 1), None)
|
|
if singleton and len(dated) - 1 >= 3:
|
|
preferred = next((event for event in month_or_better if _domain(event) == singleton[0]), None)
|
|
chosen = preferred or singleton[1][0]
|
|
return _event_id(chosen) or None
|
|
ranked = [*month_or_better, *dated]
|
|
return _event_id(ranked[-1]) if ranked else None
|
|
|
|
|
|
def sticky_holdout_id(
|
|
events: Sequence[dict[str, Any]] | None,
|
|
previous_holdout_id: str | None = None,
|
|
) -> str | None:
|
|
ids = {_event_id(event) for event in dated_events(events)}
|
|
if previous_holdout_id and previous_holdout_id in ids:
|
|
return previous_holdout_id
|
|
explicit = [
|
|
_event_id(event)
|
|
for event in (events or [])
|
|
if isinstance(event, dict) and str(event.get("usage") or "") == "holdout"
|
|
]
|
|
if explicit and explicit[0] in ids:
|
|
return explicit[0]
|
|
return pick_holdout_id(events)
|
|
|
|
|
|
def holdout_event_ids(
|
|
events: Sequence[dict[str, Any]] | None,
|
|
previous_holdout_id: str | None = None,
|
|
) -> frozenset[str]:
|
|
holdout_id = sticky_holdout_id(events, previous_holdout_id)
|
|
return frozenset({holdout_id} if holdout_id else ())
|
|
|
|
|
|
def holdout_domain_years(
|
|
events: Sequence[dict[str, Any]] | None,
|
|
previous_holdout_id: str | None = None,
|
|
) -> frozenset[str]:
|
|
holdout = holdout_event_ids(events, previous_holdout_id)
|
|
keys: set[str] = set()
|
|
for event in events or []:
|
|
if not isinstance(event, dict) or _event_id(event) not in holdout:
|
|
continue
|
|
year = event_year(event)
|
|
domain = _domain(event)
|
|
if year is None or not domain:
|
|
continue
|
|
keys.add(f"{domain}:{year}")
|
|
return frozenset(keys)
|
|
|
|
|
|
def reserved_holdout_events(
|
|
events: Sequence[dict[str, Any]] | None,
|
|
previous_holdout_id: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
holdout = holdout_event_ids(events, previous_holdout_id)
|
|
return [
|
|
event for event in (events or [])
|
|
if isinstance(event, dict) and _event_id(event) in holdout
|
|
]
|