Jyotisha/scripts/rectification/dasha_transition_proximity.py

"""Deterministic dasha-transition proximity scoring for day/month events.

Birth-time drift of about 1 minute moves Vimshottari/Narayana transition
dates by a few days. A dated event near a candidate's AD/PD change is a
bounded auxiliary signal, never larger than one day-level event body.
"""

from __future__ import annotations

from collections.abc import Callable, Sequence
from datetime import date, datetime
from typing import Any

from scripts.rectification.event_probes import _narayana_start_dates, _vim_start_dates

PROXIMITY_WINDOW_DAYS = 45
DAY_KERNEL_DAYS = 15
MONTH_KERNEL_DAYS = 45
DAY_MAX_POINTS = 1.0
MONTH_MAX_POINTS = 0.35
VIM_SHARE = 0.6
NARAYANA_SHARE = 0.4


def representative_event_date(event: dict[str, Any]) -> date | None:
    precision = str(event.get("precision") or "")
    if precision not in {"day", "month"}:
        return None
    raw_start = event.get("date_start") or event.get("date")
    raw_end = event.get("date_end") or raw_start
    try:
        start = date.fromisoformat(str(raw_start)[:10])
        end = date.fromisoformat(str(raw_end)[:10])
    except ValueError:
        return None
    if precision == "day" or start == end:
        return start
    mid_day = min(15, end.day)
    try:
        return start.replace(day=mid_day)
    except ValueError:
        return start


def _nearest_delta(starts: Sequence[date], event_date: date) -> tuple[date | None, int | None]:
    eligible = [
        item for item in starts
        if abs((item - event_date).days) <= PROXIMITY_WINDOW_DAYS
    ]
    if not eligible:
        return None, None
    nearest = min(eligible, key=lambda item: (abs((item - event_date).days), item.toordinal()))
    return nearest, abs((nearest - event_date).days)


def _kernel(delta_days: int | None, width: float) -> float:
    if delta_days is None or width <= 0:
        return 0.0
    return max(0.0, 1.0 - (delta_days / width))


def score_transition_proximity(
    *,
    event_date: date,
    precision: str,
    vim_starts: Sequence[date],
    narayana_starts: Sequence[date] | None = None,
    vim_pd_starts: Sequence[date] | None = None,
) -> dict[str, Any]:
    if precision not in {"day", "month"}:
        return {
            "points": 0.0,
            "rule_ids": [],
            "nearest_vim_delta_days": None,
            "nearest_narayana_delta_days": None,
        }
    kernel_width = float(DAY_KERNEL_DAYS if precision == "day" else MONTH_KERNEL_DAYS)
    cap = DAY_MAX_POINTS if precision == "day" else MONTH_MAX_POINTS
    ad_starts = list(vim_starts)
    pd_starts = list(vim_pd_starts or ())
    ad_date, ad_delta = _nearest_delta(ad_starts, event_date)
    pd_date, pd_delta = _nearest_delta(pd_starts, event_date)
    if pd_delta is not None and (ad_delta is None or pd_delta < ad_delta):
        vim_delta = pd_delta
        vim_kind = "pd"
        vim_date = pd_date
    else:
        vim_delta = ad_delta
        vim_kind = "ad"
        vim_date = ad_date
    _, narayana_delta = _nearest_delta(list(narayana_starts or ()), event_date)
    vim_kernel = _kernel(vim_delta, kernel_width)
    narayana_kernel = _kernel(narayana_delta, kernel_width)
    points = round(cap * (VIM_SHARE * vim_kernel + NARAYANA_SHARE * narayana_kernel), 4)
    rules: list[str] = []
    if vim_kernel > 0:
        rules.append(f"vim_transition_proximity_{vim_kind}")
    if narayana_kernel > 0:
        rules.append("narayana_transition_proximity_ad")
    return {
        "points": points,
        "rule_ids": rules,
        "nearest_vim_delta_days": vim_delta,
        "nearest_narayana_delta_days": narayana_delta,
        "nearest_vim_date": vim_date,
    }


