Jyotisha/scripts/research/offline_research_20260926_lib.py

"""Pure helpers for the 2026-09-26 offline rectification research (R1/R2/R3).

Offline only. Nothing here is imported by production code, and nothing here
changes scoring, thresholds, probe gates or Skill text.
"""

from __future__ import annotations

import statistics
from itertools import combinations
from random import Random
from typing import Any, Iterable, Sequence

#: Vimshottari lords in nakshatra order starting at Ashwini, and their years.
VIM_LORDS: tuple[str, ...] = (
    "Ketu", "Venus", "Sun", "Moon", "Mars", "Rahu", "Jupiter", "Saturn", "Mercury",
)
VIM_YEARS: dict[str, float] = {
    "Ketu": 7.0, "Venus": 20.0, "Sun": 6.0, "Moon": 10.0, "Mars": 7.0,
    "Rahu": 18.0, "Jupiter": 16.0, "Saturn": 19.0, "Mercury": 17.0,
}
NAKSHATRA_SPAN = 360.0 / 27.0
DAYS_PER_YEAR = 365.25  # scripts/dasha_analyzer.py build_dasha_timeline
FLIP_SEED = 20260926
PERCENTILES = (0, 10, 25, 50, 75, 90, 100)


def nakshatra_lord(moon_longitude: float) -> str:
    index = int((float(moon_longitude) % 360.0) // NAKSHATRA_SPAN)
    return VIM_LORDS[index % 9]


def shift_days_per_minute(moon_longitude: float, moon_speed_deg_per_day: float) -> float:
    """Days every Vimshottari boundary moves when birth time moves one minute.

    The balance of the first dasha is (1 - progress) * years(lord). Every later
    MD/AD/PD start is birth + balance + fixed durations, so all of them move by
    the same amount: d(balance)/d(moon) * d(moon)/d(minute).
    """
    years = VIM_YEARS[nakshatra_lord(moon_longitude)]
    deg_per_minute = abs(float(moon_speed_deg_per_day)) / 1440.0
    return deg_per_minute / NAKSHATRA_SPAN * years * DAYS_PER_YEAR


def percentile(values: Sequence[float], q: float) -> float | None:
    """Nearest-rank percentile on a sorted copy (q in 0..100)."""
    items = sorted(float(item) for item in values)
    if not items:
        return None
    if q <= 0:
        return items[0]
    if q >= 100:
        return items[-1]
    rank = int(round(q / 100.0 * (len(items) - 1)))
    return items[rank]


def distribution(values: Sequence[float], digits: int = 2) -> dict[str, Any]:
    items = [float(item) for item in values]
    if not items:
        return {"n": 0}
    row: dict[str, Any] = {"n": len(items), "mean": round(statistics.fmean(items), digits)}
    for q in PERCENTILES:
        row[f"p{q}"] = round(float(percentile(items, q)), digits)
    return row


def minutes_for_gap(days: float, shift_per_minute: float) -> float | None:
    if shift_per_minute <= 0:
        return None
    return float(days) / float(shift_per_minute)


def answerable_indices(answers: Sequence[str | None]) -> list[int]:
    return [index for index, answer in enumerate(answers) if answer in {"yes", "no"}]


def flip(answer: str | None) -> str | None:
    return {"yes": "no", "no": "yes"}.get(str(answer), answer)


def flipped_answers(answers: Sequence[str | None], indices: Iterable[int]) -> list[str | None]:
    chosen = set(indices)
    return [flip(answer) if index in chosen else answer for index, answer in enumerate(answers)]


def sample_flip_sets(
    answers: Sequence[str | None],
    k: int,
    repeats: int,
    *,
    case_id: str,
    radius: int,
    seed: int = FLIP_SEED,
) -> list[tuple[int, ...]]:
    """Fixed-seed random flip sets: `repeats` draws of k answered questions.

    The RNG key includes case, radius, k and repeat index so every draw is
    reproducible on any machine (string seeding is stable across processes).
    """
    pool = answerable_indices(answers)
    if k <= 0 or len(pool) < k:
        return []
    rows: list[tuple[int, ...]] = []
    for repeat in range(int(repeats)):
        rng = Random(f"{seed}:{case_id}:{radius}:{k}:{repeat}")
        rows.append(tuple(sorted(rng.sample(pool, k))))
    return rows


def exhaustive_flip_sets(answers: Sequence[str | None], k: int) -> list[tuple[int, ...]]:
    pool = answerable_indices(answers)
    if k <= 0 or len(pool) < k:
        return []
    return [tuple(item) for item in combinations(pool, k)]


def rate(rows: Sequence[dict[str, Any]], key: str) -> float | None:
    if not rows:
        return None
    return round(sum(1 for row in rows if row.get(key)) / len(rows), 4)


def median_of(rows: Sequence[dict[str, Any]], key: str) -> float | None:
    values = [row[key] for row in rows if row.get(key) is not None]
    return statistics.median(values) if values else None


def all_pairs(reps: Sequence[dict[str, Any]], clock_of) -> list[tuple[dict[str, Any], dict[str, Any]]]:
    """Every unordered pair of representatives, sorted by clock (research variant)."""
    ordered = sorted(reps, key=clock_of)
    return [(ordered[i], ordered[j]) for i in range(len(ordered)) for j in range(i + 1, len(ordered))]


def zip_misaligned(left: Sequence[Any], right: Sequence[Any], slack_days: int = 3) -> bool:
    """True when `_boundary_windows`' positional zip pairs different boundaries.

    For Vimshottari, two candidates' boundary lists are the same sequence
    shifted by one constant number of days (see `shift_days_per_minute`). When
    one list gains or loses an element at the lower year edge, zip() pairs
    boundary i with boundary i+1 of the other candidate and the signed
    differences stop being constant. `slack_days` absorbs date rounding.
    """
    diffs = [(two - one).days for one, two in zip(left, right)]
    if not diffs:
        return False
    return max(diffs) - min(diffs) > slack_days