jesse-ux
b85c4a686a
Carry explicit local date intervals instead of inferring the day from clock order. Cluster width, delivery, adoption, and reports keep the actual civil date; adopted date is stored separately from the reported birth_date. Algorithm identity is scoring-9 / spec-v5. Scoring weights, confirmation thresholds, and Skill version are unchanged. Isolated Linux final-3 gates passed; four pre-existing Python failures remain. This is not a production release.
811 lines
32 KiB
Python
811 lines
32 KiB
Python
# /// script
|
|
# requires-python = ">=3.11"
|
|
# dependencies = []
|
|
# ///
|
|
# ─── How to run ───
|
|
# PYTHONPATH=scripts .venv/bin/python scripts/active_rectification_event_engine.py
|
|
"""Local chart and dual-Dasha computation for birth-time event scoring."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
from datetime import date, datetime, time, timedelta
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from collections.abc import Sequence
|
|
from typing import Any, Final, assert_never
|
|
|
|
from scripts.active_rectification_events import (
|
|
CandidateEvidence,
|
|
CandidateResult,
|
|
CandidateScoreRow,
|
|
EventDomain,
|
|
LifeEvent,
|
|
RectificationEventRequest,
|
|
adjudicate_candidate_rows,
|
|
precision_weight,
|
|
)
|
|
|
|
SCRIPTS: Final = Path(__file__).resolve().parent
|
|
if str(SCRIPTS) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
import dasha_analyzer # noqa: E402
|
|
import domain_calculation_service # noqa: E402
|
|
import ashtakavarga # noqa: E402
|
|
import divisional_charts_extended # noqa: E402
|
|
import functional_benefics # noqa: E402
|
|
import jaimini # noqa: E402
|
|
import shadbala # noqa: E402
|
|
import narayana_dasha # noqa: E402
|
|
import saham_daynight # noqa: E402
|
|
import special_lagnas # noqa: E402
|
|
import varga # noqa: E402
|
|
from ayanamsa_utils import DEFAULT_AYANAMSA_NAME # noqa: E402
|
|
from scripts.rectification.refinement_packet import NAKSHATRA_SPAN # noqa: E402
|
|
from scripts.rectification.kp_cusp_observation import observe_kp_cusps # noqa: E402
|
|
|
|
AYANAMSA: Final = DEFAULT_AYANAMSA_NAME
|
|
NODE_MODE: Final = "mean"
|
|
INPUT_CONTRACT_VERSION: Final = "rectification-candidate-input-v2"
|
|
|
|
DomainConfig = tuple[tuple[str, ...], tuple[int, ...]]
|
|
DOMAIN_CONFIG: Final[dict[EventDomain, DomainConfig]] = {
|
|
"education": (("D24", "D5"), (4, 5, 9)),
|
|
"relocation": (("D4",), (4, 12)),
|
|
"relationship": (("D9",), (7,)),
|
|
"career": (("D10",), (10,)),
|
|
"finance": (("D2", "D11"), (2, 11)),
|
|
"health_pressure": (("D30",), (6, 8, 12)),
|
|
# 六亲: D12 parents, D7 children/spouse detail, D3 siblings, plus D1 houses 3/4/5/9.
|
|
"family": (("D12", "D7", "D3"), (3, 4, 5, 9)),
|
|
# Dated appearance/marks: D1 lagna / 1st house only. Not a primary formula.
|
|
"appearance": ((), (1,)),
|
|
# Occupation notes: D10 + D1 10th house, auxiliary, no type labels.
|
|
"occupation": (("D10",), (10,)),
|
|
}
|
|
AUXILIARY_DOMAINS: Final[frozenset[str]] = frozenset({"appearance", "occupation"})
|
|
AUXILIARY_SCORE_FACTOR: Final = 0.4
|
|
OBSERVATION_ONLY_LAYERS: Final[frozenset[str]] = frozenset({"KP_cusps"})
|
|
|
|
|
|
class RectificationEventCalculationError(RuntimeError):
|
|
"""Raised when stored rectification evidence cannot be calculated safely."""
|
|
|
|
|
|
def _event_datetime(event: LifeEvent) -> datetime:
|
|
match event["precision"]:
|
|
case "day":
|
|
return datetime.strptime(event["date"], "%Y-%m-%d")
|
|
case "month":
|
|
return datetime.strptime(f"{event['date']}-15", "%Y-%m-%d")
|
|
case "year":
|
|
return datetime.strptime(f"{event['date']}-07-01", "%Y-%m-%d")
|
|
case unreachable:
|
|
assert_never(unreachable)
|
|
|
|
|
|
def _candidate_datetimes(request: RectificationEventRequest) -> list[datetime]:
|
|
from scripts.rectification.candidate_window import enumerate_candidate_window
|
|
|
|
try:
|
|
return enumerate_candidate_window(request)
|
|
except ValueError as exc:
|
|
raise RectificationEventCalculationError(str(exc)) from exc
|
|
|
|
|
|
def _active_vimshottari(
|
|
birth_date: str,
|
|
moon_longitude: float,
|
|
event_at: datetime,
|
|
timeline: list[dict[str, Any]] | None = None,
|
|
) -> tuple[str, str, str]:
|
|
if timeline is None:
|
|
nakshatra, progress, _ = dasha_analyzer.lon_to_nakshatra(moon_longitude)
|
|
timeline, _, _, _ = dasha_analyzer.build_dasha_timeline(
|
|
birth_date,
|
|
nakshatra,
|
|
progress,
|
|
)
|
|
_, major = dasha_analyzer.find_current(timeline, event_at)
|
|
minor = dasha_analyzer.find_current_sub(
|
|
dasha_analyzer.build_antardasha(major),
|
|
event_at,
|
|
)
|
|
pratyantar = dasha_analyzer.find_current_sub(
|
|
dasha_analyzer.build_antardasha(minor),
|
|
event_at,
|
|
)
|
|
return str(major["lord"]), str(minor["lord"]), str(pratyantar["lord"])
|
|
|
|
|
|
def _active_narayana(
|
|
ascendant_index: int,
|
|
planet_longitudes: dict[str, float],
|
|
birth_at: datetime,
|
|
event_at: datetime,
|
|
periods: list[dict[str, Any]] | None = None,
|
|
) -> tuple[int | None, int | None]:
|
|
if periods is None:
|
|
periods = narayana_dasha.calc_narayana_mahadasha(
|
|
ascendant_index,
|
|
planet_longitudes,
|
|
)
|
|
age = max((event_at - birth_at).total_seconds() / (365.2425 * 86_400), 0.0)
|
|
active = narayana_dasha.get_current_narayana_dasha(periods, age)
|
|
major = active.get("md") or {}
|
|
minor = active.get("ad") or {}
|
|
return major.get("sign_idx"), minor.get("sign_idx")
|
|
|
|
|
|
def _varga_chart(charts: dict, prefix: str) -> dict | None:
|
|
return next(
|
|
(chart for name, chart in charts.items() if name.startswith(f"{prefix}_")),
|
|
None,
|
|
)
|
|
|
|
|
|
def _d11_chart(planet_longitudes: dict[str, float], ascendant_longitude: float) -> dict:
|
|
"""Adapt the repository's Rudramsa implementation to the event-score shape."""
|
|
raw = divisional_charts_extended.DivisionalChartsCalculator().calculate_all_vargas(
|
|
planet_longitudes, ascendant_longitude,
|
|
)["Rudramsa"]
|
|
return {
|
|
"Ascendant": {"sign_idx": raw["ascendant"]["sign_index"]},
|
|
**{
|
|
planet: {"sign_idx": value["sign_index"]}
|
|
for planet, value in raw["planets"].items()
|
|
},
|
|
}
|
|
|
|
|
|
def _relative_house(sign_index: int, ascendant_index: int) -> int:
|
|
return (sign_index - ascendant_index) % 12 + 1
|
|
|
|
|
|
def _house_lords(ascendant_index: int, houses: tuple[int, ...]) -> set[str]:
|
|
return {
|
|
narayana_dasha.SIGN_LORDS[
|
|
narayana_dasha.SIGNS[(ascendant_index + house - 1) % 12]
|
|
]
|
|
for house in houses
|
|
}
|
|
|
|
|
|
def _planet_house(chart: dict, planet: str) -> int | None:
|
|
raw = (chart.get("planets", {}).get(planet) or {}).get("house")
|
|
return int(raw) if isinstance(raw, int | float) else None
|
|
|
|
|
|
def _varga_house(chart: dict, planet: str) -> int | None:
|
|
ascendant = chart.get("Ascendant") or {}
|
|
placement = chart.get(planet) or {}
|
|
ascendant_index = ascendant.get("sign_idx")
|
|
planet_index = placement.get("sign_idx")
|
|
if not isinstance(ascendant_index, int) or not isinstance(planet_index, int):
|
|
return None
|
|
return _relative_house(planet_index, ascendant_index)
|
|
|
|
|
|
def _score_event(
|
|
*,
|
|
candidate_time: str,
|
|
event: LifeEvent,
|
|
natal_chart: dict,
|
|
varga_charts: list[dict],
|
|
vimshottari: tuple[str, str, str],
|
|
narayana: tuple[int | None, int | None],
|
|
arudha_padas: dict,
|
|
) -> CandidateEvidence:
|
|
_, target_houses = DOMAIN_CONFIG[event["domain"]]
|
|
ascendant_index = int(natal_chart["ascendant"]["lon"] // 30)
|
|
target_lords = _house_lords(ascendant_index, target_houses)
|
|
functional = functional_benefics.derive_functional_benefic_malefic(
|
|
natal_chart["ascendant"].get("sign")
|
|
)
|
|
functional_benefics_set = set(functional.get("functional_benefics") or [])
|
|
functional_malefics_set = set(functional.get("functional_malefics") or [])
|
|
major_lord, minor_lord, pratyantar_lord = vimshottari
|
|
rules: list[str] = []
|
|
points = 0.0
|
|
|
|
for lord, weight, label in (
|
|
(major_lord, 2.0, "vim_md"),
|
|
(minor_lord, 1.5, "vim_ad"),
|
|
(pratyantar_lord, 0.75, "vim_pd"),
|
|
):
|
|
if _planet_house(natal_chart, lord) in target_houses:
|
|
rules.append(f"{label}_domain_house")
|
|
points += weight
|
|
if lord in target_lords:
|
|
rules.append(f"{label}_domain_lord")
|
|
points += weight
|
|
for varga_chart in varga_charts:
|
|
if _varga_house(varga_chart, lord) in target_houses:
|
|
rules.append(f"{label}_domain_varga")
|
|
points += weight / (2 * len(varga_charts))
|
|
if lord in functional_benefics_set:
|
|
rules.append(f"{label}_functional_benefic_auxiliary")
|
|
points += 0.2
|
|
elif lord in functional_malefics_set:
|
|
rules.append(f"{label}_functional_malefic_auxiliary")
|
|
points -= 0.1
|
|
|
|
for sign_index, weight, label in (
|
|
(narayana[0], 2.0, "narayana_md"),
|
|
(narayana[1], 1.0, "narayana_ad"),
|
|
):
|
|
if sign_index is not None and _relative_house(sign_index, ascendant_index) in target_houses:
|
|
rules.append(f"{label}_domain_house")
|
|
points += weight
|
|
arudha_keys = (
|
|
("A7", "UL") if event["domain"] == "relationship"
|
|
else ("A10",) if event["domain"] in {"career", "occupation"}
|
|
else ()
|
|
)
|
|
arudha_signs = {
|
|
value.get("sign_idx") for key in arudha_keys
|
|
if isinstance((value := arudha_padas.get(key)), dict) and isinstance(value.get("sign_idx"), int)
|
|
}
|
|
if arudha_signs:
|
|
for lord, label in ((major_lord, "vim_md"), (minor_lord, "vim_ad"), (pratyantar_lord, "vim_pd")):
|
|
planet = natal_chart.get("planets", {}).get(lord) or {}
|
|
if isinstance(planet.get("lon"), (int, float)) and int(planet["lon"] // 30) in arudha_signs:
|
|
rules.append(f"{label}_arudha_auxiliary")
|
|
points += 0.35
|
|
|
|
event_kind = event.get("event_kind", event["domain"])
|
|
if event["domain"] in AUXILIARY_DOMAINS:
|
|
points *= AUXILIARY_SCORE_FACTOR
|
|
rules.append(
|
|
"occupation_auxiliary_not_primary"
|
|
if event["domain"] == "occupation"
|
|
else "appearance_auxiliary_not_primary"
|
|
)
|
|
if not rules:
|
|
rules.append("no_domain_activation")
|
|
rules.append(f"event_kind:{event_kind}")
|
|
weighted_points = round(points * precision_weight(event["precision"]), 4)
|
|
return {
|
|
"event_id": event["id"],
|
|
"domain": event["domain"],
|
|
"candidate_time": candidate_time,
|
|
"rule_ids": rules,
|
|
"points": weighted_points,
|
|
}
|
|
|
|
|
|
def _transit_chart_cache_key(
|
|
request: RectificationEventRequest,
|
|
event_at: datetime,
|
|
) -> tuple[Any, ...]:
|
|
return (
|
|
event_at.date().isoformat(),
|
|
float(request["lat"]),
|
|
float(request["lon"]),
|
|
float(request["tz"]),
|
|
request.get("ayanamsa", AYANAMSA),
|
|
request.get("node_mode", NODE_MODE),
|
|
)
|
|
|
|
|
|
def _controlled_transit_rules(
|
|
request: RectificationEventRequest,
|
|
event: LifeEvent,
|
|
natal_ascendant_index: int,
|
|
target_houses: tuple[int, ...],
|
|
transit_chart_cache: dict[tuple[Any, ...], dict[str, Any]] | None = None,
|
|
) -> list[str]:
|
|
"""Use only Jupiter/Saturn and only day/month dated events as a weak check."""
|
|
if event["precision"] == "year":
|
|
return []
|
|
event_at = _event_datetime(event)
|
|
payload = {
|
|
"year": event_at.year, "month": event_at.month, "day": event_at.day,
|
|
"hour": 12, "minute": 0, "lat": request["lat"], "lon": request["lon"],
|
|
"tz": request["tz"],
|
|
"ayanamsa": request.get("ayanamsa", AYANAMSA),
|
|
"node_mode": request.get("node_mode", NODE_MODE),
|
|
}
|
|
cache_key = _transit_chart_cache_key(request, event_at)
|
|
if transit_chart_cache is not None and cache_key in transit_chart_cache:
|
|
transit_chart = transit_chart_cache[cache_key]
|
|
else:
|
|
transit_chart = domain_calculation_service.compute_chart(payload)
|
|
if transit_chart_cache is not None:
|
|
transit_chart_cache[cache_key] = transit_chart
|
|
rules: list[str] = []
|
|
for planet in ("Jupiter", "Saturn"):
|
|
item = transit_chart.get("planets", {}).get(planet) or {}
|
|
if isinstance(item.get("lon"), (int, float)) and _relative_house(int(item["lon"] // 30), natal_ascendant_index) in target_houses:
|
|
rules.append(f"controlled_transit_{planet.lower()}_domain_house")
|
|
return rules
|
|
|
|
|
|
def _ashtakavarga_auxiliary(
|
|
natal_chart: dict,
|
|
ascendant_index: int,
|
|
target_houses: tuple[int, ...],
|
|
ashtakavarga_result: dict[str, Any] | None = None,
|
|
) -> tuple[list[str], float]:
|
|
"""Return a bounded SAV consistency adjustment, never a standalone trigger."""
|
|
result = ashtakavarga_result
|
|
if result is None:
|
|
result = ashtakavarga.calc_ashtakavarga(natal_chart.get("planets", {}), ascendant_index)
|
|
if not result.get("all_bav_valid") or not (result.get("sav") or {}).get("valid"):
|
|
return [], 0.0
|
|
house_scores = result.get("house_scores_full") or {}
|
|
values = [house_scores.get(f"house_{house}", {}).get("sav_score") for house in target_houses]
|
|
numeric = [float(value) for value in values if isinstance(value, (int, float))]
|
|
if not numeric:
|
|
return [], 0.0
|
|
average = sum(numeric) / len(numeric)
|
|
if average >= 32:
|
|
return ["ashtakavarga_target_house_support_auxiliary"], 0.2
|
|
if average <= 24:
|
|
return ["ashtakavarga_target_house_pressure_auxiliary"], -0.1
|
|
return [], 0.0
|
|
|
|
|
|
def _shadbala_verified_components_auxiliary(
|
|
natal_chart: dict,
|
|
birth_hour: float,
|
|
dasha_lords: tuple[str, str, str],
|
|
shadbala_result: dict[str, Any] | None = None,
|
|
) -> tuple[list[str], float]:
|
|
"""Use only Sthana/Drik/Naisargika, whose oracle comparison is already matched."""
|
|
planets = natal_chart.get("planets", {})
|
|
sun = planets.get("Sun") or {}
|
|
moon = planets.get("Moon") or {}
|
|
if not isinstance(sun.get("lon"), (int, float)) or not isinstance(moon.get("lon"), (int, float)):
|
|
return [], 0.0
|
|
result = shadbala_result
|
|
if result is None:
|
|
result = shadbala.calc_shadbala(
|
|
planets, str(natal_chart["ascendant"].get("sign") or "Aries"), birth_hour,
|
|
float(sun["lon"]), float(moon["lon"]),
|
|
)
|
|
values = {
|
|
planet: float((row.get("sthana_bala") or {}).get("total", 0)) + float(row.get("drik_bala", 0)) + float(row.get("naisargika_bala", 0))
|
|
for planet, row in (result.get("planets") or {}).items()
|
|
}
|
|
if not values:
|
|
return [], 0.0
|
|
baseline = sum(values.values()) / len(values)
|
|
active = [values[lord] for lord in dasha_lords if lord in values]
|
|
if not active:
|
|
return [], 0.0
|
|
average = sum(active) / len(active)
|
|
if average > baseline:
|
|
return ["shadbala_sthana_drik_naisargika_support_auxiliary"], 0.1
|
|
if average < baseline:
|
|
return ["shadbala_sthana_drik_naisargika_pressure_auxiliary"], -0.05
|
|
return [], 0.0
|
|
|
|
|
|
def _feature_hash(value: Any) -> str:
|
|
normalized = json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), default=str)
|
|
return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _arudha_sign(arudha_padas: dict, key: str) -> int | None:
|
|
value = arudha_padas.get(key) or {}
|
|
sign_index = value.get("sign_idx")
|
|
return int(sign_index) if isinstance(sign_index, int) and 0 <= sign_index <= 11 else None
|
|
|
|
|
|
def _pada_index(longitude: float) -> int:
|
|
return int((float(longitude) % 360.0) / (NAKSHATRA_SPAN / 4.0)) % 108
|
|
|
|
|
|
@lru_cache(maxsize=64)
|
|
def _sunrise_local_naive(date_iso: str, lat: float, lon: float, tz: float) -> datetime | None:
|
|
"""Real sunrise only. Polar or unavailable locations omit Hora/Ghati; never invent 06:00."""
|
|
try:
|
|
import swisseph as swe
|
|
noon = datetime.fromisoformat(f"{date_iso}T12:00:00")
|
|
row = saham_daynight.determine_daytime(noon, lat=lat, lon=lon, tz=tz)
|
|
year, month, day, ut_hours = swe.revjul(float(row["sunrise_jd_ut"]))
|
|
return datetime(int(year), int(month), int(day)) + timedelta(hours=float(ut_hours) + float(tz))
|
|
except (saham_daynight.SahamDayNightError, TypeError, ValueError, OverflowError, OSError):
|
|
return None
|
|
|
|
|
|
def _fine_minute_indices(
|
|
*,
|
|
ascendant_longitude: float,
|
|
candidate_at: datetime,
|
|
lat: float,
|
|
lon: float,
|
|
tz: float,
|
|
sun_degree: float | None,
|
|
moon_degree: float | None,
|
|
) -> dict[str, int]:
|
|
wrapped = float(ascendant_longitude) % 360.0
|
|
indices: dict[str, int] = {"pada_index": _pada_index(ascendant_longitude)}
|
|
calculator = special_lagnas.SpecialLagnasCalculator()
|
|
if isinstance(sun_degree, (int, float)) and isinstance(moon_degree, (int, float)):
|
|
try:
|
|
bhava = calculator.calculate_bhava_lagna(wrapped, float(sun_degree), float(moon_degree))
|
|
indices["bhava_sign_index"] = int(float(bhava["degree"]) // 30.0) % 12
|
|
except (TypeError, ValueError, KeyError, OverflowError):
|
|
pass
|
|
sunrise = _sunrise_local_naive(candidate_at.date().isoformat(), lat, lon, tz)
|
|
if sunrise is None:
|
|
return indices
|
|
try:
|
|
hora = calculator.calculate_hora_lagna(wrapped, 0.0, candidate_at, sunrise)
|
|
ghati = calculator.calculate_ghati_lagna(wrapped, candidate_at, sunrise)
|
|
hora_degree = float(hora["degree"])
|
|
ghati_degree = float(ghati["degree"])
|
|
except (TypeError, ValueError, KeyError, OverflowError):
|
|
return indices
|
|
hora_index = int(hora_degree // 30.0) % 12
|
|
ghati_index = int(ghati_degree // 30.0) % 12
|
|
indices["hora_sign_index"] = hora_index
|
|
indices["ghati_sign_index"] = ghati_index
|
|
indices["pranapada_sign_index"] = (hora_index + ghati_index) % 12
|
|
return indices
|
|
|
|
|
|
def build_candidate_static_context(
|
|
request: RectificationEventRequest,
|
|
candidate_at: datetime,
|
|
) -> dict[str, Any]:
|
|
"""Compute every candidate-minute natal layer once for scoring and diagnostics."""
|
|
chart = domain_calculation_service.compute_chart({
|
|
"year": candidate_at.year,
|
|
"month": candidate_at.month,
|
|
"day": candidate_at.day,
|
|
"hour": candidate_at.hour,
|
|
"minute": candidate_at.minute,
|
|
"lat": request["lat"],
|
|
"lon": request["lon"],
|
|
"tz": request["tz"],
|
|
"ayanamsa": request.get("ayanamsa", AYANAMSA),
|
|
"node_mode": request.get("node_mode", NODE_MODE),
|
|
})
|
|
planet_longitudes = {
|
|
name: float(data["lon"])
|
|
for name, data in chart.get("planets", {}).items()
|
|
if isinstance(data, dict) and isinstance(data.get("lon"), int | float)
|
|
}
|
|
ascendant_longitude = float(chart["ascendant"]["lon"])
|
|
ascendant_index = int(ascendant_longitude // 30)
|
|
arudha = jaimini.calc_arudha_padas(ascendant_index, planet_longitudes)
|
|
arudha_padas = {**(arudha.get("padas") or {}), "UL": arudha.get("upapada") or {}}
|
|
charts = varga.calc_all_vargas(
|
|
planet_longitudes,
|
|
ascendant_longitude,
|
|
divisions=[2, 3, 4, 5, 7, 9, 10, 12, 24, 30],
|
|
)
|
|
d11_chart = _d11_chart(planet_longitudes, ascendant_longitude)
|
|
varga_charts = {
|
|
prefix: d11_chart if prefix == "D11" else _varga_chart(charts, prefix)
|
|
for prefix in ("D2", "D3", "D4", "D5", "D7", "D9", "D10", "D11", "D12", "D24", "D30")
|
|
}
|
|
available_layers = ["D1"]
|
|
blocked_layers: list[str] = []
|
|
varga_ascendants: dict[str, int] = {}
|
|
for prefix, value in varga_charts.items():
|
|
ascendant = (value or {}).get("Ascendant") or {}
|
|
sign_index = ascendant.get("sign_idx")
|
|
if isinstance(sign_index, int) and 0 <= sign_index <= 11:
|
|
varga_ascendants[prefix] = sign_index
|
|
available_layers.append(prefix)
|
|
else:
|
|
blocked_layers.append(prefix)
|
|
|
|
arudha_signs = {key: _arudha_sign(arudha_padas, key) for key in ("A7", "A10", "UL")}
|
|
for key, sign_index in arudha_signs.items():
|
|
(available_layers if sign_index is not None else blocked_layers).append(key)
|
|
|
|
ashtakavarga_result = None
|
|
try:
|
|
ashtakavarga_result = ashtakavarga.calc_ashtakavarga(chart.get("planets", {}), ascendant_index)
|
|
available_layers.append("Ashtakavarga")
|
|
except (KeyError, TypeError, ValueError):
|
|
blocked_layers.append("Ashtakavarga")
|
|
|
|
shadbala_result = None
|
|
try:
|
|
shadbala_result = shadbala.calc_shadbala(
|
|
chart.get("planets", {}),
|
|
str(chart["ascendant"].get("sign")),
|
|
candidate_at.hour + candidate_at.minute / 60,
|
|
planet_longitudes["Sun"],
|
|
planet_longitudes["Moon"],
|
|
birth_minute=float(candidate_at.minute),
|
|
)
|
|
available_layers.append("Shadbala")
|
|
except (KeyError, TypeError, ValueError):
|
|
blocked_layers.append("Shadbala")
|
|
|
|
vimshottari_timeline = None
|
|
try:
|
|
nakshatra, progress, _ = dasha_analyzer.lon_to_nakshatra(planet_longitudes["Moon"])
|
|
vimshottari_timeline, _, _, _ = dasha_analyzer.build_dasha_timeline(
|
|
candidate_at.date().isoformat(),
|
|
nakshatra,
|
|
progress,
|
|
)
|
|
except (KeyError, TypeError, ValueError):
|
|
vimshottari_timeline = None
|
|
|
|
narayana_periods = None
|
|
try:
|
|
narayana_periods = narayana_dasha.calc_narayana_mahadasha(
|
|
ascendant_index,
|
|
planet_longitudes,
|
|
)
|
|
except (KeyError, TypeError, ValueError):
|
|
narayana_periods = None
|
|
|
|
birth_info = chart.get("birth_info") if isinstance(chart.get("birth_info"), dict) else {}
|
|
kp_snapshot = observe_kp_cusps(
|
|
birth_info.get("julian_day"),
|
|
float(request["lat"]),
|
|
float(request["lon"]),
|
|
)
|
|
if kp_snapshot.get("status") == "executed":
|
|
available_layers.append("KP_cusps")
|
|
else:
|
|
blocked_layers.append("KP_cusps")
|
|
|
|
feature_payload = {
|
|
"time": candidate_at.strftime("%H:%M"),
|
|
"ascendant_degree": ascendant_longitude,
|
|
"ascendant_sign_index": ascendant_index,
|
|
**_fine_minute_indices(
|
|
ascendant_longitude=ascendant_longitude,
|
|
candidate_at=candidate_at,
|
|
lat=float(request["lat"]),
|
|
lon=float(request["lon"]),
|
|
tz=float(request["tz"]),
|
|
sun_degree=planet_longitudes.get("Sun") if isinstance(planet_longitudes.get("Sun"), (int, float)) else None,
|
|
moon_degree=planet_longitudes.get("Moon") if isinstance(planet_longitudes.get("Moon"), (int, float)) else None,
|
|
),
|
|
**(kp_snapshot.get("indices") or {}),
|
|
"kp_cusps": kp_snapshot,
|
|
"varga_ascendants": varga_ascendants,
|
|
"arudha_signs": arudha_signs,
|
|
"available_layers": sorted(set(available_layers)),
|
|
"blocked_layers": sorted(set(blocked_layers)),
|
|
"fingerprints": {
|
|
"natal": str(chart.get("result_hash") or _feature_hash({"ascendant": chart.get("ascendant"), "planets": chart.get("planets")})),
|
|
"vargas": _feature_hash(varga_ascendants),
|
|
"arudha": _feature_hash(arudha_signs),
|
|
"ashtakavarga": _feature_hash(ashtakavarga_result) if ashtakavarga_result is not None else "blocked",
|
|
"shadbala": _feature_hash(shadbala_result) if shadbala_result is not None else "blocked",
|
|
},
|
|
}
|
|
feature_payload["fingerprints"]["static"] = _feature_hash(feature_payload)
|
|
return {
|
|
"candidate_at": candidate_at,
|
|
"chart": chart,
|
|
"planet_longitudes": planet_longitudes,
|
|
"ascendant_longitude": ascendant_longitude,
|
|
"ascendant_index": ascendant_index,
|
|
"arudha_padas": arudha_padas,
|
|
"varga_charts": varga_charts,
|
|
"feature": feature_payload,
|
|
"ashtakavarga_result": ashtakavarga_result,
|
|
"shadbala_result": shadbala_result,
|
|
"vimshottari_timeline": vimshottari_timeline,
|
|
"narayana_periods": narayana_periods,
|
|
}
|
|
|
|
|
|
def compute_candidate_static_contexts(
|
|
request: RectificationEventRequest,
|
|
*,
|
|
candidates: Sequence[datetime] | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
candidate_datetimes = list(candidates) if candidates is not None else _candidate_datetimes(request)
|
|
from scripts.rectification.candidate_window import candidate_positions
|
|
|
|
positions = candidate_positions(request, candidate_datetimes)
|
|
return [{**build_candidate_static_context(request, candidate), **position}
|
|
for candidate, position in zip(candidate_datetimes, positions)]
|
|
|
|
|
|
def _candidate_row(
|
|
request: RectificationEventRequest,
|
|
context: dict[str, Any],
|
|
) -> CandidateScoreRow:
|
|
candidate_at = context["candidate_at"]
|
|
chart = context["chart"]
|
|
planet_longitudes = context["planet_longitudes"]
|
|
ascendant_index = context["ascendant_index"]
|
|
arudha_padas = context["arudha_padas"]
|
|
varga_charts = context["varga_charts"]
|
|
moon_longitude = planet_longitudes["Moon"]
|
|
evidence: list[CandidateEvidence] = []
|
|
missing_layers: list[str] = []
|
|
|
|
for event in request["events"]:
|
|
event_at = _event_datetime(event)
|
|
prefixes, _ = DOMAIN_CONFIG[event["domain"]]
|
|
domain_vargas = [varga_charts[prefix] for prefix in prefixes]
|
|
if any(chart is None for chart in domain_vargas):
|
|
missing_layers.extend(prefixes)
|
|
continue
|
|
try:
|
|
vimshottari = _active_vimshottari(
|
|
candidate_at.date().isoformat(),
|
|
moon_longitude,
|
|
event_at,
|
|
context.get("vimshottari_timeline"),
|
|
)
|
|
except (KeyError, TypeError, ValueError):
|
|
missing_layers.append("Vimshottari_MD_AD_PD")
|
|
continue
|
|
try:
|
|
narayana = _active_narayana(
|
|
ascendant_index,
|
|
planet_longitudes,
|
|
candidate_at,
|
|
event_at,
|
|
context.get("narayana_periods"),
|
|
)
|
|
except (KeyError, TypeError, ValueError):
|
|
missing_layers.append("Narayana_MD_AD")
|
|
continue
|
|
if narayana[0] is None or narayana[1] is None:
|
|
missing_layers.append("Narayana_MD_AD")
|
|
continue
|
|
evidence.append(_score_event(
|
|
candidate_time=candidate_at.strftime("%H:%M"),
|
|
event=event,
|
|
natal_chart=chart,
|
|
varga_charts=[chart for chart in domain_vargas if chart is not None],
|
|
vimshottari=vimshottari,
|
|
narayana=narayana,
|
|
arudha_padas=arudha_padas,
|
|
))
|
|
transit_rules = _controlled_transit_rules(
|
|
request,
|
|
event,
|
|
ascendant_index,
|
|
DOMAIN_CONFIG[event["domain"]][1],
|
|
context.get("_transit_chart_cache"),
|
|
)
|
|
if transit_rules:
|
|
evidence[-1]["rule_ids"].extend(transit_rules)
|
|
evidence[-1]["points"] = round(evidence[-1]["points"] + 0.25 * len(transit_rules) * precision_weight(event["precision"]), 4)
|
|
av_rules, av_points = _ashtakavarga_auxiliary(
|
|
chart,
|
|
ascendant_index,
|
|
DOMAIN_CONFIG[event["domain"]][1],
|
|
context.get("ashtakavarga_result"),
|
|
)
|
|
if av_rules:
|
|
evidence[-1]["rule_ids"].extend(av_rules)
|
|
evidence[-1]["points"] = round(evidence[-1]["points"] + av_points * precision_weight(event["precision"]), 4)
|
|
shadbala_rules, shadbala_points = _shadbala_verified_components_auxiliary(
|
|
chart,
|
|
candidate_at.hour + candidate_at.minute / 60,
|
|
vimshottari,
|
|
context.get("shadbala_result"),
|
|
)
|
|
if shadbala_rules:
|
|
evidence[-1]["rule_ids"].extend(shadbala_rules)
|
|
evidence[-1]["points"] = round(evidence[-1]["points"] + shadbala_points * precision_weight(event["precision"]), 4)
|
|
|
|
return {
|
|
"time": candidate_at.strftime("%H:%M"),
|
|
"score": round(sum(item["points"] for item in evidence), 4),
|
|
"evidence": evidence,
|
|
"missing_layers": sorted(set(
|
|
missing_layers
|
|
+ [layer for layer in context["feature"]["blocked_layers"] if layer not in OBSERVATION_ONLY_LAYERS]
|
|
)),
|
|
}
|
|
|
|
|
|
def _canonical_input_contract(request: RectificationEventRequest) -> tuple[dict, str]:
|
|
payload = {
|
|
"schema_version": INPUT_CONTRACT_VERSION,
|
|
"birth_date": request["birth_date"],
|
|
"candidate_range": {
|
|
"start_time": request["start_time"],
|
|
"end_time": request["end_time"],
|
|
"step_minutes": int(request.get("minute_step") or 1),
|
|
},
|
|
"location": {
|
|
"latitude": float(request["lat"]),
|
|
"longitude": float(request["lon"]),
|
|
"timezone_offset": float(request["tz"]),
|
|
"timezone_source": "explicit_offset",
|
|
},
|
|
"events": [{
|
|
"id": event["id"],
|
|
"domain": event["domain"],
|
|
"event_kind": event.get("event_kind", event["domain"]),
|
|
"date": event["date"],
|
|
"precision": event["precision"],
|
|
"summary": event.get("summary", ""),
|
|
} for event in request["events"]],
|
|
"calculation": {
|
|
"ayanamsa": request.get("ayanamsa", AYANAMSA),
|
|
"node_mode": request.get("node_mode", NODE_MODE),
|
|
"ephemeris_source": "swisseph_calc_ut",
|
|
},
|
|
}
|
|
if "candidate_intervals" in request:
|
|
payload["schema_version"] = "rectification-candidate-input-v3"
|
|
payload["candidate_intervals"] = request["candidate_intervals"]
|
|
normalized = json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(",", ":"))
|
|
return payload, hashlib.sha256(normalized.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _leave_one_event_out(rows: list[CandidateScoreRow], events: list[LifeEvent]) -> dict:
|
|
if len(events) < 2:
|
|
return {"status": "blocked", "runs": [], "reason": "insufficient_events"}
|
|
original_top = max(row["score"] for row in rows)
|
|
original_times = {row["time"] for row in rows if row["score"] == original_top}
|
|
runs = []
|
|
all_stable = True
|
|
for event in events:
|
|
rescored = []
|
|
for row in rows:
|
|
removed = sum(item["points"] for item in row["evidence"] if item["event_id"] == event["id"])
|
|
rescored.append((row["time"], round(row["score"] - removed, 4)))
|
|
top_score = max(score for _, score in rescored)
|
|
top_times = [candidate_time for candidate_time, score in rescored if score == top_score]
|
|
stable = len(top_times) == 1 and set(top_times) == original_times
|
|
all_stable = all_stable and stable
|
|
runs.append({
|
|
"removed_event_id": event["id"],
|
|
"top_times": top_times,
|
|
"top_score": top_score,
|
|
"original_leader_retained": stable,
|
|
})
|
|
return {
|
|
"status": "pass" if all_stable else "fail",
|
|
"runs": runs,
|
|
"boundary": "Leave-one-event-out must retain the same unique leading minute.",
|
|
}
|
|
|
|
|
|
def compute_event_candidate_result(request: RectificationEventRequest) -> CandidateResult:
|
|
"""Compute actual minute candidates locally and return a guarded result."""
|
|
return adjudicate_event_candidate_rows(request, compute_event_candidate_rows(request))
|
|
|
|
|
|
def adjudicate_event_candidate_rows(
|
|
request: RectificationEventRequest,
|
|
rows: list[CandidateScoreRow],
|
|
) -> CandidateResult:
|
|
"""Adjudicate already-computed rows using the production evidence gates."""
|
|
normalized = json.dumps(request, ensure_ascii=True, sort_keys=True, separators=(",", ":"))
|
|
fingerprint = hashlib.sha256(normalized.encode("utf-8")).hexdigest()
|
|
input_contract, input_hash = _canonical_input_contract(request)
|
|
leave_one_out = _leave_one_event_out(rows, request["events"])
|
|
return adjudicate_candidate_rows(
|
|
rows,
|
|
event_count=len(request["events"]),
|
|
domain_count=len({event["domain"] for event in request["events"]}),
|
|
request_fingerprint=fingerprint,
|
|
canonical_input_hash=input_hash,
|
|
calculation_contract=input_contract,
|
|
leave_one_event_out=leave_one_out,
|
|
)
|
|
|
|
|
|
def compute_event_candidate_rows(
|
|
request: RectificationEventRequest,
|
|
*,
|
|
candidates: Sequence[datetime] | None = None,
|
|
static_contexts: Sequence[dict[str, Any]] | None = None,
|
|
) -> list[CandidateScoreRow]:
|
|
"""Return every computed minute row while reusing one static chart scan per candidate."""
|
|
contexts = list(static_contexts) if static_contexts is not None else compute_candidate_static_contexts(request, candidates=candidates)
|
|
transit_chart_cache: dict[tuple[Any, ...], dict[str, Any]] = {}
|
|
return [
|
|
_candidate_row(request, {**context, "_transit_chart_cache": transit_chart_cache})
|
|
for context in contexts
|
|
]
|