"""Scaffold (R-0) for TASK-rectification-scoring-research-20260929. Offline only. Nothing here changes a production default or a frozen scoring file. The module records, for every (case, candidate minute, event, sample date), which scoring rules fire — the eleven production rules from `active_rectification_event_engine._score_event` plus the auxiliary transit / Ashtakavarga / Shadbala rules — and a set of *extra* observations that production computes but does not score (KP cusp sub-lords, Pranapada / Hora / Ghati / Bhava lagna, D60). Likelihood-ratio weights are estimated per feature with leave-one-**case**-out folds, and a weighted row provider feeds the same contribution-matrix / probe / six-question replay machinery that the 09-14 / 09-26 / 09-29 studies used. Recording provider vs production: the provider returns exactly the production evidence (same rule ids, same points), so `build_event_contribution_matrix` yields the same matrix as the production path. `reconcile_rows` asserts that per case before any experiment (task hard line 4). """ from __future__ import annotations import math import statistics import sys from collections import defaultdict from dataclasses import dataclass, field from datetime import date, datetime from pathlib import Path from random import Random from typing import Any, Callable, Sequence ROOT = Path(__file__).resolve().parents[2] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from scripts.active_rectification_event_engine import ( # noqa: E402 DOMAIN_CONFIG, OBSERVATION_ONLY_LAYERS, _active_narayana, _active_vimshottari, _ashtakavarga_auxiliary, _controlled_transit_rules, _event_datetime, _relative_house, _shadbala_verified_components_auxiliary, _varga_chart, _varga_house, ) import scripts.rectification.event_probes as event_probes # noqa: E402 from scripts.rectification.candidate_contrast import cluster_contexts_by_signature # noqa: E402 from scripts.rectification.case_holdout import holdout_event_ids # noqa: E402 from scripts.rectification.contracts import is_scoreable_event # noqa: E402 from scripts.rectification.refinement_packet import window_scan # noqa: E402 from scripts.rectification.scoring_service import ( # noqa: E402 build_event_contribution_matrix, score_from_matrix, ) from scripts.research.cluster_width_lib import ( # noqa: E402 SEPARATION_LEAD, delivery_from_public, merge_adjacent_traced, metrics_bundle, public_from_clusters, raw_signature_clusters, shannon_entropy, still_valid_public, ) from scripts.research.precision_gate_lib import ( # noqa: E402 VargaPolicy, score_event_with_policy, ) from scripts.research.probe_supply_after_six import ( # noqa: E402 ASK_COUNT, SCORE_DELTA, apply_answer, inherit_direction, optimal_answer, outcome_groups, ) import varga # noqa: E402 TODAY = date(2026, 9, 16) # same probe "today" as the 09-16 / 09-29 replays LAPLACE_ALPHA = 1.0 EPS = 1e-9 EXTRA_PREFIXES = ("kp_", "pranapada_", "hora_lagna_", "ghati_lagna_", "bhava_lagna_") D60_SUFFIX = "_domain_varga_d60" IGNORED_RULE_PREFIXES = ("event_kind:", "event_kind_profile:", "no_domain_activation") CONSTANT_RULES = frozenset({"occupation_auxiliary_not_primary", "appearance_auxiliary_not_primary"}) VARIANTS = ("A1", "A2", "A3") VARIANT_SPEC = { # (uses extra features, allows negative / absence weights) "A1": (False, False), "A2": (True, False), "A3": (True, True), } # --------------------------------------------------------------------------- # small helpers # --------------------------------------------------------------------------- def hhmm(value: object) -> str | None: text = str(value or "")[:5] return text if len(text) == 5 and text[2] == ":" else None def clock(value: str) -> int: return int(value[:2]) * 60 + int(value[3:5]) def is_extra_feature(name: str) -> bool: return name.startswith(EXTRA_PREFIXES) or name.endswith(D60_SUFFIX) def is_scoring_rule(name: str) -> bool: if name.startswith(IGNORED_RULE_PREFIXES) or name in CONSTANT_RULES: return False return not is_extra_feature(name) def median(values: Sequence[float]) -> float | None: items = [float(item) for item in values if item is not None] return statistics.median(items) if items else None # --------------------------------------------------------------------------- # R-0 · feature recording provider (production-identical evidence + extras) # --------------------------------------------------------------------------- @dataclass class FeatureStore: """(event_id, sample_date) -> HH:MM -> {rules, extras}.""" rows: dict[tuple[str, str], dict[str, dict[str, Any]]] = field(default_factory=dict) def put(self, event_id: str, sample: str, time: str, rules: Sequence[str], extras: Sequence[str]) -> None: self.rows.setdefault((event_id, sample), {})[time] = { "rules": sorted(set(rules)), "extras": sorted(set(extras)), } def samples_for(self, event_id: str) -> list[str]: return sorted(sample for (item, sample) in self.rows if item == event_id) def d60_charts(contexts: Sequence[dict[str, Any]]) -> dict[str, dict[str, Any] | None]: out: dict[str, dict[str, Any] | None] = {} for context in contexts: stamp = context["candidate_at"].strftime("%H:%M") try: computed = varga.calc_all_vargas( context["planet_longitudes"], float(context["ascendant_longitude"]), divisions=[60], ) out[stamp] = _varga_chart(computed, "D60") except (KeyError, TypeError, ValueError): out[stamp] = None return out def extra_features( context: dict[str, Any], target_houses: tuple[int, ...], vimshottari: tuple[str, str, str], d60: dict[str, Any] | None, ) -> list[str]: feature = context.get("feature") if isinstance(context.get("feature"), dict) else {} ascendant_index = int(context["ascendant_index"]) kp_houses = ((feature.get("kp_cusps") or {}).get("houses") or {}) target_sub = {str((kp_houses.get(str(h)) or {}).get("sub_lord") or "") for h in target_houses} target_star = {str((kp_houses.get(str(h)) or {}).get("nakshatra_lord") or "") for h in target_houses} asc_sub = str((kp_houses.get("1") or {}).get("sub_lord") or "") out: list[str] = [] for lord, label in zip(vimshottari, ("md", "ad", "pd")): if lord in target_sub: out.append(f"kp_target_cusp_sublord_is_vim_{label}") if lord in target_star: out.append(f"kp_target_cusp_starlord_is_vim_{label}") if asc_sub and lord == asc_sub: out.append(f"kp_asc_sublord_is_vim_{label}") if isinstance(d60, dict) and _varga_house(d60, lord) in target_houses: out.append(f"vim_{label}{D60_SUFFIX}") for key, name in ( ("pranapada_sign_index", "pranapada"), ("hora_sign_index", "hora_lagna"), ("ghati_sign_index", "ghati_lagna"), ("bhava_sign_index", "bhava_lagna"), ): index = feature.get(key) if isinstance(index, int) and _relative_house(index, ascendant_index) in target_houses: out.append(f"{name}_in_target_house") return out def recorded_row( request: dict[str, Any], context: dict[str, Any], *, store: FeatureStore, transit_cache: dict[tuple[Any, ...], dict[str, Any]], d60_by_time: dict[str, dict[str, Any] | None], ) -> dict[str, Any]: """Production `_candidate_row` for one legacy (single event, one date) request, plus extras.""" candidate_at = context["candidate_at"] chart = context["chart"] planet_longitudes = context["planet_longitudes"] ascendant_index = context["ascendant_index"] arudha_padas = context["arudha_padas"] varga_charts = context["varga_charts"] moon_longitude = planet_longitudes["Moon"] stamp = candidate_at.strftime("%H:%M") policy = VargaPolicy(name="V0", window_minutes=0.0) evidence: list[dict[str, Any]] = [] missing: list[str] = [] for event in request["events"]: event_at = _event_datetime(event) prefixes, target_houses = DOMAIN_CONFIG[event["domain"]] selected = {prefix: varga_charts.get(prefix) for prefix in prefixes} if any(item is None for item in selected.values()): missing.extend(prefixes) continue try: vimshottari = _active_vimshottari( candidate_at.date().isoformat(), moon_longitude, event_at, context.get("vimshottari_timeline"), ) except (KeyError, TypeError, ValueError): missing.append("Vimshottari_MD_AD_PD") continue try: narayana = _active_narayana( ascendant_index, planet_longitudes, candidate_at, event_at, context.get("narayana_periods"), ) except (KeyError, TypeError, ValueError): missing.append("Narayana_MD_AD") continue if narayana[0] is None or narayana[1] is None: missing.append("Narayana_MD_AD") continue row = score_event_with_policy( candidate_time=stamp, event=event, natal_chart=chart, varga_by_prefix={k: v for k, v in selected.items() if isinstance(v, dict)}, vimshottari=vimshottari, narayana=narayana, arudha_padas=arudha_padas, policy=policy, ) weight = 1.0 # legacy requests carry precision "day"; the matrix applies the real weight transit_rules = _controlled_transit_rules( request, event, ascendant_index, target_houses, transit_cache, ) if transit_rules: row["rule_ids"].extend(transit_rules) row["points"] = round(row["points"] + 0.25 * len(transit_rules) * weight, 4) av_rules, av_points = _ashtakavarga_auxiliary( chart, ascendant_index, target_houses, context.get("ashtakavarga_result"), ) if av_rules: row["rule_ids"].extend(av_rules) row["points"] = round(row["points"] + av_points * weight, 4) sb_rules, sb_points = _shadbala_verified_components_auxiliary( chart, candidate_at.hour + candidate_at.minute / 60, vimshottari, context.get("shadbala_result"), ) if sb_rules: row["rule_ids"].extend(sb_rules) row["points"] = round(row["points"] + sb_points * weight, 4) evidence.append(row) store.put( str(event["id"]), str(event["date"]), stamp, row["rule_ids"], extra_features(context, target_houses, vimshottari, d60_by_time.get(stamp)), ) return { "time": stamp, "score": round(sum(item["points"] for item in evidence), 4), "evidence": evidence, "missing_layers": sorted(set( missing + [layer for layer in context["feature"]["blocked_layers"] if layer not in OBSERVATION_ONLY_LAYERS] )), } def make_recording_provider( contexts: Sequence[dict[str, Any]], store: FeatureStore, d60_by_time: dict[str, dict[str, Any] | None], ) -> Callable[[dict[str, Any]], list[dict[str, Any]]]: transit_cache: dict[tuple[Any, ...], dict[str, Any]] = {} def provider(request: dict[str, Any]) -> list[dict[str, Any]]: return [ recorded_row(request, context, store=store, transit_cache=transit_cache, d60_by_time=d60_by_time) for context in contexts ] return provider def score_map(rows: Sequence[dict[str, Any]]) -> dict[str, float]: return {str(row["time"])[:5]: float(row.get("score") or 0) for row in rows} def reconcile_rows(production: Sequence[dict[str, Any]], research: Sequence[dict[str, Any]]) -> dict[str, Any]: """Per-minute score and rule-id comparison. `changed == 0` is the R-0 gate.""" prod = {str(r["time"])[:5]: r for r in production} rese = {str(r["time"])[:5]: r for r in research} changed: list[dict[str, Any]] = [] for time, row in prod.items(): other = rese.get(time) if other is None: changed.append({"time": time, "reason": "missing"}) continue if abs(float(row.get("score") or 0) - float(other.get("score") or 0)) > EPS: changed.append({"time": time, "reason": "score", "production": row.get("score"), "research": other.get("score")}) continue prod_rules = { (item["event_id"], tuple(sorted(item["rule_ids"]))) for item in row.get("evidence") or [] } rese_rules = { (item["event_id"], tuple(sorted(item["rule_ids"]))) for item in other.get("evidence") or [] } if prod_rules != rese_rules: changed.append({"time": time, "reason": "rule_ids"}) return {"candidates": len(prod), "changed": len(changed), "details": changed[:5]} # --------------------------------------------------------------------------- # feature matrices and likelihood-ratio weights # --------------------------------------------------------------------------- def event_feature_matrix( store: FeatureStore, event_id: str, times: Sequence[str], *, include_extras: bool, ) -> dict[str, dict[str, float]]: """HH:MM -> feature -> fraction of sample dates on which it fired.""" samples = store.samples_for(event_id) out: dict[str, dict[str, float]] = {} if not samples: return {time: {} for time in times} for time in times: counts: dict[str, int] = defaultdict(int) for sample in samples: cell = store.rows.get((event_id, sample), {}).get(time) if not cell: continue names = [name for name in cell["rules"] if is_scoring_rule(name)] if include_extras: names += list(cell["extras"]) for name in names: counts[name] += 1 out[time] = {name: round(count / len(samples), 6) for name, count in counts.items()} return out def training_event_ids(request: dict[str, Any]) -> list[str]: holdout = holdout_event_ids(request["events"]) return [str(e["id"]) for e in request["events"] if is_scoreable_event(e) and str(e["id"]) not in holdout] @dataclass class LRTable: alpha: float positives: float negatives: float present: dict[str, float] # log LR of the feature firing absent: dict[str, float] # log LR of the feature not firing (naive Bayes complement) support: dict[str, dict[str, float]] def weight(self, name: str, *, allow_negative: bool) -> tuple[float, float]: """(present weight, absent weight) under the variant policy.""" present = self.present.get(name, 0.0) absent = self.absent.get(name, 0.0) if allow_negative: return present, absent return max(present, 0.0), 0.0 def estimate_lr(examples: Sequence[tuple[dict[str, float], bool]], *, alpha: float = LAPLACE_ALPHA) -> LRTable: """Per-feature log likelihood ratios from (feature fractions, is-truth-cluster) examples.""" positives = float(sum(1 for _values, truth in examples if truth)) negatives = float(sum(1 for _values, truth in examples if not truth)) names: set[str] = set() pos_sum: dict[str, float] = defaultdict(float) neg_sum: dict[str, float] = defaultdict(float) for values, truth in examples: for name, value in values.items(): names.add(name) if truth: pos_sum[name] += float(value) else: neg_sum[name] += float(value) present: dict[str, float] = {} absent: dict[str, float] = {} support: dict[str, dict[str, float]] = {} for name in sorted(names): p_pos = (pos_sum[name] + alpha) / (positives + 2 * alpha) p_neg = (neg_sum[name] + alpha) / (negatives + 2 * alpha) present[name] = round(math.log(p_pos / p_neg), 6) absent[name] = round(math.log((1 - p_pos) / (1 - p_neg)), 6) support[name] = { "fires_truth": round(pos_sum[name], 4), "fires_other": round(neg_sum[name], 4), "p_truth": round(p_pos, 6), "p_other": round(p_neg, 6), } return LRTable(alpha=alpha, positives=positives, negatives=negatives, present=present, absent=absent, support=support) def bootstrap_lr( per_case_examples: dict[str, list[tuple[dict[str, float], bool]]], *, resamples: int = 200, seed: int = 20260929, alpha: float = LAPLACE_ALPHA, ) -> dict[str, dict[str, float]]: """Case-level bootstrap 5–95 % interval of the present-log-LR per feature.""" ids = sorted(per_case_examples) rng = Random(f"{seed}:lr-bootstrap") draws: dict[str, list[float]] = defaultdict(list) for _ in range(resamples): picked = [rng.choice(ids) for _ in ids] table = estimate_lr([ex for cid in picked for ex in per_case_examples[cid]], alpha=alpha) for name, value in table.present.items(): draws[name].append(value) out: dict[str, dict[str, float]] = {} for name, values in draws.items(): ordered = sorted(values) lo = ordered[int(0.05 * (len(ordered) - 1))] hi = ordered[int(0.95 * (len(ordered) - 1))] out[name] = {"p05": round(lo, 4), "p95": round(hi, 4), "draws": len(ordered)} return out def weighted_points(values: dict[str, float], table: LRTable, *, allow_negative: bool, universe: Sequence[str]) -> float: total = 0.0 for name in universe: present, absent = table.weight(name, allow_negative=allow_negative) x = float(values.get(name, 0.0)) total += x * present + (1.0 - x) * absent return total def make_weighted_provider( store: FeatureStore, table: LRTable, *, variant: str, scale: float = 1.0, baseline_rows_by_sample: dict[tuple[str, str], dict[str, dict[str, Any]]] | None = None, ) -> Callable[[dict[str, Any]], list[dict[str, Any]]]: """Row provider that scores each (event, sample date) as Σ feature weights. Production rule ids are kept on the evidence so the matrix's kind factor and technique layers behave as in production; only the points change. """ include_extras, allow_negative = VARIANT_SPEC[variant] universe = [ name for name in sorted(table.present) if (include_extras or not is_extra_feature(name)) ] def provider(request: dict[str, Any]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for event in request["events"]: key = (str(event["id"]), str(event["date"])) cells = store.rows.get(key, {}) for time in sorted(cells, key=clock): cell = cells[time] names = [name for name in cell["rules"] if is_scoring_rule(name)] if include_extras: names += list(cell["extras"]) values = {name: 1.0 for name in names} points = round(scale * weighted_points(values, table, allow_negative=allow_negative, universe=universe), 4) rows.append({ "time": time, "score": points, "evidence": [{ "event_id": event["id"], "domain": event["domain"], "candidate_time": time, "rule_ids": list(cell["rules"]), "points": points, }], "missing_layers": [], }) return rows return provider def loo_folds(case_ids: Sequence[str]) -> list[tuple[str, list[str]]]: ids = list(case_ids) return [(held, [other for other in ids if other != held]) for held in ids] # --------------------------------------------------------------------------- # truth labels, priors, replay # --------------------------------------------------------------------------- def truth_cluster_times(contexts: Sequence[dict[str, Any]], true_time: str) -> list[str]: for cluster in raw_signature_clusters(contexts): times = [str(item)[:5] for item in cluster.get("times") or []] if true_time[:5] in times: return sorted(times, key=clock) return [true_time[:5]] def public_for(rows: Sequence[dict[str, Any]], contexts: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: raw = raw_signature_clusters(contexts) by_time = {stamp: row for row in rows if (stamp := hhmm(row.get("time")))} merged, _trace = merge_adjacent_traced(raw, by_time) public = public_from_clusters(merged, rows) for row in public: row["score"] = float(row.get("score") or 0) return public def percent_proportional(scores: dict[str, float]) -> dict[str, float]: total = sum(max(value, 0.0) for value in scores.values()) if total <= 0: return {key: 0.0 for key in scores} return {key: float(round(max(value, 0.0) / total * 100)) for key, value in scores.items()} def softmax_probabilities(scores: dict[str, float], temperature: float = 1.0) -> dict[str, float]: if not scores: return {} temp = float(temperature) if temperature and temperature > 0 else 1.0 peak = max(scores.values()) weights = {key: math.exp((value - peak) / temp) for key, value in scores.items()} total = sum(weights.values()) return {key: weight / total for key, weight in weights.items()} def percent_softmax(scores: dict[str, float], temperature: float = 1.0) -> dict[str, float]: return {key: float(round(p * 100)) for key, p in softmax_probabilities(scores, temperature).items()} TEMPERATURE_GRID = (0.25, 0.5, 1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 128.0, 256.0) def truth_mass(public: Sequence[dict[str, Any]], probabilities: dict[str, float], truth_times: Sequence[str]) -> float: truth = {str(t)[:5] for t in truth_times} mass = 0.0 for row in public: stamp = str(row["time"])[:5] members = {str(t)[:5] for t in (row.get("cluster_times") or [stamp])} if members & truth: mass += probabilities.get(stamp, 0.0) return mass def fit_temperature( training: Sequence[tuple[Sequence[dict[str, Any]], Sequence[str]]], grid: Sequence[float] = TEMPERATURE_GRID, ) -> float: """Platt-style single temperature: maximise Σ log P(truth cluster) on training cases only.""" best_t, best_ll = 1.0, None for temp in grid: ll = 0.0 for public, truth_times in training: base = {str(r["time"])[:5]: float(r.get("score") or 0) for r in public} mass = truth_mass(public, softmax_probabilities(base, temp), truth_times) ll += math.log(max(mass, 1e-9)) if best_ll is None or ll > best_ll + 1e-12: best_t, best_ll = float(temp), ll return best_t def prior_for(public: Sequence[dict[str, Any]], *, mode: str, softmax: bool, temperature: float = 1.0) -> dict[str, float]: base = {str(row["time"])[:5]: float(row.get("score") or 0) for row in public} if mode == "raw": return base return percent_softmax(base, temperature) if softmax else percent_proportional(base) def apply_scaled_answer( scores: dict[str, float], conflicts: dict[str, int], eliminated: set[str], probe: dict[str, Any], answer: str, candidate_times: Sequence[str], factor: float, ) -> tuple[dict[str, float], dict[str, int], set[str]]: """`apply_answer` at card strength (factor == 1); weaker evidence scales the delta and never counts a strong conflict (so it can never eliminate).""" if factor >= 1.0 - EPS: return apply_answer(scores, conflicts, eliminated, probe, answer, candidate_times) yes, no = outcome_groups(probe) next_scores = dict(scores) for time in candidate_times: if time in eliminated: continue if time in yes: raw = "support" elif time in no: raw = "conflict" else: raw = inherit_direction(time, yes, no) if answer == "no": raw = {"support": "conflict", "conflict": "support", "neutral": "neutral"}[raw] next_scores[time] = next_scores.get(time, 0.0) + SCORE_DELTA[raw] * factor return next_scores, dict(conflicts), set(eliminated) def replay( *, public: Sequence[dict[str, Any]], prior: dict[str, float], probes: Sequence[dict[str, Any]], true_time: str, window_times: Sequence[str], pre_answers: Sequence[tuple[dict[str, Any], str, float]] = (), answers: Sequence[str | None] | None = None, ) -> dict[str, Any]: """Prior → optional pre-answers (probe, answer, factor) → six probes answered from truth.""" reps = [str(row["time"])[:5] for row in public] scores = {time: float(prior.get(time, 0.0)) for time in reps} conflicts = {time: 0 for time in reps} eliminated: set[str] = set() for probe, answer, factor in pre_answers: scores, conflicts, eliminated = apply_scaled_answer(scores, conflicts, eliminated, probe, answer, reps, factor) asked = list(probes)[:ASK_COUNT] given = list(answers) if answers is not None else [optimal_answer(p, true_time) for p in asked] answered = 0 for probe, answer in zip(asked, given): if answer is None: continue answered += 1 scores, conflicts, eliminated = apply_answer(scores, conflicts, eliminated, probe, answer, reps) posterior = [{**row, "score": scores.get(str(row["time"])[:5], row.get("score") or 0)} for row in public] valid = still_valid_public(posterior, scores, eliminated, lead=SEPARATION_LEAD) delivery = delivery_from_public(valid) alive = [row for row in posterior if str(row["time"])[:5] not in eliminated] metrics = metrics_bundle( public=alive, true_time=true_time, window_times=window_times, delivery_times=delivery["times"], delivery_width=delivery["width"], independent=True, entropy_scores=[scores.get(str(row["time"])[:5], 0.0) for row in alive], ) truth_eliminated = any( true_time in (row.get("cluster_times") or [str(row.get("time"))[:5]]) and str(row["time"])[:5] in eliminated for row in public ) return { "top1": bool(metrics["top1_hit"]), "coverage": bool(metrics["coverage"]), "width": delivery["width"], "tie": bool(metrics["tie"]), "squeezed": bool(metrics["truth_squeezed"]), "eliminated": len(eliminated), "truth_eliminated": bool(truth_eliminated), "questions": answered, "entropy": round(shannon_entropy(max(s, 0.0) for t, s in scores.items() if t not in eliminated), 4), } def engine_top1(public: Sequence[dict[str, Any]], true_time: str) -> bool: from scripts.research.cluster_width_lib import top1_from_public return bool(top1_from_public(public, true_time)) def within_case_range(rows: Sequence[dict[str, Any]]) -> float: scores = [float(row.get("score") or 0) for row in rows] return (max(scores) - min(scores)) if scores else 0.0 def summarize(rows: Sequence[dict[str, Any]]) -> dict[str, Any]: """Same keys as precision_gate_sweep.summarize so `gate_verdict` can read it.""" if not rows: return {"n": 0, "top1": None, "coverage": None, "width_median": None, "tie": None, "squeezed": 0, "engine_top1": None, "refresh_mean": 0.0, "entropy": None, "truth_eliminated": 0} n = len(rows) return { "n": n, "top1": round(sum(1 for r in rows if r.get("top1")) / n, 4), "coverage": round(sum(1 for r in rows if r.get("coverage")) / n, 4), "width_median": median([r.get("width") for r in rows]), "tie": round(sum(1 for r in rows if r.get("tie")) / n, 4), "squeezed": sum(1 for r in rows if r.get("squeezed")), "engine_top1": round(sum(1 for r in rows if r.get("engine_top1")) / n, 4), "refresh_mean": 0.0, "entropy": round(sum(float(r.get("entropy") or 0) for r in rows) / n, 4), "truth_eliminated": sum(1 for r in rows if r.get("truth_eliminated")), } # --------------------------------------------------------------------------- # probes on representatives (typed-event convention), absence, precision follow-up # --------------------------------------------------------------------------- def production_probes(request: dict[str, Any], built: dict[str, Any], times: Sequence[str], true_time: str) -> list[dict[str, Any]]: payload = {**request, "refresh_probes": False, "asked_probe_keys": []} return event_probes.discriminating_event_probes( payload, built, scan=window_scan(built), candidate_times=list(times), representative_time=true_time, today=TODAY, ) @dataclass class Representatives: reps: list[dict[str, Any]] clusters: list[dict[str, Any]] set_version: str def representatives_for(contexts: Sequence[dict[str, Any]]) -> Representatives: full = [item for item in contexts if event_probes._context_time(item)] full.sort(key=lambda item: event_probes._clock(str(event_probes._context_time(item)))) clusters = cluster_contexts_by_signature(full) reps = [c["representative"] for c in clusters if event_probes._scoreable(c["representative"])] if len(reps) < 2: reps = [item for item in full if event_probes._scoreable(item)] version = event_probes.candidate_set_version([c["times"] for c in clusters]) return Representatives(reps=reps, clusters=clusters, set_version=version) def split_probe( representatives: Representatives, *, birth_date: str, domain: str, year: int, month: int | None, source: str, ) -> dict[str, Any] | None: canonical = event_probes.canonical_domain(domain) if canonical not in event_probes.DOMAIN_CATALOG: return None return event_probes._evaluate_contexts( representatives.reps, birth_date=birth_date, domain=canonical, year=year, month=month, source=source, clusters=representatives.clusters, set_version=representatives.set_version, ) def event_year(event: dict[str, Any]) -> int | None: raw = str(event.get("date_start") or event.get("date") or "") return int(raw[:4]) if raw[:4].isdigit() else None def event_month(event: dict[str, Any]) -> int | None: raw = str(event.get("date_start") or event.get("date") or "") return int(raw[5:7]) if len(raw) >= 7 and raw[4] == "-" else None def absent_years(events: Sequence[dict[str, Any]]) -> dict[str, list[int]]: """Per domain: years strictly inside the span of that domain's dated events with no event.""" years: dict[str, set[int]] = defaultdict(set) for event in events: year = event_year(event) if year is None: continue years[str(event.get("domain") or "")].add(year) out: dict[str, list[int]] = {} for domain, known in years.items(): if len(known) < 2: continue lo, hi = min(known), max(known) gaps = [year for year in range(lo + 1, hi) if year not in known] if gaps: out[domain] = gaps return dict(sorted(out.items())) def drop_events(events: Sequence[dict[str, Any]], count: int, *, seed: str) -> list[dict[str, Any]]: rng = Random(seed) if count <= 0 or count >= len(events): return list(events) dropped = set(rng.sample(range(len(events)), count)) return [event for index, event in enumerate(events) if index not in dropped] def degrade_to_year(events: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: """The 'spoken' form of a case: every day/month event becomes year precision.""" out = [] for event in events: item = dict(event) if str(item.get("precision")) in {"day", "month"}: item["precision"] = "year" item["date"] = str(item.get("date") or "")[:4] out.append(item) return out def upgrade_to_month(events: Sequence[dict[str, Any]], event_ids: set[str], originals: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: out = [] for event in events: item = dict(event) original = originals.get(str(item.get("id"))) if str(item.get("id")) in event_ids and original is not None and len(str(original.get("date") or "")) >= 7: item["precision"] = "month" item["date"] = str(original["date"])[:7] out.append(item) return out def shift_day_events_by_months(events: Sequence[dict[str, Any]], *, seed: str, choices: Sequence[int] = (-3, -2, -1, 1, 2, 3)) -> list[dict[str, Any]]: rng = Random(seed) out = [] for event in events: item = dict(event) raw = str(item.get("date") or "") if str(item.get("precision")) == "day" and len(raw) >= 10: year, month, day = int(raw[:4]), int(raw[5:7]), int(raw[8:10]) index = year * 12 + (month - 1) + rng.choice(list(choices)) new_year, new_month = divmod(index, 12) item["date"] = f"{new_year:04d}-{new_month + 1:02d}-{min(day, 28):02d}" item["shift_months"] = index - (year * 12 + month - 1) out.append(item) return out def calibration_bins(points: Sequence[tuple[float, bool]], edges: Sequence[float] = (0.0, 0.05, 0.1, 0.2, 0.3, 0.5, 0.7, 1.0001)) -> list[dict[str, Any]]: out = [] for lo, hi in zip(edges, edges[1:]): inside = [(p, t) for p, t in points if lo <= p < hi] out.append({ "bin": f"[{lo:.2f},{min(hi, 1.0):.2f})", "n": len(inside), "predicted_mean": round(sum(p for p, _ in inside) / len(inside), 4) if inside else None, "observed_rate": round(sum(1 for _, t in inside if t) / len(inside), 4) if inside else None, }) return out __all__ = [name for name in dir() if not name.startswith("__")]