research(rectification): scoring-method scaffold — LR weights, absence, precision follow-up pipelines on v4, verdicts pending v5 (BUG-1091)
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017eEAG8HD3mm8gsKXgk8uU8
This commit is contained in:
Jesse_Chen
co-authored by
Claude Fable 5.1
Claude Fable 5.1
parent
5f84637ea9
commit
abf2ad6aea
@@ -0,0 +1,820 @@
|
||||
"""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("__")]
|
||||
Reference in New Issue
Block a user