Files
Jyotisha/scripts/research/varga_resolution_lib.py
T

635 lines
26 KiB
Python

#!/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("__")]