research(rectification): varga-resolution M0 — per-minute ascendant segments and six-probe interval chart types on v5 (BUG-1105)
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
3263db0f36
commit
f798366f89
@@ -1,302 +1,634 @@
|
||||
"""Offline segment-oriented rectification research for BUG-1105.
|
||||
#!/usr/bin/env python3
|
||||
"""Varga-resolution research library (TASK-rectification-varga-resolution-research-20260930).
|
||||
|
||||
The production candidate scorer is only used as an observation source. This
|
||||
module never changes production defaults. A segment is a maximal contiguous
|
||||
run of equal divisional ascendant sign; equal signs separated by another sign
|
||||
remain different segments.
|
||||
Offline only. Nothing here is imported by production code, and nothing here
|
||||
changes production scoring: the six-probe replay reuses the production probe
|
||||
generator (`event_probes.discriminating_event_probes`), the production card
|
||||
update (`probe_supply_after_six.apply_answer`, ±2 / three strong conflicts
|
||||
eliminate) and the production delivery rule (`cluster_width_lib.still_valid_public`,
|
||||
lead 8). What this module adds on top is bookkeeping *by varga ascendant
|
||||
segment* instead of by candidate minute:
|
||||
|
||||
* `signs_per_minute` — D1 / D9 / D10 / D12 ascendant sign for every minute
|
||||
offset in the search window (step 1);
|
||||
* `segments_for` — contiguous runs of one sign = "chart-type segments";
|
||||
* `segment_shares` — posterior mass per segment (raw / percent / uniform)
|
||||
over the still-valid candidate minutes;
|
||||
* `segment_information_gain` + `select_probes_by_segment` — research-side
|
||||
re-ranking of the *same* probe pool by how well an
|
||||
answer separates the segments of a target varga.
|
||||
|
||||
Calculation profile is the repository default (`raman`, mean node), the
|
||||
candidate step is the production research step (2 minutes) and the six-probe
|
||||
replay answers from the true minute, exactly as the 09-14 / 09-26 / 09-29
|
||||
studies did.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
from typing import Any, Iterable, Sequence
|
||||
import json
|
||||
import math
|
||||
import pickle
|
||||
import sys
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from random import Random
|
||||
from typing import Any, Sequence
|
||||
|
||||
from scripts.active_rectification_event_engine import compute_candidate_static_contexts
|
||||
from scripts.rectification.refinement_packet import window_scan
|
||||
from scripts.rectification.scoring_service import build_event_contribution_matrix, score_from_matrix
|
||||
from scripts.research.minute_resolution_sweep import scoring_request_for
|
||||
from scripts.research.probe_supply_after_six import ASK_COUNT, apply_answer, optimal_answer
|
||||
from scripts.research.scoring_research_lib import (
|
||||
public_for,
|
||||
replay,
|
||||
reconcile_rows,
|
||||
score_map,
|
||||
truth_cluster_times,
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
SCRIPTS = ROOT / "scripts"
|
||||
if str(SCRIPTS) not in sys.path:
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
|
||||
import domain_calculation_service # noqa: E402
|
||||
import varga as varga_module # noqa: E402
|
||||
from scripts.active_rectification_event_engine import ( # noqa: E402
|
||||
AYANAMSA,
|
||||
NODE_MODE,
|
||||
compute_candidate_static_contexts,
|
||||
)
|
||||
from scripts.research.cluster_width_lib import SEPARATION_LEAD, delivery_from_public, still_valid_public
|
||||
from scripts.rectification.event_probes import discriminating_event_probes
|
||||
from scripts.rectification.event_probes import discriminating_event_probes # 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,
|
||||
scoreable_request,
|
||||
)
|
||||
from scripts.research.cluster_width_lib import ( # noqa: E402
|
||||
SEPARATION_LEAD,
|
||||
delivery_from_public,
|
||||
hhmm,
|
||||
still_valid_public,
|
||||
)
|
||||
from scripts.research.guided_collect_holdout_replay import TODAY # noqa: E402
|
||||
from scripts.research.minute_resolution_sweep import MINUTE_STEP, scoring_request_for # noqa: E402
|
||||
from scripts.research.probe_supply_after_six import ( # noqa: E402
|
||||
ASK_COUNT,
|
||||
apply_answer,
|
||||
inherit_direction,
|
||||
optimal_answer,
|
||||
outcome_groups,
|
||||
)
|
||||
from scripts.research.scoring_research_lib import public_for # noqa: E402
|
||||
|
||||
VARGA_PREFIXES = ("D1", "D9", "D10")
|
||||
RADII = (10, 15, 30, 60)
|
||||
THRESHOLDS = (0.5, 0.6, 0.7, 0.8, 0.9)
|
||||
TODAY = date(2026, 9, 16)
|
||||
HOLDOUT_V5 = ROOT / "references" / "real_case_calibration" / "minute_rectification_holdout_v5.json"
|
||||
VARGAS: tuple[str, ...] = ("D1", "D9", "D10", "D12")
|
||||
RADII: tuple[int, ...] = (10, 30, 60)
|
||||
COUNT_RADII: tuple[int, ...] = (10, 15, 30, 60)
|
||||
MODES: tuple[str, ...] = ("raw", "percent", "uniform")
|
||||
THRESHOLDS: tuple[float, ...] = (0.5, 0.6, 0.7, 0.8, 0.9)
|
||||
TARGET_ACCURACIES: tuple[float, ...] = (0.8, 0.9)
|
||||
MIN_TRAINING_CASES = 5
|
||||
#: Question-domain strategies (decision record: 事业 → D1+D10, 婚恋 → D1+D9, 综合 → D1+D9+D10).
|
||||
STRATEGIES: dict[str, tuple[str, ...]] = {
|
||||
"career": ("D1", "D10"),
|
||||
"relationship": ("D1", "D9"),
|
||||
"general": ("D1", "D9", "D10"),
|
||||
}
|
||||
LMT_ERA_BEFORE_YEAR = 1900
|
||||
DAY_SHIFT = 7
|
||||
FLIP_SEEDS = 5
|
||||
EPS = 1e-9
|
||||
|
||||
|
||||
def clock(stamp: str) -> int:
|
||||
hour, minute = str(stamp)[:5].split(":")
|
||||
return int(hour) * 60 + int(minute)
|
||||
# ---------------------------------------------------------------------------
|
||||
# cases and clocks
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def load_cases(path: Path = HOLDOUT_V5) -> list[dict[str, Any]]:
|
||||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||||
return list(payload.get("cases") or [])
|
||||
|
||||
|
||||
def sign_value(row: dict[str, Any], prefix: str) -> Any:
|
||||
value = row.get(prefix)
|
||||
if isinstance(value, dict):
|
||||
return value.get("sign_idx", value.get("sign"))
|
||||
return value
|
||||
def clock(value: str) -> int:
|
||||
hours, minutes = str(value)[:5].split(":")
|
||||
return int(hours) * 60 + int(minutes)
|
||||
|
||||
|
||||
def segment_rows(rows: Sequence[dict[str, Any]], prefix: str) -> list[dict[str, Any]]:
|
||||
"""Return maximal sampled runs without merging a non-contiguous sign.
|
||||
|
||||
Clock time is cyclic here: 23:59 followed by 00:00 is one minute apart.
|
||||
A missing sample (for example 23:59 followed by 00:01) still starts a new
|
||||
run. This deliberately uses the observation order rather than a set of
|
||||
signs, so A-B-A produces three separately numbered segments.
|
||||
"""
|
||||
segments: list[dict[str, Any]] = []
|
||||
current: dict[str, Any] | None = None
|
||||
for row in rows:
|
||||
stamp = str(row.get("time") or row.get("stamp") or "")[:5]
|
||||
value = sign_value(row, prefix)
|
||||
if not stamp or value is None:
|
||||
current = None
|
||||
continue
|
||||
minute = clock(stamp)
|
||||
contiguous = (
|
||||
current is not None
|
||||
and (minute - int(current["last_minute"])) % 1440 == 1
|
||||
)
|
||||
if current is None or current["value"] != value or not contiguous:
|
||||
current = {
|
||||
"segment_id": len(segments),
|
||||
"varga": prefix,
|
||||
"value": value,
|
||||
"start": stamp,
|
||||
"end": stamp,
|
||||
"times": [stamp],
|
||||
"last_minute": minute,
|
||||
}
|
||||
segments.append(current)
|
||||
else:
|
||||
current["end"] = stamp
|
||||
current["times"].append(stamp)
|
||||
current["last_minute"] = minute
|
||||
for segment in segments:
|
||||
segment.pop("last_minute", None)
|
||||
return segments
|
||||
def offset_of(stamp: str, true_time: str) -> int:
|
||||
"""Signed minute offset of a HH:MM candidate from the true minute, nearest across midnight."""
|
||||
delta = (clock(stamp) - clock(true_time)) % 1440
|
||||
return delta - 1440 if delta > 720 else delta
|
||||
|
||||
|
||||
def segment_members(rows: Sequence[dict[str, Any]], prefix: str) -> list[list[str]]:
|
||||
return [list(item["times"]) for item in segment_rows(rows, prefix)]
|
||||
def birth_datetime(case: dict[str, Any]) -> datetime:
|
||||
birth = case["birth"]
|
||||
return datetime.strptime(f"{birth['date']} {str(birth['time'])[:5]}", "%Y-%m-%d %H:%M")
|
||||
|
||||
|
||||
def row_signature(context: dict[str, Any], prefix: str) -> int | str | None:
|
||||
if prefix == "D1":
|
||||
value = context.get("ascendant_index")
|
||||
else:
|
||||
charts = context.get("varga_charts") or {}
|
||||
value = ((charts.get(prefix) or {}).get("Ascendant") or {}).get("sign_idx")
|
||||
return value if isinstance(value, (int, str)) else None
|
||||
def is_lmt_era(case: dict[str, Any]) -> bool:
|
||||
return int(str(case["birth"]["date"])[:4]) < LMT_ERA_BEFORE_YEAR
|
||||
|
||||
|
||||
def contexts_to_rows(contexts: Sequence[dict[str, Any]], prefixes: Iterable[str] = VARGA_PREFIXES) -> list[dict[str, Any]]:
|
||||
rows: list[dict[str, Any]] = []
|
||||
for context in contexts:
|
||||
feature = context.get("feature")
|
||||
stamp = feature.get("time") if isinstance(feature, dict) else None
|
||||
if not stamp:
|
||||
stamp = context["candidate_at"].strftime("%H:%M")
|
||||
rows.append({"time": str(stamp)[:5], **{p: row_signature(context, p) for p in prefixes}})
|
||||
return rows
|
||||
# ---------------------------------------------------------------------------
|
||||
# M0 · ascendant signs per minute and segments
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _divisions(vargas: Sequence[str]) -> list[int]:
|
||||
return sorted({int(name[1:]) for name in vargas if name != "D1"})
|
||||
|
||||
|
||||
def truth_segment(rows: Sequence[dict[str, Any]], prefix: str, truth_time: str) -> dict[str, Any] | None:
|
||||
for segment in segment_rows(rows, prefix):
|
||||
if truth_time[:5] in segment["times"]:
|
||||
return segment
|
||||
return None
|
||||
|
||||
|
||||
def unique_sign_count(rows: Sequence[dict[str, Any]], prefix: str) -> int:
|
||||
return len({sign_value(row, prefix) for row in rows if sign_value(row, prefix) is not None})
|
||||
|
||||
|
||||
def unique_segment_count(rows: Sequence[dict[str, Any]], prefix: str) -> int:
|
||||
return len(segment_rows(rows, prefix))
|
||||
|
||||
|
||||
def window_payload(case: dict[str, Any], radius: int) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]:
|
||||
"""Build the two-minute scoring grid and an independent one-minute scan grid."""
|
||||
scoring_request = scoring_request_for({**case, "candidate_radius_minutes": radius}, radius)
|
||||
scoring_request["minute_step"] = 2
|
||||
scoring_contexts = compute_candidate_static_contexts(scoring_request)
|
||||
scan_request = scoring_request_for({**case, "candidate_radius_minutes": radius}, radius)
|
||||
scan_request["minute_step"] = 1
|
||||
scan_contexts = compute_candidate_static_contexts(scan_request)
|
||||
return scoring_request, scoring_contexts, contexts_to_rows(scan_contexts)
|
||||
|
||||
|
||||
def _probe_payload(request: dict[str, Any], built: dict[str, Any], times: Sequence[str], true_time: str) -> list[dict[str, Any]]:
|
||||
return discriminating_event_probes(
|
||||
{**request, "refresh_probes": False, "asked_probe_keys": []},
|
||||
built,
|
||||
scan=window_scan(built),
|
||||
candidate_times=list(times),
|
||||
representative_time=true_time,
|
||||
today=TODAY,
|
||||
)
|
||||
|
||||
|
||||
def replay_state(rows: Sequence[dict[str, Any]], contexts: Sequence[dict[str, Any]], request: dict[str, Any], true_time: str, *, probes: Sequence[dict[str, Any]] | None = None, answers: Sequence[str | None] | None = None) -> dict[str, Any]:
|
||||
times = [str(row["time"])[:5] for row in rows]
|
||||
public = public_for(rows, contexts)
|
||||
reps = [str(row["time"])[:5] for row in public]
|
||||
scores = {stamp: float(row.get("score") or 0) for stamp, row in ((str(item["time"])[:5], item) for item in public)}
|
||||
conflicts = {stamp: 0 for stamp in reps}
|
||||
eliminated: set[str] = set()
|
||||
actual_probes = list(probes) if probes is not None else _probe_payload(request, {"static_contexts": list(contexts), "rows": list(rows)}, times, true_time)
|
||||
given = list(answers) if answers is not None else [optimal_answer(p, true_time) for p in actual_probes[:ASK_COUNT]]
|
||||
for probe, answer in zip(actual_probes[:ASK_COUNT], given):
|
||||
if answer in {"yes", "no", "weak_yes"}:
|
||||
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)
|
||||
return {
|
||||
"result": {"questions": sum(answer is not None for answer in given)},
|
||||
"public": public,
|
||||
"posterior": posterior,
|
||||
"valid": valid,
|
||||
"scores": scores,
|
||||
"eliminated": eliminated,
|
||||
"probes": actual_probes,
|
||||
"delivery": delivery_from_public(valid),
|
||||
def signs_at(birth: dict[str, Any], moment: datetime, vargas: Sequence[str] = VARGAS) -> dict[str, int]:
|
||||
chart = domain_calculation_service.compute_chart({
|
||||
"year": moment.year, "month": moment.month, "day": moment.day,
|
||||
"hour": moment.hour, "minute": moment.minute,
|
||||
"lat": float(birth["latitude"]), "lon": float(birth["longitude"]),
|
||||
"tz": float(birth["timezone_offset"]),
|
||||
"ayanamsa": AYANAMSA, "node_mode": NODE_MODE,
|
||||
})
|
||||
planet_longitudes = {
|
||||
name: float(item["lon"]) for name, item in chart.get("planets", {}).items()
|
||||
if isinstance(item, dict) and isinstance(item.get("lon"), (int, float))
|
||||
}
|
||||
|
||||
|
||||
def native_case(case: dict[str, Any], radius: int, *, do_reconcile: bool = False) -> dict[str, Any]:
|
||||
request, contexts, chart_rows = window_payload(case, radius)
|
||||
true_time = str(case["birth"]["time"])[:5]
|
||||
built = build_event_contribution_matrix(request, static_contexts=contexts)
|
||||
rows = score_from_matrix(request, built)
|
||||
times = [str(row["time"])[:5] for row in rows]
|
||||
probes = _probe_payload(request, built, times, true_time)
|
||||
state = replay_state(rows, contexts, request, true_time, probes=probes)
|
||||
reconciliation = {"status": "not_run"}
|
||||
if do_reconcile:
|
||||
from scripts.research.scoring_research_lib import FeatureStore, d60_charts, make_recording_provider
|
||||
store = FeatureStore()
|
||||
recording = build_event_contribution_matrix(request, row_provider=make_recording_provider(contexts, store, d60_charts(contexts)), static_contexts=contexts)
|
||||
recording_rows = score_from_matrix(request, recording)
|
||||
reconciliation = reconcile_rows(rows, recording_rows)
|
||||
return {"case_id": str(case["case_id"]), "radius": radius, "true_time": true_time, "request": request, "contexts": contexts, "chart_rows": chart_rows, "rows": rows, "probes": probes, "state": state, "reconciliation": reconciliation}
|
||||
|
||||
|
||||
def valid_minute_scores(
|
||||
state: dict[str, Any],
|
||||
chart_rows: Sequence[dict[str, Any]],
|
||||
scores: dict[str, float] | None = None,
|
||||
) -> dict[str, float]:
|
||||
"""Project representative posterior scores onto each minute in its cluster."""
|
||||
effective_scores = scores if scores is not None else state["scores"]
|
||||
output: dict[str, float] = {}
|
||||
for row in state["posterior"]:
|
||||
representative = str(row["time"])[:5]
|
||||
value = float(effective_scores.get(representative, row.get("score") or 0))
|
||||
for stamp in row.get("cluster_times") or [representative]:
|
||||
output[str(stamp)[:5]] = value
|
||||
return output
|
||||
|
||||
|
||||
def segment_metrics(state: dict[str, Any], chart_rows: Sequence[dict[str, Any]], prefix: str, true_time: str, mode: str) -> dict[str, Any]:
|
||||
segments = segment_rows(chart_rows, prefix)
|
||||
scores = {str(key)[:5]: float(value) for key, value in state["scores"].items()}
|
||||
if mode == "percent":
|
||||
total = sum(max(value, 0.0) for value in scores.values())
|
||||
scores = (
|
||||
{key: max(value, 0.0) / total * 100.0 for key, value in scores.items()}
|
||||
if total > 0
|
||||
else {key: 0.0 for key in scores}
|
||||
)
|
||||
minute_scores = valid_minute_scores(state, chart_rows, scores)
|
||||
valid_times = {str(t)[:5] for row in state["valid"] for t in (row.get("cluster_times") or [row.get("time")])}
|
||||
truth = truth_segment(chart_rows, prefix, true_time)
|
||||
qualities: list[float] = []
|
||||
for segment in segments:
|
||||
values = [minute_scores.get(t, 0.0) for t in segment["times"] if t in valid_times]
|
||||
if mode == "uniform":
|
||||
quality = float(len(values))
|
||||
elif mode == "percent":
|
||||
quality = sum(max(v, 0.0) for v in values)
|
||||
else:
|
||||
quality = sum(values)
|
||||
qualities.append(quality)
|
||||
total = sum(qualities)
|
||||
share = max(qualities) / total if qualities and total > 0 else None
|
||||
leaders = [i for i, value in enumerate(qualities) if share is not None and abs(value - max(qualities)) <= 1e-9]
|
||||
truth_id = truth["segment_id"] if truth else None
|
||||
retained = truth_id is not None and truth_id in {segment["segment_id"] for segment in segments if any(t in valid_times for t in segment["times"])}
|
||||
correct = truth_id is not None and truth_id in leaders
|
||||
valid_segment_count = sum(1 for segment in segments if any(t in valid_times for t in segment["times"]))
|
||||
return {
|
||||
"prefix": prefix,
|
||||
"mode": mode,
|
||||
"segment_count_window": len(segments),
|
||||
"valid_segment_count": valid_segment_count,
|
||||
"truth_segment_id": truth_id,
|
||||
"truth_retained": bool(retained),
|
||||
"top_segment_correct": bool(correct),
|
||||
"top_segment_tie": len(leaders) > 1,
|
||||
"top_segment_ids": leaders,
|
||||
"top_share": None if share is None else round(share, 8),
|
||||
"segment_qualities": [round(v, 8) for v in qualities],
|
||||
}
|
||||
|
||||
|
||||
def threshold_scan(rows: Sequence[dict[str, Any]], thresholds: Sequence[float] = THRESHOLDS) -> dict[str, Any]:
|
||||
out: dict[str, Any] = {}
|
||||
for threshold in thresholds:
|
||||
eligible = [row for row in rows if row.get("top_share") is not None and float(row["top_share"]) >= threshold]
|
||||
out[str(threshold)] = {"n": len(eligible), "denominator": len(rows), "coverage": round(len(eligible) / len(rows), 8) if rows else None, "accuracy": round(sum(bool(row.get("top_segment_correct")) for row in eligible) / len(eligible), 8) if eligible else None, "truth_retained": round(sum(bool(row.get("truth_retained")) for row in eligible) / len(eligible), 8) if eligible else None}
|
||||
ascendant = float(chart["ascendant"]["lon"])
|
||||
out: dict[str, int] = {}
|
||||
if "D1" in vargas:
|
||||
out["D1"] = int(ascendant // 30) % 12
|
||||
divisions = _divisions(vargas)
|
||||
if divisions:
|
||||
computed = varga_module.calc_all_vargas(planet_longitudes, ascendant, divisions=divisions)
|
||||
for value in computed.values():
|
||||
division = int(value["_meta"]["div"])
|
||||
out[f"D{division}"] = int(value["Ascendant"]["sign_idx"])
|
||||
return out
|
||||
|
||||
|
||||
def choose_loo_threshold(training: Sequence[dict[str, Any]], thresholds: Sequence[float] = THRESHOLDS, minimum: int = 5) -> float | None:
|
||||
if not training:
|
||||
return None
|
||||
candidates = []
|
||||
required = min(minimum, len(training))
|
||||
for threshold in thresholds:
|
||||
eligible = [row for row in training if row.get("top_share") is not None and float(row["top_share"]) >= threshold]
|
||||
if len(eligible) < required or not eligible:
|
||||
def signs_per_minute(
|
||||
case: dict[str, Any],
|
||||
radius: int,
|
||||
*,
|
||||
vargas: Sequence[str] = VARGAS,
|
||||
cache_dir: Path | None = None,
|
||||
) -> dict[int, dict[str, int]]:
|
||||
"""{offset: {varga: sign_idx}} for every minute in [-radius, radius] (step 1)."""
|
||||
key = f"{case['case_id']}_{radius}_{'-'.join(vargas)}.json"
|
||||
if cache_dir is not None:
|
||||
path = cache_dir / key
|
||||
if path.exists():
|
||||
raw = json.loads(path.read_text(encoding="utf-8"))
|
||||
return {int(offset): dict(signs) for offset, signs in raw.items()}
|
||||
origin = birth_datetime(case)
|
||||
out = {
|
||||
offset: signs_at(case["birth"], origin + timedelta(minutes=offset), vargas)
|
||||
for offset in range(-radius, radius + 1)
|
||||
}
|
||||
if cache_dir is not None:
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
(cache_dir / key).write_text(
|
||||
json.dumps({str(k): v for k, v in out.items()}, sort_keys=True), encoding="utf-8",
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def segments_for(signs: dict[int, dict[str, int]], vargas: Sequence[str]) -> list[dict[str, Any]]:
|
||||
"""Contiguous minute runs where every varga in `vargas` keeps its ascendant sign.
|
||||
|
||||
Each segment: {"index", "start", "end", "key"} with key = tuple of signs.
|
||||
"""
|
||||
offsets = sorted(signs)
|
||||
segments: list[dict[str, Any]] = []
|
||||
for offset in offsets:
|
||||
key = tuple(int(signs[offset][name]) for name in vargas)
|
||||
if segments and segments[-1]["key"] == key and segments[-1]["end"] == offset - 1:
|
||||
segments[-1]["end"] = offset
|
||||
else:
|
||||
segments.append({"index": len(segments), "start": offset, "end": offset, "key": key})
|
||||
return segments
|
||||
|
||||
|
||||
def segment_index_of(segments: Sequence[dict[str, Any]], offset: int) -> int | None:
|
||||
for segment in segments:
|
||||
if segment["start"] <= offset <= segment["end"]:
|
||||
return int(segment["index"])
|
||||
return None
|
||||
|
||||
|
||||
def distinct_signs(signs: dict[int, dict[str, int]], radius: int, vargas: Sequence[str]) -> int:
|
||||
return len({tuple(int(signs[o][name]) for name in vargas) for o in range(-radius, radius + 1) if o in signs})
|
||||
|
||||
|
||||
def count_table(per_case: dict[str, dict[int, dict[str, int]]], radii: Sequence[int] = COUNT_RADII) -> dict[str, Any]:
|
||||
"""Table 1 of the brief: how many ascendant signs / combinations inside each window."""
|
||||
keys: dict[str, tuple[str, ...]] = {name: (name,) for name in VARGAS}
|
||||
keys["D1xD9xD10"] = ("D1", "D9", "D10")
|
||||
out: dict[str, Any] = {}
|
||||
for radius in radii:
|
||||
row: dict[str, Any] = {}
|
||||
for label, vargas in keys.items():
|
||||
counts = [distinct_signs(signs, radius, vargas) for signs in per_case.values()]
|
||||
row[label] = {
|
||||
"single": sum(1 for c in counts if c == 1),
|
||||
"n": len(counts),
|
||||
"mean": round(sum(counts) / len(counts), 2) if counts else None,
|
||||
"max": max(counts) if counts else None,
|
||||
}
|
||||
out[str(radius)] = row
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# six-probe replay with segment bookkeeping
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class CaseReplay:
|
||||
"""One case at one radius: production scores, production probe pool, public clusters."""
|
||||
|
||||
def __init__(self, case: dict[str, Any], radius: int, *, cache_dir: Path | None = None) -> None:
|
||||
self.case = case
|
||||
self.case_id = str(case["case_id"])
|
||||
self.radius = int(radius)
|
||||
self.true_time = str(case["birth"]["time"])[:5]
|
||||
self.request = scoring_request_for({**case, "candidate_radius_minutes": radius}, radius)
|
||||
self.request["ayanamsa"] = AYANAMSA
|
||||
self.request["node_mode"] = NODE_MODE
|
||||
self.request["minute_step"] = MINUTE_STEP
|
||||
self.contexts = self._contexts(cache_dir)
|
||||
self.times = [ctx["candidate_at"].strftime("%H:%M") for ctx in self.contexts]
|
||||
self.rows, self.probes = self.rows_and_probes(self.request)
|
||||
self.public = public_for(self.rows, self.contexts)
|
||||
self.signs = signs_per_minute(case, radius, cache_dir=cache_dir)
|
||||
|
||||
def _contexts(self, cache_dir: Path | None) -> list[dict[str, Any]]:
|
||||
if cache_dir is not None:
|
||||
path = cache_dir / f"{self.case_id}_{self.radius}.pkl"
|
||||
if path.exists():
|
||||
with path.open("rb") as handle:
|
||||
return pickle.load(handle)
|
||||
contexts = compute_candidate_static_contexts(self.request)
|
||||
if cache_dir is not None:
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
with (cache_dir / f"{self.case_id}_{self.radius}.pkl").open("wb") as handle:
|
||||
pickle.dump(contexts, handle)
|
||||
return contexts
|
||||
|
||||
def rows_and_probes(self, request: dict[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||||
built = build_event_contribution_matrix(request, static_contexts=self.contexts)
|
||||
rows = score_from_matrix(request, built)
|
||||
times = [stamp for row in rows if (stamp := hhmm(row.get("time")))]
|
||||
probes = discriminating_event_probes(
|
||||
{**request, "refresh_probes": False, "asked_probe_keys": []},
|
||||
built, scan=window_scan(built), candidate_times=times,
|
||||
representative_time=self.true_time, today=TODAY,
|
||||
)
|
||||
return rows, list(probes)
|
||||
|
||||
def shifted(self, days: int, *, seed: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]:
|
||||
"""Rows / probes / public after moving every day-precision event by ±days (random sign)."""
|
||||
rng = Random(seed)
|
||||
events = []
|
||||
for event in self.request["events"]:
|
||||
item = dict(event)
|
||||
raw = str(item.get("date") or "")
|
||||
if str(item.get("precision")) == "day" and len(raw) >= 10:
|
||||
when = datetime.strptime(raw[:10], "%Y-%m-%d") + timedelta(days=days * rng.choice((-1, 1)))
|
||||
item["date"] = when.strftime("%Y-%m-%d")
|
||||
events.append(item)
|
||||
request = scoreable_request({**self.request, "events": events})
|
||||
rows, probes = self.rows_and_probes(request)
|
||||
return rows, probes, public_for(rows, self.contexts)
|
||||
|
||||
# --- segment helpers ------------------------------------------------
|
||||
def segments(self, vargas: Sequence[str]) -> list[dict[str, Any]]:
|
||||
return segments_for(self.signs, vargas)
|
||||
|
||||
def truth_segment(self, segments: Sequence[dict[str, Any]]) -> int:
|
||||
index = segment_index_of(segments, 0)
|
||||
assert index is not None
|
||||
return index
|
||||
|
||||
|
||||
def replay_scores(
|
||||
public: Sequence[dict[str, Any]],
|
||||
probes: Sequence[dict[str, Any]],
|
||||
true_time: str,
|
||||
*,
|
||||
answers: Sequence[str | None] | None = None,
|
||||
ask_count: int = ASK_COUNT,
|
||||
) -> dict[str, Any]:
|
||||
"""Production posterior after `ask_count` probes: scores per cluster
|
||||
representative, eliminated set, still-valid clusters and delivery."""
|
||||
reps = [str(row["time"])[:5] for row in public]
|
||||
scores = {time: float(row.get("score") or 0) for time, row in zip(reps, public)}
|
||||
conflicts = {time: 0 for time in reps}
|
||||
eliminated: set[str] = set()
|
||||
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
|
||||
accuracy = sum(bool(row.get("top_segment_correct")) for row in eligible) / len(eligible)
|
||||
retained = sum(bool(row.get("truth_retained")) for row in eligible) / len(eligible)
|
||||
candidates.append((accuracy, retained, len(eligible), -float(threshold), float(threshold)))
|
||||
if not candidates:
|
||||
return None
|
||||
return max(candidates)[-1]
|
||||
answered += 1
|
||||
scores, conflicts, eliminated = apply_answer(scores, conflicts, eliminated, probe, answer, reps)
|
||||
return finish_state(public, scores, eliminated, answered)
|
||||
|
||||
|
||||
def segment_probe_score(probe: dict[str, Any], chart_rows: Sequence[dict[str, Any]], prefix: str, minute_weights: dict[str, float]) -> float:
|
||||
segments = segment_rows(chart_rows, prefix)
|
||||
segment_by_time = {t: segment["segment_id"] for segment in segments for t in segment["times"]}
|
||||
yes = {str(t)[:5] for item in probe.get("expected_outcomes") or [] if item.get("answer_class") in {"yes", "weak_yes"} for t in item.get("supports") or []}
|
||||
no = {str(t)[:5] for item in probe.get("expected_outcomes") or [] if item.get("answer_class") == "no" for t in item.get("supports") or []}
|
||||
totals: dict[int, float] = defaultdict(float)
|
||||
for stamp in yes | no:
|
||||
if stamp in segment_by_time:
|
||||
totals[segment_by_time[stamp]] += max(minute_weights.get(stamp, 0.0), 0.0)
|
||||
if len(totals) < 2:
|
||||
def finish_state(
|
||||
public: Sequence[dict[str, Any]],
|
||||
scores: dict[str, float],
|
||||
eliminated: set[str],
|
||||
answered: int,
|
||||
) -> dict[str, Any]:
|
||||
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)
|
||||
return {
|
||||
"scores": scores,
|
||||
"eliminated": set(eliminated),
|
||||
"public": posterior,
|
||||
"valid": valid,
|
||||
"delivery": delivery,
|
||||
"answered": answered,
|
||||
}
|
||||
|
||||
|
||||
def minute_weights(state: dict[str, Any], mode: str, *, valid_only: bool = True) -> dict[str, float]:
|
||||
"""Posterior mass per candidate minute (HH:MM).
|
||||
|
||||
raw — cluster score (clipped at 0) spread evenly over its member minutes;
|
||||
percent — proportional percent over clusters, spread evenly over members;
|
||||
uniform — every surviving minute weighs 1.
|
||||
"""
|
||||
rows = state["valid"] if valid_only else [r for r in state["public"] if str(r["time"])[:5] not in state["eliminated"]]
|
||||
scores = state["scores"]
|
||||
cluster_mass: dict[str, float] = {}
|
||||
for row in rows:
|
||||
stamp = str(row["time"])[:5]
|
||||
cluster_mass[stamp] = max(float(scores.get(stamp, row.get("score") or 0)), 0.0)
|
||||
if mode == "percent":
|
||||
total = sum(cluster_mass.values())
|
||||
cluster_mass = {k: (v / total if total > 0 else 0.0) for k, v in cluster_mass.items()}
|
||||
weights: dict[str, float] = {}
|
||||
for row in rows:
|
||||
stamp = str(row["time"])[:5]
|
||||
members = [str(t)[:5] for t in (row.get("cluster_times") or [stamp])]
|
||||
if mode == "uniform":
|
||||
for member in members:
|
||||
weights[member] = 1.0
|
||||
continue
|
||||
share = cluster_mass[stamp] / len(members) if members else 0.0
|
||||
for member in members:
|
||||
weights[member] = weights.get(member, 0.0) + share
|
||||
return weights
|
||||
|
||||
|
||||
def segment_shares(
|
||||
weights: dict[str, float],
|
||||
segments: Sequence[dict[str, Any]],
|
||||
true_time: str,
|
||||
) -> dict[str, Any]:
|
||||
mass = [0.0 for _ in segments]
|
||||
for stamp, weight in weights.items():
|
||||
index = segment_index_of(segments, offset_of(stamp, true_time))
|
||||
if index is not None:
|
||||
mass[index] += float(weight)
|
||||
total = sum(mass)
|
||||
shares = [m / total if total > 0 else 0.0 for m in mass]
|
||||
alive = [i for i, m in enumerate(mass) if m > EPS]
|
||||
truth = segment_index_of(segments, 0)
|
||||
ranked = sorted(alive, key=lambda i: (-shares[i], i))
|
||||
top = ranked[0] if ranked else None
|
||||
return {
|
||||
"shares": [round(s, 6) for s in shares],
|
||||
"alive_segments": len(alive),
|
||||
"top_segment": top,
|
||||
"top_share": round(shares[top], 6) if top is not None else 0.0,
|
||||
"truth_segment": truth,
|
||||
"truth_retained": truth in alive,
|
||||
"top_is_truth": top is not None and top == truth,
|
||||
"truth_share": round(shares[truth], 6) if truth is not None else 0.0,
|
||||
}
|
||||
|
||||
|
||||
def truth_in_delivery(state: dict[str, Any], true_time: str) -> bool:
|
||||
start, end = state["delivery"].get("start"), state["delivery"].get("end")
|
||||
if start is None or end is None:
|
||||
return False
|
||||
span = (clock(end) - clock(start)) % 1440
|
||||
return (clock(true_time) - clock(start)) % 1440 <= span
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# M2 · segment-aware probe ordering
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _entropy(values: Sequence[float]) -> float:
|
||||
total = sum(values)
|
||||
if total <= 0:
|
||||
return 0.0
|
||||
values = sorted(totals.values(), reverse=True)
|
||||
return round(values[0] - values[1], 8)
|
||||
out = 0.0
|
||||
for value in values:
|
||||
if value > 0:
|
||||
p = value / total
|
||||
out -= p * math.log2(p)
|
||||
return out
|
||||
|
||||
|
||||
def reorder_probes_by_segments(probes: Sequence[dict[str, Any]], chart_rows: Sequence[dict[str, Any]], prefix: str, minute_weights: dict[str, float]) -> list[dict[str, Any]]:
|
||||
return sorted(enumerate(probes), key=lambda item: (-segment_probe_score(item[1], chart_rows, prefix, minute_weights), item[0])) and [item[1] for item in sorted(enumerate(probes), key=lambda item: (-segment_probe_score(item[1], chart_rows, prefix, minute_weights), item[0]))]
|
||||
def probe_side(probe: dict[str, Any], stamp: str) -> str:
|
||||
yes, no = outcome_groups(probe)
|
||||
if stamp in yes:
|
||||
return "yes"
|
||||
if stamp in no:
|
||||
return "no"
|
||||
direction = inherit_direction(stamp, yes, no)
|
||||
return {"support": "yes", "conflict": "no"}.get(direction, "neutral")
|
||||
|
||||
|
||||
def strategy_prefixes(domain: str) -> tuple[str, ...]:
|
||||
if domain == "career": return ("D1", "D10")
|
||||
if domain == "relationship": return ("D1", "D9")
|
||||
return VARGA_PREFIXES
|
||||
def segment_information_gain(
|
||||
probe: dict[str, Any],
|
||||
weights: dict[str, float],
|
||||
segments: Sequence[dict[str, Any]],
|
||||
true_time: str,
|
||||
) -> float:
|
||||
"""I(segment ; answer) under the current minute mass.
|
||||
|
||||
A candidate minute on the probe's yes side answers "yes" with probability 1,
|
||||
one on the no side answers "no" with probability 1; a minute the probe says
|
||||
nothing about (neutral) is an uninformative coin (½ / ½), which is what the
|
||||
production card update does with it (delta 0)."""
|
||||
n = len(segments)
|
||||
joint: dict[str, list[float]] = {"yes": [0.0] * n, "no": [0.0] * n}
|
||||
for stamp, weight in weights.items():
|
||||
index = segment_index_of(segments, offset_of(stamp, true_time))
|
||||
if index is None or weight <= 0:
|
||||
continue
|
||||
side = probe_side(probe, stamp)
|
||||
if side == "neutral":
|
||||
joint["yes"][index] += float(weight) / 2
|
||||
joint["no"][index] += float(weight) / 2
|
||||
else:
|
||||
joint[side][index] += float(weight)
|
||||
prior = [sum(joint[a][i] for a in joint) for i in range(n)]
|
||||
total = sum(prior)
|
||||
if total <= 0:
|
||||
return 0.0
|
||||
conditional = 0.0
|
||||
for answer, mass in joint.items():
|
||||
p_answer = sum(mass) / total
|
||||
if p_answer > 0:
|
||||
conditional += p_answer * _entropy(mass)
|
||||
return round(_entropy(prior) - conditional, 6)
|
||||
|
||||
|
||||
def select_probes_by_segment(
|
||||
public: Sequence[dict[str, Any]],
|
||||
pool: Sequence[dict[str, Any]],
|
||||
segments: Sequence[dict[str, Any]],
|
||||
true_time: str,
|
||||
*,
|
||||
ask_count: int = ASK_COUNT,
|
||||
mode: str = "raw",
|
||||
stop_share: float | None = None,
|
||||
answers_override: dict[str, str | None] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Sequential selection: at each step pick the unasked probe with the largest
|
||||
segment information gain under the current posterior, answer it from the
|
||||
true minute (as the production replay does), apply the production card
|
||||
update, repeat. Optional `stop_share` stops as soon as the top segment holds
|
||||
that share of the mass. Returns the final state plus the asked order."""
|
||||
reps = [str(row["time"])[:5] for row in public]
|
||||
scores = {time: float(row.get("score") or 0) for time, row in zip(reps, public)}
|
||||
conflicts = {time: 0 for time in reps}
|
||||
eliminated: set[str] = set()
|
||||
remaining = list(pool)
|
||||
asked: list[dict[str, Any]] = []
|
||||
answered = 0
|
||||
gains: list[float] = []
|
||||
while remaining and len(asked) < ask_count:
|
||||
state = finish_state(public, scores, eliminated, answered)
|
||||
weights = minute_weights(state, mode)
|
||||
if stop_share is not None:
|
||||
share = segment_shares(weights, segments, true_time)
|
||||
if share["alive_segments"] <= 1 or share["top_share"] >= stop_share - EPS:
|
||||
break
|
||||
scored = [(segment_information_gain(p, weights, segments, true_time), -i, i) for i, p in enumerate(remaining)]
|
||||
gain, _neg, index = max(scored)
|
||||
probe = remaining.pop(index)
|
||||
asked.append(probe)
|
||||
gains.append(gain)
|
||||
key = str(probe.get("semantic_key") or probe.get("id") or "")
|
||||
answer = answers_override.get(key, optimal_answer(probe, true_time)) if answers_override else optimal_answer(probe, true_time)
|
||||
if answer is None:
|
||||
continue
|
||||
answered += 1
|
||||
scores, conflicts, eliminated = apply_answer(scores, conflicts, eliminated, probe, answer, reps)
|
||||
state = finish_state(public, scores, eliminated, answered)
|
||||
state["asked"] = asked
|
||||
state["gains"] = gains
|
||||
return state
|
||||
|
||||
|
||||
def production_order_with_stop(
|
||||
public: Sequence[dict[str, Any]],
|
||||
pool: Sequence[dict[str, Any]],
|
||||
segments: Sequence[dict[str, Any]],
|
||||
true_time: str,
|
||||
*,
|
||||
ask_count: int = ASK_COUNT,
|
||||
mode: str = "raw",
|
||||
stop_share: float | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Production probe order (first N of the ranked pool) with the same optional stop rule."""
|
||||
reps = [str(row["time"])[:5] for row in public]
|
||||
scores = {time: float(row.get("score") or 0) for time, row in zip(reps, public)}
|
||||
conflicts = {time: 0 for time in reps}
|
||||
eliminated: set[str] = set()
|
||||
asked: list[dict[str, Any]] = []
|
||||
answered = 0
|
||||
for probe in list(pool)[:ask_count]:
|
||||
if stop_share is not None:
|
||||
state = finish_state(public, scores, eliminated, answered)
|
||||
share = segment_shares(minute_weights(state, mode), segments, true_time)
|
||||
if share["alive_segments"] <= 1 or share["top_share"] >= stop_share - EPS:
|
||||
break
|
||||
asked.append(probe)
|
||||
answer = optimal_answer(probe, true_time)
|
||||
if answer is None:
|
||||
continue
|
||||
answered += 1
|
||||
scores, conflicts, eliminated = apply_answer(scores, conflicts, eliminated, probe, answer, reps)
|
||||
state = finish_state(public, scores, eliminated, answered)
|
||||
state["asked"] = asked
|
||||
return state
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# M1 · threshold / accuracy tables
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def threshold_table(points: Sequence[tuple[float, bool]], thresholds: Sequence[float] = THRESHOLDS) -> list[dict[str, Any]]:
|
||||
"""Full-fit: among cases whose top share ≥ t, how often the top segment is the truth."""
|
||||
out = []
|
||||
for t in thresholds:
|
||||
inside = [ok for share, ok in points if share + EPS >= t]
|
||||
out.append({
|
||||
"threshold": t,
|
||||
"n": len(inside),
|
||||
"accuracy": round(sum(inside) / len(inside), 4) if inside else None,
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
def loo_threshold(
|
||||
points: Sequence[tuple[float, bool]],
|
||||
*,
|
||||
target: float,
|
||||
thresholds: Sequence[float] = THRESHOLDS,
|
||||
min_training: int = MIN_TRAINING_CASES,
|
||||
) -> dict[str, Any]:
|
||||
"""Leave-one-case-out: pick the smallest grid threshold whose *training*
|
||||
accuracy reaches `target` (with at least `min_training` training cases
|
||||
above it), then score the held-out case: labelled 'confident' when its
|
||||
share clears that threshold. Reports how often 'confident' was right."""
|
||||
labelled: list[tuple[float | None, bool, bool]] = [] # (threshold used, confident?, correct?)
|
||||
for hold in range(len(points)):
|
||||
training = [p for i, p in enumerate(points) if i != hold]
|
||||
chosen: float | None = None
|
||||
for t in thresholds:
|
||||
inside = [ok for share, ok in training if share + EPS >= t]
|
||||
if len(inside) >= min_training and sum(inside) / len(inside) + EPS >= target:
|
||||
chosen = t
|
||||
break
|
||||
share, ok = points[hold]
|
||||
confident = chosen is not None and share + EPS >= chosen
|
||||
labelled.append((chosen, confident, ok))
|
||||
confident_rows = [ok for _t, c, ok in labelled if c]
|
||||
thresholds_used = sorted({t for t, _c, _ok in labelled if t is not None})
|
||||
return {
|
||||
"target": target,
|
||||
"confident_n": len(confident_rows),
|
||||
"confident_accuracy": round(sum(confident_rows) / len(confident_rows), 4) if confident_rows else None,
|
||||
"not_confident_n": len(labelled) - len(confident_rows),
|
||||
"no_threshold_folds": sum(1 for t, _c, _ok in labelled if t is None),
|
||||
"thresholds_used": thresholds_used,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# answer perturbation (robustness)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def flipped_answers(
|
||||
probes: Sequence[dict[str, Any]],
|
||||
true_time: str,
|
||||
*,
|
||||
flips: int,
|
||||
seed: str,
|
||||
ask_count: int = ASK_COUNT,
|
||||
) -> list[str | None]:
|
||||
"""Optimal answers with `flips` of the answered probes inverted (yes↔no)."""
|
||||
asked = list(probes)[:ask_count]
|
||||
answers: list[str | None] = [optimal_answer(p, true_time) for p in asked]
|
||||
answerable = [i for i, a in enumerate(answers) if a in {"yes", "no"}]
|
||||
rng = Random(seed)
|
||||
for index in rng.sample(answerable, min(flips, len(answerable))):
|
||||
answers[index] = "no" if answers[index] == "yes" else "yes"
|
||||
return answers
|
||||
|
||||
|
||||
def stable_json(payload: Any) -> str:
|
||||
return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
||||
|
||||
|
||||
__all__ = [name for name in dir() if not name.startswith("__")]
|
||||
|
||||
@@ -1,348 +1,468 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Run the deterministic BUG-1105 varga-resolution M0 replay.
|
||||
"""Varga-resolution research runner (TASK-rectification-varga-resolution-research-20260930).
|
||||
|
||||
The production scorer remains untouched. Scoring candidates are sampled at a
|
||||
fixed two-minute step, while an independent one-minute context scan supplies
|
||||
varga rising-sign segments. ``refresh_probes=False`` is intentional and is
|
||||
recorded in the machine result; an empty probe pool is reported rather than
|
||||
silently replaced with refreshed dasha-boundary probes.
|
||||
Stages (``--stages``, comma list, default all):
|
||||
|
||||
* ``m0`` — Table 1 (signs per window) and Table 2 (chart type inside the
|
||||
six-probe delivery interval) of the brief, reproduced from
|
||||
`varga_resolution_lib`; written to ``--baseline-out``.
|
||||
* ``m1`` — segment shares after the production six probes, raw / percent /
|
||||
uniform mass; full-fit threshold table and leave-one-case-out
|
||||
confidence labels per varga × radius; truth-segment retention.
|
||||
* ``m2`` — production probe order vs segment-information-gain order on the
|
||||
same pool, per target varga; also with a stop rule.
|
||||
* ``m3`` — question-domain strategies (career / relationship / general):
|
||||
joint segments, "no rectification needed" share, accuracy,
|
||||
questions with stop rule.
|
||||
* ``robust`` — 1 / 2 flipped answers (5 seeds), ±7-day shift of day-precision
|
||||
events, LMT-era cases split out.
|
||||
|
||||
Everything is offline; no production file is touched. ``PYTHONHASHSEED=0`` and
|
||||
two runs must be byte-identical (``--json-out`` is written with sorted keys).
|
||||
|
||||
Usage::
|
||||
|
||||
PYTHONHASHSEED=0 python3 scripts/research/varga_resolution_probe.py \
|
||||
--stages m0,m1,m2,m3,robust --radii 10,30,60 \
|
||||
--cache-dir /tmp/varga-cache \
|
||||
--baseline-out docs/research/varga_resolution_baseline_2026_09_30.json \
|
||||
--json-out docs/research/varga_resolution_research_2026_09_30.json
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import statistics
|
||||
import sys
|
||||
from collections import Counter
|
||||
import time
|
||||
from collections import Counter, defaultdict
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable, Sequence
|
||||
from typing import Any, Sequence
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from scripts.research.cluster_width_lib import delivery_from_public # noqa: E402
|
||||
from scripts.research.varga_resolution_lib import ( # noqa: E402
|
||||
RADII,
|
||||
VARGA_PREFIXES,
|
||||
native_case,
|
||||
segment_rows,
|
||||
sign_value,
|
||||
)
|
||||
from scripts.research import varga_resolution_lib as vr # noqa: E402
|
||||
|
||||
HOLDOUT = ROOT / "references" / "real_case_calibration" / "minute_rectification_holdout_v5.json"
|
||||
SCHEMA = "bug-1105-varga-resolution-v1"
|
||||
SCORING_STEP = 2
|
||||
SCAN_STEP = 1
|
||||
REFRESH_PROBES = False
|
||||
BASELINE_JSON = ROOT / "docs" / "research" / "varga_resolution_baseline_2026_09_30.json"
|
||||
REPORT_JSON = ROOT / "docs" / "research" / "varga_resolution_research_2026_09_30.json"
|
||||
TABLE_KEYS: dict[str, tuple[str, ...]] = {
|
||||
"D1": ("D1",), "D9": ("D9",), "D10": ("D10",), "D12": ("D12",), "D1xD9xD10": ("D1", "D9", "D10"),
|
||||
}
|
||||
STOP_SHARES: tuple[float, ...] = (0.7, 0.8)
|
||||
|
||||
|
||||
def load_cases() -> list[dict[str, Any]]:
|
||||
payload = json.loads(HOLDOUT.read_text(encoding="utf-8"))
|
||||
return list(payload.get("cases") or [])
|
||||
def _rate(values: Sequence[bool]) -> float | None:
|
||||
return round(sum(1 for v in values if v) / len(values), 4) if values else None
|
||||
|
||||
|
||||
def hhmm(value: object) -> str:
|
||||
return str(value or "")[:5]
|
||||
def _median(values: Sequence[float | int | None]) -> float | None:
|
||||
clean = [float(v) for v in values if v is not None]
|
||||
return statistics.median(clean) if clean else None
|
||||
|
||||
|
||||
def clock(stamp: str) -> int:
|
||||
hour, minute = stamp[:5].split(":")
|
||||
return int(hour) * 60 + int(minute)
|
||||
# ---------------------------------------------------------------------------
|
||||
# M0
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def interval_offsets(state: dict[str, Any], true_time: str, radius: int) -> list[int]:
|
||||
"""Every minute (step 1) from delivery start to end, as offsets from the true minute."""
|
||||
start, end = state["delivery"].get("start"), state["delivery"].get("end")
|
||||
if start is None or end is None:
|
||||
return []
|
||||
# Forward span from start to end (through midnight when the window straddles
|
||||
# it, e.g. a 00:01–23:59 range around a 23:15 birth), every offset folded
|
||||
# into [-720, 720] and clipped to the search window. Interior minutes that
|
||||
# were eliminated stay inside: this is the range the product shows.
|
||||
span = (vr.clock(end) - vr.clock(start)) % 1440
|
||||
first = vr.offset_of(start, true_time)
|
||||
offsets = []
|
||||
for step in range(span + 1):
|
||||
offset = ((first + step + 720) % 1440) - 720
|
||||
if -radius <= offset <= radius:
|
||||
offsets.append(offset)
|
||||
return sorted(set(offsets))
|
||||
|
||||
|
||||
def in_envelope(stamp: str, start: str | None, end: str | None) -> bool:
|
||||
if not start or not end:
|
||||
return False
|
||||
value, lower, upper = clock(stamp), clock(start), clock(end)
|
||||
return lower <= value <= upper if lower <= upper else value >= lower or value <= upper
|
||||
|
||||
|
||||
def stable_unique(values: Iterable[str]) -> list[str]:
|
||||
return list(dict.fromkeys(str(value)[:5] for value in values if value))
|
||||
|
||||
|
||||
def interval_times(chart_rows: Sequence[dict[str, Any]], delivery: dict[str, Any]) -> list[str]:
|
||||
return [
|
||||
hhmm(row.get("time"))
|
||||
for row in chart_rows
|
||||
if in_envelope(hhmm(row.get("time")), delivery.get("start"), delivery.get("end"))
|
||||
]
|
||||
|
||||
|
||||
def valid_candidate_times(state: dict[str, Any]) -> list[str]:
|
||||
values: list[str] = []
|
||||
for row in state.get("valid") or []:
|
||||
values.extend(hhmm(value) for value in row.get("cluster_times") or [row.get("time")])
|
||||
return stable_unique(values)
|
||||
|
||||
|
||||
def represented_segments(
|
||||
chart_rows: Sequence[dict[str, Any]],
|
||||
prefix: str,
|
||||
times: Iterable[str],
|
||||
) -> list[dict[str, Any]]:
|
||||
wanted = set(stable_unique(times))
|
||||
return [
|
||||
segment for segment in segment_rows(chart_rows, prefix)
|
||||
if wanted.intersection(segment.get("times") or [])
|
||||
]
|
||||
|
||||
|
||||
def represented_values(rows: Sequence[dict[str, Any]], prefix: str, times: Iterable[str]) -> list[Any]:
|
||||
wanted = set(stable_unique(times))
|
||||
return stable_values(sign_value(row, prefix) for row in rows if hhmm(row.get("time")) in wanted)
|
||||
|
||||
|
||||
def stable_values(values: Iterable[Any]) -> list[Any]:
|
||||
result: list[Any] = []
|
||||
for value in values:
|
||||
if value is None or value in result:
|
||||
def table2_row(signs: dict[int, dict[str, int]], offsets: Sequence[int]) -> dict[str, Any]:
|
||||
row: dict[str, Any] = {}
|
||||
for label, vargas in TABLE_KEYS.items():
|
||||
keys = [tuple(signs[o][v] for v in vargas) for o in offsets if o in signs]
|
||||
if not keys:
|
||||
row[label] = {"kinds": 0, "single": False, "le2": False, "mode_is_truth": False}
|
||||
continue
|
||||
result.append(value)
|
||||
return result
|
||||
|
||||
|
||||
def majority(values: Sequence[Any], truth: Any) -> tuple[bool | None, bool]:
|
||||
if not values:
|
||||
return None, False
|
||||
counts = Counter(values)
|
||||
peak = max(counts.values())
|
||||
leaders = {value for value, count in counts.items() if count == peak}
|
||||
return (truth in leaders if len(leaders) == 1 else None), len(leaders) > 1
|
||||
|
||||
|
||||
def window_summary(chart_rows: Sequence[dict[str, Any]]) -> dict[str, Any]:
|
||||
counts: dict[str, int] = {}
|
||||
segments: dict[str, int] = {}
|
||||
for prefix in VARGA_PREFIXES:
|
||||
counts[prefix] = len(stable_values(sign_value(row, prefix) for row in chart_rows))
|
||||
segments[prefix] = len(segment_rows(chart_rows, prefix))
|
||||
combinations = len({tuple(sign_value(row, prefix) for prefix in VARGA_PREFIXES) for row in chart_rows})
|
||||
return {
|
||||
"scan_point_count": len(chart_rows),
|
||||
"sign_counts": counts,
|
||||
"segment_counts": segments,
|
||||
"combination_count": combinations,
|
||||
"d1_single_sign": counts["D1"] == 1,
|
||||
}
|
||||
|
||||
|
||||
def interval_summary(
|
||||
chart_rows: Sequence[dict[str, Any]],
|
||||
state: dict[str, Any],
|
||||
true_time: str,
|
||||
) -> dict[str, Any]:
|
||||
delivery = state.get("delivery") or {}
|
||||
envelope = interval_times(chart_rows, delivery)
|
||||
real = valid_candidate_times(state)
|
||||
values: dict[str, dict[str, Any]] = {}
|
||||
for prefix in VARGA_PREFIXES:
|
||||
truth_row = next((row for row in chart_rows if hhmm(row.get("time")) == true_time), None)
|
||||
truth = sign_value(truth_row or {}, prefix)
|
||||
envelope_segments = represented_segments(chart_rows, prefix, envelope)
|
||||
real_segments = represented_segments(chart_rows, prefix, real)
|
||||
envelope_values = [sign_value(row, prefix) for row in chart_rows if hhmm(row.get("time")) in set(envelope)]
|
||||
exact_values = represented_values(chart_rows, prefix, real)
|
||||
envelope_majority, envelope_tie = majority(envelope_values, truth)
|
||||
exact_majority, exact_tie = majority(exact_values, truth)
|
||||
truth_segment = next(
|
||||
(segment for segment in segment_rows(chart_rows, prefix) if true_time in (segment.get("times") or [])),
|
||||
None,
|
||||
)
|
||||
truth_segment_id = truth_segment.get("segment_id") if truth_segment else None
|
||||
values[prefix] = {
|
||||
"truth_sign": truth,
|
||||
"truth_segment_id": truth_segment_id,
|
||||
"envelope_start": delivery.get("start"),
|
||||
"envelope_end": delivery.get("end"),
|
||||
"envelope_scan_point_count": len(envelope),
|
||||
"envelope_sign_count": len(stable_values(envelope_values)),
|
||||
"envelope_segment_count": len(envelope_segments),
|
||||
"envelope_majority_truth": envelope_majority,
|
||||
"envelope_majority_tie": envelope_tie,
|
||||
"real_valid_candidate_count": len(real),
|
||||
"real_valid_sign_count": len(exact_values),
|
||||
"real_valid_segment_count": len(real_segments),
|
||||
"real_valid_majority_truth": exact_majority,
|
||||
"real_valid_majority_tie": exact_tie,
|
||||
"truth_segment_retained_in_real_set": truth_segment_id is not None and any(
|
||||
segment.get("segment_id") == truth_segment_id for segment in real_segments
|
||||
),
|
||||
counter = Counter(keys) # insertion order = ascending offset; ties go to the earliest sign
|
||||
ranked = counter.most_common()
|
||||
mode_key = ranked[0][0]
|
||||
tied = len(ranked) > 1 and ranked[1][1] == ranked[0][1]
|
||||
truth_key = tuple(signs[0][v] for v in vargas)
|
||||
row[label] = {
|
||||
"kinds": len(counter),
|
||||
"single": len(counter) == 1,
|
||||
"le2": len(counter) <= 2,
|
||||
"mode_is_truth": mode_key == truth_key,
|
||||
"mode_tied": tied,
|
||||
}
|
||||
combo_envelope = {
|
||||
tuple(sign_value(row, prefix) for prefix in VARGA_PREFIXES)
|
||||
for row in chart_rows
|
||||
if hhmm(row.get("time")) in set(envelope)
|
||||
}
|
||||
combo_real = {
|
||||
tuple(sign_value(row, prefix) for prefix in VARGA_PREFIXES)
|
||||
for row in chart_rows
|
||||
if hhmm(row.get("time")) in set(real)
|
||||
}
|
||||
truth_row = next((row for row in chart_rows if hhmm(row.get("time")) == true_time), None)
|
||||
truth_combo = tuple(sign_value(truth_row or {}, prefix) for prefix in VARGA_PREFIXES)
|
||||
envelope_combo_majority, envelope_combo_tie = majority(
|
||||
[tuple(sign_value(row, prefix) for prefix in VARGA_PREFIXES) for row in chart_rows if hhmm(row.get("time")) in set(envelope)],
|
||||
truth_combo,
|
||||
)
|
||||
return {
|
||||
"scoring_candidate_times": real,
|
||||
"interval_envelope": {
|
||||
"start": delivery.get("start"),
|
||||
"end": delivery.get("end"),
|
||||
"scan_point_count": len(envelope),
|
||||
"times": envelope,
|
||||
},
|
||||
"by_varga": values,
|
||||
"combination": {
|
||||
"envelope_count": len(combo_envelope),
|
||||
"real_valid_count": len(combo_real),
|
||||
"envelope_majority_truth": envelope_combo_majority,
|
||||
"envelope_majority_tie": envelope_combo_tie,
|
||||
},
|
||||
}
|
||||
return row
|
||||
|
||||
|
||||
def case_result(case: dict[str, Any], radius: int) -> dict[str, Any]:
|
||||
result = native_case(case, radius, do_reconcile=True)
|
||||
state = result["state"]
|
||||
chart_rows = result["chart_rows"]
|
||||
return {
|
||||
"case_id": str(case.get("case_id") or ""),
|
||||
"radius": radius,
|
||||
"true_time": result["true_time"],
|
||||
"scoring_candidate_count": len(result["rows"]),
|
||||
"scan_point_count": len(chart_rows),
|
||||
"probe_count": len(result["probes"]),
|
||||
"answered_count": int((state.get("result") or {}).get("questions") or 0),
|
||||
"reconciliation": result["reconciliation"],
|
||||
"window": window_summary(chart_rows),
|
||||
"six_question_delivery": interval_summary(chart_rows, state, result["true_time"]),
|
||||
}
|
||||
|
||||
|
||||
def ratio(numerator: int, denominator: int) -> float | None:
|
||||
return round(numerator / denominator, 8) if denominator else None
|
||||
|
||||
|
||||
def aggregate(results: Sequence[dict[str, Any]], radius: int) -> dict[str, Any]:
|
||||
rows = [row for row in results if row["radius"] == radius]
|
||||
n = len(rows)
|
||||
window = {
|
||||
"case_count": n,
|
||||
"d1_single_sign": sum(row["window"]["d1_single_sign"] for row in rows),
|
||||
"mean_sign_counts": {
|
||||
prefix: round(sum(row["window"]["sign_counts"][prefix] for row in rows) / n, 8) if n else None
|
||||
for prefix in VARGA_PREFIXES
|
||||
},
|
||||
"mean_segment_counts": {
|
||||
prefix: round(sum(row["window"]["segment_counts"][prefix] for row in rows) / n, 8) if n else None
|
||||
for prefix in VARGA_PREFIXES
|
||||
},
|
||||
"mean_combination_count": round(sum(row["window"]["combination_count"] for row in rows) / n, 8) if n else None,
|
||||
}
|
||||
by_varga: dict[str, Any] = {}
|
||||
for prefix in VARGA_PREFIXES:
|
||||
items = [row["six_question_delivery"]["by_varga"][prefix] for row in rows]
|
||||
by_varga[prefix] = {
|
||||
"envelope_only_one_sign": sum(item["envelope_sign_count"] == 1 for item in items),
|
||||
"envelope_at_most_two_signs": sum(item["envelope_sign_count"] <= 2 for item in items),
|
||||
"envelope_majority_truth": sum(item["envelope_majority_truth"] is True for item in items),
|
||||
"envelope_majority_ties": sum(item["envelope_majority_tie"] for item in items),
|
||||
"real_truth_segment_retained": sum(item["truth_segment_retained_in_real_set"] for item in items),
|
||||
"real_at_most_two_segments": sum(item["real_valid_segment_count"] <= 2 for item in items),
|
||||
"denominator": n,
|
||||
def stage_m0(replays: dict[tuple[str, int], vr.CaseReplay], signs60: dict[str, dict[int, dict[str, int]]], radii: Sequence[int]) -> dict[str, Any]:
|
||||
table1 = vr.count_table(signs60)
|
||||
per_case: list[dict[str, Any]] = []
|
||||
for (case_id, radius), rep in sorted(replays.items()):
|
||||
state = vr.replay_scores(rep.public, rep.probes, rep.true_time)
|
||||
offsets = interval_offsets(state, rep.true_time, radius)
|
||||
per_case.append({
|
||||
"case_id": case_id, "radius": radius,
|
||||
"start": state["delivery"].get("start"), "end": state["delivery"].get("end"),
|
||||
"width": state["delivery"].get("width"),
|
||||
"truth_in_range": vr.truth_in_delivery(state, rep.true_time),
|
||||
"table2": table2_row(rep.signs, offsets),
|
||||
})
|
||||
table2: dict[str, Any] = {}
|
||||
for radius in radii:
|
||||
rows = [r for r in per_case if r["radius"] == radius]
|
||||
summary: dict[str, Any] = {
|
||||
"n": len(rows),
|
||||
"truth_in_range": sum(1 for r in rows if r["truth_in_range"]),
|
||||
"width_median": _median([r["width"] for r in rows]),
|
||||
}
|
||||
for key in (
|
||||
"envelope_only_one_sign",
|
||||
"envelope_at_most_two_signs",
|
||||
"envelope_majority_truth",
|
||||
"real_truth_segment_retained",
|
||||
"real_at_most_two_segments",
|
||||
):
|
||||
by_varga[prefix][f"{key}_rate"] = ratio(by_varga[prefix][key], n)
|
||||
combo = [row["six_question_delivery"]["combination"] for row in rows]
|
||||
for label in TABLE_KEYS:
|
||||
summary[label] = {
|
||||
"single": sum(1 for r in rows if r["table2"][label]["single"]),
|
||||
"le2": sum(1 for r in rows if r["table2"][label]["le2"]),
|
||||
"mode_is_truth": sum(1 for r in rows if r["table2"][label]["mode_is_truth"]),
|
||||
"mode_tied": sum(1 for r in rows if r["table2"][label]["mode_tied"]),
|
||||
"mean_kinds": round(sum(r["table2"][label]["kinds"] for r in rows) / len(rows), 2) if rows else None,
|
||||
}
|
||||
table2[str(radius)] = summary
|
||||
return {"table1_signs_per_window": table1, "table2_delivery_interval": table2, "per_case": per_case}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# M1
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def share_points(replays: dict[tuple[str, int], vr.CaseReplay], radius: int, vargas: Sequence[str], mode: str,
|
||||
*, cases_filter: set[str] | None = None) -> list[dict[str, Any]]:
|
||||
out = []
|
||||
for (case_id, r), rep in sorted(replays.items()):
|
||||
if r != radius or (cases_filter is not None and case_id not in cases_filter):
|
||||
continue
|
||||
state = vr.replay_scores(rep.public, rep.probes, rep.true_time)
|
||||
share = vr.segment_shares(vr.minute_weights(state, mode), rep.segments(vargas), rep.true_time)
|
||||
out.append({"case_id": case_id, **share, "segments_in_window": len(rep.segments(vargas))})
|
||||
return out
|
||||
|
||||
|
||||
def summarize_shares(points: Sequence[dict[str, Any]]) -> dict[str, Any]:
|
||||
pairs = [(p["top_share"], p["top_is_truth"]) for p in points]
|
||||
return {
|
||||
"radius": radius,
|
||||
"case_count": n,
|
||||
"reconciliation": {
|
||||
"all_zero_diff": all(not row["reconciliation"].get("changed") for row in rows),
|
||||
"case_count": n,
|
||||
"changed_case_count": sum(bool(row["reconciliation"].get("changed")) for row in rows),
|
||||
"denominator": n,
|
||||
},
|
||||
"window": window,
|
||||
"delivery_envelope": {
|
||||
"by_varga": by_varga,
|
||||
"combination_only_one": sum(item["envelope_count"] == 1 for item in combo),
|
||||
"combination_at_most_two": sum(item["envelope_count"] <= 2 for item in combo),
|
||||
"combination_majority_truth": sum(item["envelope_majority_truth"] is True for item in combo),
|
||||
"combination_majority_ties": sum(item["envelope_majority_tie"] for item in combo),
|
||||
"denominator": n,
|
||||
},
|
||||
"n": len(points),
|
||||
"truth_retained": sum(1 for p in points if p["truth_retained"]),
|
||||
"top_is_truth": sum(1 for p in points if p["top_is_truth"]),
|
||||
"alive_le2": sum(1 for p in points if p["alive_segments"] <= 2),
|
||||
"alive_single": sum(1 for p in points if p["alive_segments"] == 1),
|
||||
"top_share_median": _median([p["top_share"] for p in points]),
|
||||
"full_fit": vr.threshold_table(pairs),
|
||||
"loo": [vr.loo_threshold(pairs, target=t) for t in vr.TARGET_ACCURACIES],
|
||||
}
|
||||
|
||||
|
||||
def stage_m1(replays: dict[tuple[str, int], vr.CaseReplay], radii: Sequence[int], lmt_ids: set[str]) -> dict[str, Any]:
|
||||
out: dict[str, Any] = {"by_radius": {}, "lmt_era_case_ids": sorted(lmt_ids)}
|
||||
for radius in radii:
|
||||
block: dict[str, Any] = {}
|
||||
for label, vargas in TABLE_KEYS.items():
|
||||
block[label] = {}
|
||||
for mode in vr.MODES:
|
||||
points = share_points(replays, radius, vargas, mode)
|
||||
summary = summarize_shares(points)
|
||||
non_lmt = [p for p in points if p["case_id"] not in lmt_ids]
|
||||
summary["excluding_lmt_era"] = {
|
||||
"n": len(non_lmt),
|
||||
"truth_retained": sum(1 for p in non_lmt if p["truth_retained"]),
|
||||
"top_is_truth": sum(1 for p in non_lmt if p["top_is_truth"]),
|
||||
}
|
||||
summary["per_case"] = [
|
||||
{k: p[k] for k in ("case_id", "top_share", "top_is_truth", "truth_retained", "alive_segments", "segments_in_window")}
|
||||
for p in points
|
||||
]
|
||||
block[label][mode] = summary
|
||||
out["by_radius"][str(radius)] = block
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# M2
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def order_outcome(rep: vr.CaseReplay, state: dict[str, Any], vargas: Sequence[str], mode: str) -> dict[str, Any]:
|
||||
share = vr.segment_shares(vr.minute_weights(state, mode), rep.segments(vargas), rep.true_time)
|
||||
return {
|
||||
"questions": len(state.get("asked", [])) if "asked" in state else state["answered"],
|
||||
"answered": state["answered"],
|
||||
"truth_retained": share["truth_retained"],
|
||||
"top_is_truth": share["top_is_truth"],
|
||||
"top_share": share["top_share"],
|
||||
"alive_segments": share["alive_segments"],
|
||||
"width": state["delivery"].get("width"),
|
||||
"truth_in_range": vr.truth_in_delivery(state, rep.true_time),
|
||||
}
|
||||
|
||||
|
||||
def stage_m2(replays: dict[tuple[str, int], vr.CaseReplay], radii: Sequence[int], mode: str = "raw") -> dict[str, Any]:
|
||||
targets = {"D9": ("D9",), "D10": ("D10",), "D1xD9xD10": ("D1", "D9", "D10")}
|
||||
out: dict[str, Any] = {"mode": mode, "by_radius": {}}
|
||||
for radius in radii:
|
||||
block: dict[str, Any] = {}
|
||||
for label, vargas in targets.items():
|
||||
rows = []
|
||||
for (case_id, r), rep in sorted(replays.items()):
|
||||
if r != radius:
|
||||
continue
|
||||
segments = rep.segments(vargas)
|
||||
prod = vr.replay_scores(rep.public, rep.probes, rep.true_time)
|
||||
prod["asked"] = list(rep.probes)[:vr.ASK_COUNT]
|
||||
seg = vr.select_probes_by_segment(rep.public, rep.probes, segments, rep.true_time, mode=mode)
|
||||
row: dict[str, Any] = {
|
||||
"case_id": case_id,
|
||||
"pool": len(rep.probes),
|
||||
"production": order_outcome(rep, prod, vargas, mode),
|
||||
"segment_order": order_outcome(rep, seg, vargas, mode),
|
||||
"same_first_six": [p.get("semantic_key") for p in prod["asked"]] == [p.get("semantic_key") for p in seg["asked"]],
|
||||
}
|
||||
for stop in STOP_SHARES:
|
||||
prod_stop = vr.production_order_with_stop(rep.public, rep.probes, segments, rep.true_time, mode=mode, stop_share=stop)
|
||||
seg_stop = vr.select_probes_by_segment(rep.public, rep.probes, segments, rep.true_time, mode=mode, stop_share=stop)
|
||||
row[f"stop_{stop}"] = {
|
||||
"production": order_outcome(rep, prod_stop, vargas, mode),
|
||||
"segment_order": order_outcome(rep, seg_stop, vargas, mode),
|
||||
}
|
||||
rows.append(row)
|
||||
|
||||
def agg(getter) -> dict[str, Any]:
|
||||
items = [getter(r) for r in rows]
|
||||
return {
|
||||
"n": len(items),
|
||||
"mean_questions": round(sum(i["questions"] for i in items) / len(items), 2) if items else None,
|
||||
"truth_retained": sum(1 for i in items if i["truth_retained"]),
|
||||
"top_is_truth": sum(1 for i in items if i["top_is_truth"]),
|
||||
"alive_le2": sum(1 for i in items if i["alive_segments"] <= 2),
|
||||
"top_share_median": _median([i["top_share"] for i in items]),
|
||||
"truth_in_range": sum(1 for i in items if i["truth_in_range"]),
|
||||
"width_median": _median([i["width"] for i in items]),
|
||||
}
|
||||
|
||||
summary: dict[str, Any] = {
|
||||
"six": {"production": agg(lambda r: r["production"]), "segment_order": agg(lambda r: r["segment_order"])},
|
||||
"same_first_six_cases": sum(1 for r in rows if r["same_first_six"]),
|
||||
"pool_median": _median([r["pool"] for r in rows]),
|
||||
}
|
||||
for stop in STOP_SHARES:
|
||||
summary[f"stop_{stop}"] = {
|
||||
"production": agg(lambda r, s=stop: r[f"stop_{s}"]["production"]),
|
||||
"segment_order": agg(lambda r, s=stop: r[f"stop_{s}"]["segment_order"]),
|
||||
}
|
||||
summary["per_case"] = rows
|
||||
block[label] = summary
|
||||
out["by_radius"][str(radius)] = block
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# M3
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def stage_m3(replays: dict[tuple[str, int], vr.CaseReplay], signs60: dict[str, dict[int, dict[str, int]]],
|
||||
radii: Sequence[int], mode: str = "raw") -> dict[str, Any]:
|
||||
out: dict[str, Any] = {"mode": mode, "strategies": {k: list(v) for k, v in vr.STRATEGIES.items()}, "by_radius": {}}
|
||||
for radius in list(radii) + [r for r in vr.COUNT_RADII if r not in radii]:
|
||||
block: dict[str, Any] = {}
|
||||
for name, vargas in vr.STRATEGIES.items():
|
||||
# "no rectification needed": the target vargas keep one joint sign across the whole window
|
||||
no_need = sum(1 for signs in signs60.values() if vr.distinct_signs(signs, radius, vargas) == 1)
|
||||
entry: dict[str, Any] = {"no_rectification_needed": no_need, "n_cases": len(signs60)}
|
||||
if radius in radii:
|
||||
rows = []
|
||||
for (case_id, r), rep in sorted(replays.items()):
|
||||
if r != radius:
|
||||
continue
|
||||
segments = rep.segments(vargas)
|
||||
prod = vr.replay_scores(rep.public, rep.probes, rep.true_time)
|
||||
prod["asked"] = list(rep.probes)[:vr.ASK_COUNT]
|
||||
seg = vr.select_probes_by_segment(rep.public, rep.probes, segments, rep.true_time, mode=mode)
|
||||
row = {
|
||||
"case_id": case_id,
|
||||
"segments_in_window": len(segments),
|
||||
"production": order_outcome(rep, prod, vargas, mode),
|
||||
"segment_order": order_outcome(rep, seg, vargas, mode),
|
||||
}
|
||||
for stop in STOP_SHARES:
|
||||
seg_stop = vr.select_probes_by_segment(rep.public, rep.probes, segments, rep.true_time, mode=mode, stop_share=stop)
|
||||
row[f"stop_{stop}"] = order_outcome(rep, seg_stop, vargas, mode)
|
||||
rows.append(row)
|
||||
|
||||
def agg(items: Sequence[dict[str, Any]]) -> dict[str, Any]:
|
||||
return {
|
||||
"n": len(items),
|
||||
"mean_questions": round(sum(i["questions"] for i in items) / len(items), 2) if items else None,
|
||||
"truth_retained": sum(1 for i in items if i["truth_retained"]),
|
||||
"top_is_truth": sum(1 for i in items if i["top_is_truth"]),
|
||||
"alive_le2": sum(1 for i in items if i["alive_segments"] <= 2),
|
||||
"alive_single": sum(1 for i in items if i["alive_segments"] == 1),
|
||||
"top_share_median": _median([i["top_share"] for i in items]),
|
||||
}
|
||||
|
||||
entry.update({
|
||||
"segments_in_window_mean": round(sum(r["segments_in_window"] for r in rows) / len(rows), 2) if rows else None,
|
||||
"production_six": agg([r["production"] for r in rows]),
|
||||
"segment_order_six": agg([r["segment_order"] for r in rows]),
|
||||
**{f"segment_order_stop_{s}": agg([r[f"stop_{s}"] for r in rows]) for s in STOP_SHARES},
|
||||
"per_case": rows,
|
||||
})
|
||||
block[name] = entry
|
||||
out["by_radius"][str(radius)] = block
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# robustness
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def stage_robust(replays: dict[tuple[str, int], vr.CaseReplay], radii: Sequence[int], lmt_ids: set[str], mode: str = "raw") -> dict[str, Any]:
|
||||
targets = {"D9": ("D9",), "D10": ("D10",)}
|
||||
out: dict[str, Any] = {"mode": mode, "by_radius": {}}
|
||||
for radius in radii:
|
||||
block: dict[str, Any] = {}
|
||||
reps = [(cid, rep) for (cid, r), rep in sorted(replays.items()) if r == radius]
|
||||
for label, vargas in targets.items():
|
||||
entry: dict[str, Any] = {}
|
||||
base = []
|
||||
for cid, rep in reps:
|
||||
state = vr.replay_scores(rep.public, rep.probes, rep.true_time)
|
||||
base.append((cid, vr.segment_shares(vr.minute_weights(state, mode), rep.segments(vargas), rep.true_time)))
|
||||
entry["baseline"] = {
|
||||
"n": len(base),
|
||||
"truth_retained": sum(1 for _c, s in base if s["truth_retained"]),
|
||||
"top_is_truth": sum(1 for _c, s in base if s["top_is_truth"]),
|
||||
"lmt_era": {
|
||||
"n": sum(1 for c, _s in base if c in lmt_ids),
|
||||
"truth_retained": sum(1 for c, s in base if c in lmt_ids and s["truth_retained"]),
|
||||
"top_is_truth": sum(1 for c, s in base if c in lmt_ids and s["top_is_truth"]),
|
||||
},
|
||||
}
|
||||
for flips in (1, 2):
|
||||
retained: list[bool] = []
|
||||
correct: list[bool] = []
|
||||
for seed in range(vr.FLIP_SEEDS):
|
||||
for cid, rep in reps:
|
||||
answers = vr.flipped_answers(rep.probes, rep.true_time, flips=flips, seed=f"{cid}:{radius}:{flips}:{seed}")
|
||||
state = vr.replay_scores(rep.public, rep.probes, rep.true_time, answers=answers)
|
||||
share = vr.segment_shares(vr.minute_weights(state, mode), rep.segments(vargas), rep.true_time)
|
||||
retained.append(share["truth_retained"])
|
||||
correct.append(share["top_is_truth"])
|
||||
entry[f"flip_{flips}"] = {
|
||||
"replays": len(retained),
|
||||
"truth_retained_rate": _rate(retained),
|
||||
"top_is_truth_rate": _rate(correct),
|
||||
}
|
||||
shifted_retained: list[bool] = []
|
||||
shifted_correct: list[bool] = []
|
||||
for cid, rep in reps:
|
||||
rows, probes, public = rep.shifted(vr.DAY_SHIFT, seed=f"{cid}:{radius}:shift")
|
||||
state = vr.replay_scores(public, probes, rep.true_time)
|
||||
share = vr.segment_shares(vr.minute_weights(state, mode), rep.segments(vargas), rep.true_time)
|
||||
shifted_retained.append(share["truth_retained"])
|
||||
shifted_correct.append(share["top_is_truth"])
|
||||
entry["shift_7_days"] = {
|
||||
"n": len(shifted_retained),
|
||||
"truth_retained": sum(shifted_retained),
|
||||
"top_is_truth": sum(shifted_correct),
|
||||
}
|
||||
block[label] = entry
|
||||
out["by_radius"][str(radius)] = block
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# main
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def build_replays(cases: Sequence[dict[str, Any]], radii: Sequence[int], cache_dir: Path | None, log) -> dict[tuple[str, int], vr.CaseReplay]:
|
||||
replays: dict[tuple[str, int], vr.CaseReplay] = {}
|
||||
for case in cases:
|
||||
for radius in radii:
|
||||
started = time.perf_counter()
|
||||
replays[(str(case["case_id"]), radius)] = vr.CaseReplay(case, radius, cache_dir=cache_dir)
|
||||
log(f"{case['case_id']} ±{radius} ready in {time.perf_counter() - started:.1f}s")
|
||||
return replays
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--radii", default=",".join(str(value) for value in RADII))
|
||||
parser.add_argument("--vargas", default=",".join(VARGA_PREFIXES))
|
||||
parser.add_argument("--dataset", default=str(vr.HOLDOUT_V5))
|
||||
parser.add_argument("--stages", default="m0,m1,m2,m3,robust")
|
||||
parser.add_argument("--radii", default=",".join(str(r) for r in vr.RADII))
|
||||
parser.add_argument("--vargas", default=",".join(vr.VARGAS))
|
||||
parser.add_argument("--limit", type=int, default=0)
|
||||
parser.add_argument("--json-out", required=True)
|
||||
parser.add_argument("--cache-dir", default="")
|
||||
parser.add_argument("--baseline-out", default=str(BASELINE_JSON))
|
||||
parser.add_argument("--json-out", default=str(REPORT_JSON))
|
||||
parser.add_argument("--quiet", action="store_true")
|
||||
args = parser.parse_args()
|
||||
radii = tuple(int(value) for value in str(args.radii).split(",") if value.strip())
|
||||
vargas = tuple(value.strip() for value in str(args.vargas).split(",") if value.strip())
|
||||
cases = load_cases()
|
||||
stages = {s.strip() for s in args.stages.split(",") if s.strip()}
|
||||
radii = tuple(int(r) for r in args.radii.split(",") if r.strip())
|
||||
cache_dir = Path(args.cache_dir) if args.cache_dir else None
|
||||
log = (lambda *_a, **_k: None) if args.quiet else (lambda msg: print(msg, flush=True))
|
||||
|
||||
started = time.perf_counter()
|
||||
cases = vr.load_cases(Path(args.dataset))
|
||||
if args.limit:
|
||||
cases = cases[: args.limit]
|
||||
results: list[dict[str, Any]] = []
|
||||
errors: list[dict[str, str | int]] = []
|
||||
for case in cases:
|
||||
for radius in radii:
|
||||
label = f"{case.get('case_id')} ±{radius}"
|
||||
try:
|
||||
row = case_result(case, radius)
|
||||
results.append(row)
|
||||
print(
|
||||
f"{label} score={row['scoring_candidate_count']} scan={row['scan_point_count']} "
|
||||
f"probes={row['probe_count']} reconcile_changed={bool(row['reconciliation'].get('changed'))}",
|
||||
flush=True,
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
errors.append({"case_id": str(case.get("case_id") or ""), "radius": radius, "error": f"{type(exc).__name__}: {exc}"})
|
||||
print(f"{label} ERROR {type(exc).__name__}: {exc}", flush=True)
|
||||
payload = {
|
||||
"schema": SCHEMA,
|
||||
"metadata": {
|
||||
"holdout": str(HOLDOUT.relative_to(ROOT)).replace("\\", "/"),
|
||||
"case_count_requested": len(cases),
|
||||
"case_count_completed": len({row["case_id"] for row in results}),
|
||||
"radii": list(radii),
|
||||
"vargas": list(vargas),
|
||||
"ayanamsa": "raman",
|
||||
"node_mode": "mean",
|
||||
"scoring_candidate_step_minutes": SCORING_STEP,
|
||||
"segment_scan_step_minutes": SCAN_STEP,
|
||||
"refresh_probes": REFRESH_PROBES,
|
||||
"replay_probe_source": "existing event_probes.discriminating_event_probes; no refresh",
|
||||
"questions_requested": 6,
|
||||
"questions_answered_is_recorded_per_case": True,
|
||||
"real_valid_candidate_set": "union of cluster_times from still_valid_public after replay; scoring-grid candidates only",
|
||||
"interval_envelope": "unionStillValidRange equivalent: min/max clock edges over the real valid candidate clusters; envelope is not the real set",
|
||||
"segment_definition": "maximal contiguous one-minute scan run of equal D1/D9/D10 sign; repeated non-contiguous signs retain separate IDs; 23:59->00:00 is contiguous",
|
||||
"aggregation_denominator": "77 cases per radius when full run completes; empty coverage and ties are separate counts",
|
||||
"reconciliation_scope": "every case and radius, scoring grid only; single-case zero-diff is smoke evidence, not full-set evidence",
|
||||
"deterministic_json": True,
|
||||
},
|
||||
"aggregates": [aggregate(results, radius) for radius in radii],
|
||||
"results": results,
|
||||
"errors": errors,
|
||||
lmt_ids = {str(c["case_id"]) for c in cases if vr.is_lmt_era(c)}
|
||||
signs60 = {str(c["case_id"]): vr.signs_per_minute(c, max(vr.COUNT_RADII), cache_dir=cache_dir) for c in cases}
|
||||
log(f"signs ready for {len(signs60)} cases")
|
||||
replays = build_replays(cases, radii, cache_dir, log)
|
||||
|
||||
meta = {
|
||||
"dataset": Path(args.dataset).name,
|
||||
"case_count": len(cases),
|
||||
"radii": list(radii),
|
||||
"vargas": list(vr.VARGAS),
|
||||
"ayanamsa": vr.AYANAMSA,
|
||||
"node_mode": vr.NODE_MODE,
|
||||
"minute_step_candidates": vr.MINUTE_STEP,
|
||||
"minute_step_signs": 1,
|
||||
"ask_count": vr.ASK_COUNT,
|
||||
"separation_lead": vr.SEPARATION_LEAD,
|
||||
"lmt_era_before_year": vr.LMT_ERA_BEFORE_YEAR,
|
||||
"open_set_not_blind": True,
|
||||
}
|
||||
out = Path(args.json_out)
|
||||
out.parent.mkdir(parents=True, exist_ok=True)
|
||||
out.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
print(json.dumps(payload["aggregates"], ensure_ascii=False, indent=2, sort_keys=True), flush=True)
|
||||
return 0 if not errors and all(item["reconciliation"]["all_zero_diff"] for item in payload["aggregates"]) else 1
|
||||
if "m0" in stages:
|
||||
baseline = {**meta, **stage_m0(replays, signs60, radii)}
|
||||
Path(args.baseline_out).write_text(vr.stable_json(baseline), encoding="utf-8")
|
||||
log(f"m0 written to {args.baseline_out}")
|
||||
report: dict[str, Any] = {**meta, "stages": sorted(stages - {"m0"})}
|
||||
if "m1" in stages:
|
||||
report["m1"] = stage_m1(replays, radii, lmt_ids)
|
||||
log("m1 done")
|
||||
if "m2" in stages:
|
||||
report["m2"] = stage_m2(replays, radii)
|
||||
log("m2 done")
|
||||
if "m3" in stages:
|
||||
report["m3"] = stage_m3(replays, signs60, radii)
|
||||
log("m3 done")
|
||||
if "robust" in stages:
|
||||
report["robust"] = stage_robust(replays, radii, lmt_ids)
|
||||
log("robust done")
|
||||
if stages - {"m0"}:
|
||||
# No wall-clock value goes into the JSON so two runs stay byte-identical.
|
||||
Path(args.json_out).write_text(vr.stable_json(report), encoding="utf-8")
|
||||
log(f"report written to {args.json_out}")
|
||||
log(f"elapsed {time.perf_counter() - started:.0f}s")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user