# /// script # requires-python = ">=3.11" # dependencies = [] # /// # ─── How to run ─── # .venv/bin/python -m pytest -q tests/test_active_rectification_events.py """Deterministically adjudicate dated events against birth-time candidates.""" from __future__ import annotations from collections.abc import Sequence from typing import Any, Final, Literal, NotRequired, TypedDict, assert_never from uuid import NAMESPACE_URL, uuid5 try: from scripts.rectification_policy import ( MAX_CONFIRMATION_WIDTH_MINUTES, MIN_CONFIRMATION_DOMAINS, MIN_CONFIRMATION_EVENTS, MIN_CONFIRMATION_MARGIN_PERCENT, ) except ModuleNotFoundError: # pragma: no cover - direct script execution from rectification_policy import ( MAX_CONFIRMATION_WIDTH_MINUTES, MIN_CONFIRMATION_DOMAINS, MIN_CONFIRMATION_EVENTS, MIN_CONFIRMATION_MARGIN_PERCENT, ) EventPrecision = Literal["year", "month", "day"] EventDomain = Literal[ "education", "relocation", "relationship", "career", "finance", "health_pressure", ] Confidence = Literal["low", "medium", "high"] ALGORITHM_VERSION: Final = "birth-time-event-scoring-v2" PRECISION_WEIGHTS: Final[dict[EventPrecision, float]] = { "day": 1.0, "month": 0.8, "year": 0.5, } class CandidateEvidence(TypedDict): event_id: str domain: str candidate_time: str rule_ids: list[str] points: float class LifeEvent(TypedDict): id: str domain: EventDomain event_kind: NotRequired[str] date: str precision: EventPrecision summary: NotRequired[str] class RectificationEventRequest(TypedDict): birth_date: str start_time: str end_time: str lat: float lon: float tz: float events: list[LifeEvent] class CandidateScoreRow(TypedDict): time: str score: float evidence: list[CandidateEvidence] missing_layers: list[str] class WinningSegment(TypedDict): start_time: str end_time: str representative_time: str width_minutes: int class CandidateResult(TypedDict): result_id: str confidence: Confidence can_apply: bool winning_segment: WinningSegment | None event_count: int domain_count: int top_score: float second_score: float margin_percent: float reasons: list[str] evidence: list[CandidateEvidence] algorithm_version: str canonical_input_hash: str calculation_contract: dict[str, Any] stability_diagnostics: dict[str, Any] missing_layers: list[str] candidate_ranking_summary: NotRequired[list[dict[str, Any]]] def precision_weight(precision: EventPrecision) -> float: """Return the fixed evidence weight for a declared date precision.""" match precision: case "day" | "month" | "year": return PRECISION_WEIGHTS[precision] case unreachable: assert_never(unreachable) def _minute_value(value: str) -> int: hour, minute = value.split(":", maxsplit=1) return int(hour) * 60 + int(minute) def _is_next_minute(previous: str, current: str) -> bool: return (_minute_value(current) - _minute_value(previous)) % (24 * 60) == 1 def _top_segments(rows: Sequence[CandidateScoreRow], top_score: float) -> list[list[CandidateScoreRow]]: segments: list[list[CandidateScoreRow]] = [] for row in rows: if row["score"] != top_score: continue if segments and _is_next_minute(segments[-1][-1]["time"], row["time"]): segments[-1].append(row) else: segments.append([row]) return segments def _winning_segment(rows: Sequence[CandidateScoreRow]) -> WinningSegment: width = len(rows) representative = rows[(width - 1) // 2]["time"] return { "start_time": rows[0]["time"], "end_time": rows[-1]["time"], "representative_time": representative, "width_minutes": width, } def _clock_distance(left: str, right: str) -> int: distance = abs(_minute_value(left) - _minute_value(right)) return min(distance, 24 * 60 - distance) def _time_at_offset(value: str, offset: int) -> str: total = (_minute_value(value) + offset) % (24 * 60) return f"{total // 60:02d}:{total % 60:02d}" def build_stability_diagnostics( rows: Sequence[CandidateScoreRow], *, winning_segment: WinningSegment | None, ) -> dict[str, Any]: """Describe exact-minute neighbor separation without claiming calibrated accuracy.""" representative = winning_segment["representative_time"] if winning_segment else None by_time = {row["time"]: row for row in rows} representative_row = by_time.get(representative) if representative else None neighborhoods: list[dict[str, Any]] = [] for radius in (1, 2, 5): required_times = { _time_at_offset(representative, -radius), _time_at_offset(representative, radius), } if representative else set() neighbors = [ row for row in rows if representative is not None and 0 < _clock_distance(row["time"], representative) <= radius ] if representative_row is None or not required_times.issubset(by_time): neighborhoods.append({ "radius_minutes": radius, "status": "blocked", "lead_points": None, "compared_candidate_count": len(neighbors), "reason": "candidate_range_does_not_cover_both_sides_of_neighborhood", }) continue best_neighbor = max(row["score"] for row in neighbors) lead = round(representative_row["score"] - best_neighbor, 4) neighborhoods.append({ "radius_minutes": radius, "status": "pass" if winning_segment["width_minutes"] == 1 and lead > 0 else "fail", "lead_points": lead, "compared_candidate_count": len(neighbors), "reason": ( "unique_minute_leads_neighbor_candidates" if winning_segment["width_minutes"] == 1 and lead > 0 else "minute_not_uniquely_separated_from_neighbors" ), }) return { "scope": "candidate_neighbor_stability", "representative_time": representative, "neighborhoods": neighborhoods, "all_required_passed": all(item["status"] == "pass" for item in neighborhoods), "boundary": "Neighbor separation is a diagnostic only until thresholds are frozen before public holdout replay.", } def adjudicate_candidate_rows( rows: Sequence[CandidateScoreRow], *, event_count: int, domain_count: int, request_fingerprint: str, canonical_input_hash: str = "", calculation_contract: dict[str, Any] | None = None, leave_one_event_out: dict[str, Any] | None = None, ) -> CandidateResult: """Rank precomputed candidate rows and apply conservative confidence gates.""" if not rows: return { "result_id": str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{request_fingerprint}")), "confidence": "low", "can_apply": False, "winning_segment": None, "event_count": event_count, "domain_count": domain_count, "top_score": 0.0, "second_score": 0.0, "margin_percent": 0.0, "reasons": ["no_candidate_rows"], "evidence": [], "algorithm_version": ALGORITHM_VERSION, "canonical_input_hash": canonical_input_hash, "calculation_contract": calculation_contract or {}, "stability_diagnostics": { "neighbor_stability": build_stability_diagnostics([], winning_segment=None), "leave_one_event_out": leave_one_event_out or {"status": "not_evaluated", "runs": []}, }, "missing_layers": [], } ranked_scores = sorted({row["score"] for row in rows}, reverse=True) top_score = ranked_scores[0] second_score = ranked_scores[1] if len(ranked_scores) > 1 else top_score margin = round((top_score - second_score) / max(abs(top_score), 1.0) * 100, 2) segments = _top_segments(rows, top_score) missing_layers = sorted({layer for row in rows for layer in row["missing_layers"]}) reasons: list[str] = [] if len(segments) != 1: reasons.append("tied_leader") if event_count < 3: reasons.append("insufficient_events") if domain_count < 2: reasons.append("insufficient_domains") if missing_layers: reasons.append("missing_mandatory_layers") segment = _winning_segment(segments[0]) if len(segments) == 1 else None if segment and segment["width_minutes"] > 15: reasons.append("winning_interval_too_wide") if margin < 10: reasons.append("lead_margin_below_medium_threshold") if reasons: confidence: Confidence = "low" elif ( event_count >= MIN_CONFIRMATION_EVENTS and domain_count >= MIN_CONFIRMATION_DOMAINS and segment is not None and segment["width_minutes"] <= MAX_CONFIRMATION_WIDTH_MINUTES and margin >= MIN_CONFIRMATION_MARGIN_PERCENT ): confidence = "high" else: confidence = "medium" top_rows = segments[0] if len(segments) == 1 else [] representative_row = top_rows[(len(top_rows) - 1) // 2] if top_rows else None evidence = list(representative_row["evidence"]) if representative_row else [] neighbor_stability = build_stability_diagnostics(rows, winning_segment=segment) if not neighbor_stability["all_required_passed"]: reasons.append("neighbor_stability_not_passed") if (leave_one_event_out or {}).get("status") != "pass": reasons.append("leave_one_event_out_not_passed") can_apply = ( confidence == "high" and segment is not None and neighbor_stability["all_required_passed"] and (leave_one_event_out or {}).get("status") == "pass" and not missing_layers and not any(reason in reasons for reason in ( "tied_leader", "insufficient_events", "insufficient_domains", "winning_interval_too_wide", "lead_margin_below_medium_threshold", )) ) ranking_summary: list[dict[str, Any]] = [] for score in ranked_scores[:3]: score_rows = sorted( (row for row in rows if row["score"] == score), key=lambda row: _minute_value(row["time"]), ) if not score_rows: continue representative = score_rows[(len(score_rows) - 1) // 2] ranking_summary.append({ "rank": len(ranking_summary) + 1, "time": representative["time"], "score": score, "tied_minute_count": len(score_rows), }) return { "result_id": str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{request_fingerprint}")), "confidence": confidence, "can_apply": can_apply, "winning_segment": segment, "event_count": event_count, "domain_count": domain_count, "top_score": top_score, "second_score": second_score, "margin_percent": margin, "reasons": reasons, "evidence": evidence, "algorithm_version": ALGORITHM_VERSION, "canonical_input_hash": canonical_input_hash, "calculation_contract": calculation_contract or {}, "stability_diagnostics": { "neighbor_stability": neighbor_stability, "leave_one_event_out": leave_one_event_out or {"status": "not_evaluated", "runs": []}, }, "missing_layers": missing_layers, "candidate_ranking_summary": ranking_summary, } def score_life_events(request: RectificationEventRequest) -> CandidateResult: """Compute actual candidate rows, then apply the versioned confidence gates.""" from scripts.active_rectification_event_engine import compute_event_candidate_result return compute_event_candidate_result(request)