"""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 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)) } 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 vim_key = (birth_date, round(float(moon), 6), lo, hi) if vim_key not in vim_cache: vim_cache[vim_key] = _vim_start_dates(birth_date, float(moon), lo, hi) pd_cache[vim_key] = _vim_start_dates( birth_date, float(moon), lo, hi, include_pratyantar=True, ) planets = context.get("planet_longitudes") or {} asc = context.get("ascendant_index") narayana_key = ( birth_date, int(asc) if isinstance(asc, int) else None, lo, hi, round(float(moon), 6), ) if narayana_key not in narayana_cache: narayana_cache[narayana_key] = ( _narayana_start_dates(int(asc), planets, birth_date, lo, hi) if isinstance(asc, int) and isinstance(planets, dict) else None ) ad_starts = vim_cache[vim_key] ad_set = set(ad_starts) pd_only = [item for item in pd_cache[vim_key] if item not in ad_set] scored = score_transition_proximity( event_date=event_date, precision=precision, vim_starts=ad_starts, vim_pd_starts=pd_only, narayana_starts=narayana_cache[narayana_key] 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"])