#!/usr/bin/env python3 """Varga-resolution research library (TASK-rectification-varga-resolution-research-20260930). 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 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 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.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 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 # --------------------------------------------------------------------------- # 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 clock(value: str) -> int: hours, minutes = str(value)[:5].split(":") return int(hours) * 60 + int(minutes) 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 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 is_lmt_era(case: dict[str, Any]) -> bool: return int(str(case["birth"]["date"])[:4]) < LMT_ERA_BEFORE_YEAR # --------------------------------------------------------------------------- # 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 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)) } 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 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 answered += 1 scores, conflicts, eliminated = apply_answer(scores, conflicts, eliminated, probe, answer, reps) return finish_state(public, scores, eliminated, answered) 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 out = 0.0 for value in values: if value > 0: p = value / total out -= p * math.log2(p) return out 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 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("__")]