def _context_time(context: dict[str, Any]) -> str | None:
    feature = context.get("feature") if isinstance(context.get("feature"), dict) else {}
    raw = feature.get("time")
    if isinstance(raw, str) and len(raw) >= 5:
        return raw[:5]
    at = context.get("candidate_at")
    if hasattr(at, "strftime"):
        return at.strftime("%H:%M")
    return None


def merge_transition_proximity(
    matrix: dict[str, dict[str, dict[str, Any]]],
    events: Sequence[dict[str, Any]],
    static_contexts: Sequence[dict[str, Any]],
    birth_date: str,
    *,
    public_technique_layers: Callable[[str, Sequence[str]], list[str]],
) -> None:
    by_time = {
        time: context
        for context in static_contexts
        if isinstance(context, dict) and (time := _context_time(context))
    }
    # BUG-1324: one unfiltered dasha walk per candidate, then filter by event year.
    # Released when this function returns. Not a module-level cache.
    vim_cache: dict[tuple[Any, ...], list[date]] = {}
    pd_cache: dict[tuple[Any, ...], list[date]] = {}
    narayana_cache: dict[tuple[Any, ...], list[date] | None] = {}
    for event in events:
        if not isinstance(event, dict):
            continue
        event_id = str(event.get("id") or "")
        cells = matrix.get(event_id)
        if not event_id or not isinstance(cells, dict):
            continue
        event_date = representative_event_date(event)
        if event_date is None:
            continue
        precision = str(event.get("precision") or "")
        lo, hi = event_date.year - 1, event_date.year + 1
        for time, cell in cells.items():
            context = by_time.get(str(time)[:5])
            if not isinstance(cell, dict) or not isinstance(context, dict):
                continue
            moon = (context.get("planet_longitudes") or {}).get("Moon")
            if not isinstance(moon, (int, float)):
                continue
            # Native contexts retain the local date even when the window crosses midnight.
            # Legacy time-only contexts still use the request's date.
            candidate_at = context.get("candidate_at")
            candidate_date = candidate_at.date().isoformat() if isinstance(candidate_at, datetime) else birth_date
            moon_key = round(float(moon), 6)
            vim_key = (candidate_date, moon_key)
            if vim_key not in vim_cache:
                vim_cache[vim_key] = _vim_start_dates(candidate_date, float(moon), 1, 9999)
                pd_cache[vim_key] = _vim_start_dates(
                    candidate_date,
                    float(moon),
                    1,
                    9999,
                    include_pratyantar=True,
                )
            planets = context.get("planet_longitudes") or {}
            asc = context.get("ascendant_index")
            narayana_key = (
                candidate_date,
                int(asc) if isinstance(asc, int) else None,
                moon_key,
            )
            if narayana_key not in narayana_cache:
                narayana_cache[narayana_key] = (
                    _narayana_start_dates(int(asc), planets, candidate_date, 1, 9999)
                    if isinstance(asc, int) and isinstance(planets, dict)
                    else None
                )
            ad_starts = [item for item in vim_cache[vim_key] if lo <= item.year <= hi]
            ad_set = set(ad_starts)
            pd_only = [item for item in pd_cache[vim_key] if lo <= item.year <= hi and item not in ad_set]
            narayana_starts = narayana_cache[narayana_key]
            if narayana_starts is not None:
                narayana_starts = [item for item in narayana_starts if lo <= item.year <= hi]
            scored = score_transition_proximity(
                event_date=event_date,
                precision=precision,
                vim_starts=ad_starts,
                vim_pd_starts=pd_only,
                narayana_starts=narayana_starts or [],
            )
            if scored["points"] <= 0 and not scored["rule_ids"]:
                continue
            cell["points"] = round(float(cell.get("points") or 0) + float(scored["points"]), 4)
            cell["rule_ids"] = sorted({
                *list(cell.get("rule_ids") or []),
                *scored["rule_ids"],
            })
            domain = str(event.get("domain") or cell.get("domain") or "")
            cell["technique_layers"] = public_technique_layers(domain, cell["rule_ids"])