ClaudeandClaude Opus 5.5
0df8a3283f
Raw totals are only compared when every candidate computed the same (event, layer) evidence units. Otherwise winning_segment stays the raw leader (provisional_leader) and comparable_leader re-ranks over the units all candidates share; a missing unit is dropped for everyone, never scored 0. The live stage Python engine now skips an event whose Narayana is unavailable instead of scoring it 0, repeated facts count once, and the stage labels an unequal-coverage leader as provisional. 占星师 2026-11-08 验收清单 §6 / §9 / §10. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DbBXoyn7Hj3LCW3BUs5bZ8
925 lines
38 KiB
Python
925 lines
38 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,
|
||
independent_events,
|
||
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 narayana_rath # noqa: E402
|
||
from narayana_legacy_label import narayana_age # noqa: E402
|
||
import saham_daynight # noqa: E402
|
||
import special_lagnas # noqa: E402
|
||
import varga # noqa: E402
|
||
from chart_dasha import birth_moment # 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 _vimshottari_timeline(birth_at: datetime, moon_longitude: float) -> list[dict[str, Any]]:
|
||
"""Mahadashas from the candidate's birth moment, as ``cmd_dasha`` / the chart page (BUG-1720).
|
||
|
||
The frozen ``dasha_analyzer.build_dasha_timeline`` took only the birth date
|
||
and started every period from midnight of that day.
|
||
"""
|
||
if not isinstance(birth_at, datetime):
|
||
raise TypeError("birth_at must be the candidate birth datetime")
|
||
timeline, _, _, _ = birth_moment.mahadashas(birth_at, float(moon_longitude))
|
||
return timeline
|
||
|
||
|
||
def _active_vimshottari(
|
||
birth_at: datetime,
|
||
moon_longitude: float,
|
||
event_at: datetime,
|
||
timeline: list[dict[str, Any]] | None = None,
|
||
) -> tuple[str, str, str]:
|
||
"""MD / AD / PD lords at ``event_at`` with float sub-period splits (BUG-1720).
|
||
|
||
Product approved 2026-11-08 replacing the integer-day
|
||
``dasha_analyzer.build_antardasha`` (its last child ended up to a day early)
|
||
with ``birth_moment.split_period`` at both levels.
|
||
"""
|
||
if not isinstance(birth_at, datetime):
|
||
raise TypeError("birth_at must be the candidate birth datetime")
|
||
if timeline is None:
|
||
timeline = _vimshottari_timeline(birth_at, moon_longitude)
|
||
_, major = dasha_analyzer.find_current(timeline, event_at)
|
||
minor = dasha_analyzer.find_current_sub(birth_moment.split_period(major), event_at)
|
||
pratyantar = dasha_analyzer.find_current_sub(birth_moment.split_period(minor), event_at)
|
||
return str(major["lord"]), str(minor["lord"]), str(pratyantar["lord"])
|
||
|
||
|
||
# 原值：校正打分调用 narayana_dasha.calc_narayana_mahadasha / calc_narayana_antardasha /
|
||
# get_current_narayana_dasha（旧算法：从上升起、一律顺数，固定宫主），年龄按 365.2425 天。
|
||
# 新值：与报告、咨询卡同一个来源版 Narayana（narayana_rath.calc_narayana_rath，两轮 24 行；
|
||
# 小运 narayana_rath.antardasha），年龄按 narayana_legacy_label.narayana_age（出生时刻起、365.25 天）。
|
||
# 来源版定不出（blocked / parameter_sensitive）时这一层不可用，返回 None，不回退旧算法。
|
||
# 原因：产品 2026-11-08 批准；占星师第五轮裁定 1、3；与报告/卡口径一致。
|
||
def _narayana_lons(planet_longitudes: dict[str, Any]) -> dict[str, float]:
|
||
return {
|
||
str(name): float(lon) for name, lon in planet_longitudes.items()
|
||
if isinstance(lon, (int, float)) and not isinstance(lon, bool)
|
||
}
|
||
|
||
|
||
def _narayana_periods(
|
||
ascendant_index: int,
|
||
planet_longitudes: dict[str, Any],
|
||
) -> list[dict[str, Any]] | None:
|
||
"""Source-version Narayana MD rows (two cycles), or ``None`` when that layer is blocked."""
|
||
result = narayana_rath.calc_narayana_rath(
|
||
int(ascendant_index) % 12, _narayana_lons(planet_longitudes), cycle_count=2,
|
||
)
|
||
rows = result.get("mahadasha")
|
||
return list(rows) if rows else None
|
||
|
||
|
||
def _narayana_antardashas(
|
||
period: dict[str, Any],
|
||
planet_longitudes: dict[str, Any],
|
||
) -> list[dict[str, Any]] | None:
|
||
"""The twelve source-version ADs of one MD row, or ``None`` when the AD step is blocked."""
|
||
result = narayana_rath.antardasha(period, _narayana_lons(planet_longitudes))
|
||
return list(result["rows"]) if result.get("status") == "selected" and result.get("rows") else None
|
||
|
||
|
||
def _narayana_antardasha_table(
|
||
periods: list[dict[str, Any]] | None,
|
||
planet_longitudes: dict[str, Any],
|
||
) -> dict[str, list[dict[str, Any]] | None] | None:
|
||
"""ADs per MD row, keyed by ``period_key`` (cycle 2 repeats the signs with other lengths)."""
|
||
if not periods:
|
||
return None
|
||
return {str(row["period_key"]): _narayana_antardashas(row, planet_longitudes) for row in periods}
|
||
|
||
|
||
def _running_row(rows: list[dict[str, Any]] | None, age: float) -> dict[str, Any] | None:
|
||
return next(
|
||
(row for row in rows or [] if float(row["start_age"]) <= age < float(row["end_age"])),
|
||
None,
|
||
)
|
||
|
||
|
||
def _active_narayana(
|
||
ascendant_index: int,
|
||
planet_longitudes: dict[str, float],
|
||
birth_at: datetime,
|
||
event_at: datetime,
|
||
periods: list[dict[str, Any]] | None = None,
|
||
antardasha_by_md: dict[str, list[dict[str, Any]] | None] | None = None,
|
||
) -> tuple[int | None, int | None]:
|
||
"""Running source-version Narayana MD / AD signs at ``event_at``.
|
||
|
||
Same rows and age as the product layer (``narayana_legacy_label.rath_narayana_view``
|
||
with ``current_age = narayana_age(birth, event)``). ``None`` marks a blocked
|
||
step; there is no fallback to the old algorithm.
|
||
"""
|
||
if not isinstance(birth_at, datetime):
|
||
raise TypeError("birth_at must be the candidate birth datetime")
|
||
if periods is None:
|
||
periods = _narayana_periods(ascendant_index, planet_longitudes)
|
||
if not periods:
|
||
return None, None
|
||
age = max(narayana_age(birth_at, event_at), 0.0)
|
||
major = _running_row(periods, age)
|
||
if major is None:
|
||
return None, None
|
||
if antardasha_by_md is not None and str(major.get("period_key")) in antardasha_by_md:
|
||
minors = antardasha_by_md[str(major["period_key"])]
|
||
else:
|
||
minors = _narayana_antardashas(major, planet_longitudes)
|
||
minor = _running_row(minors, age)
|
||
return major.get("sign_idx"), None if minor is None else 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,
|
||
functional: dict | None = None,
|
||
) -> CandidateEvidence:
|
||
_, target_houses = DOMAIN_CONFIG[event["domain"]]
|
||
ascendant_index = int(natal_chart["ascendant"]["lon"] // 30)
|
||
target_lords = _house_lords(ascendant_index, target_houses)
|
||
if functional is None:
|
||
# 原值：明确选中 functional_profile=FUNCTIONAL_PROFILE_BPHS（原产品表）。
|
||
# 新值：默认档（PVR 表 30），并按产品其他调用方传月亮盈亏（动星座上升的月亮）。
|
||
# 原因：产品 2026-11-08 批准；占星师第五轮裁定 1、3；与报告/卡口径一致（BUG-1759）。
|
||
functional = functional_benefics.derive_functional_benefic_malefic(
|
||
natal_chart["ascendant"].get("sign"),
|
||
moon_paksha=functional_benefics.moon_paksha_from_planets(natal_chart.get("planets")),
|
||
)
|
||
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:
|
||
vimshottari_timeline = _vimshottari_timeline(candidate_at, planet_longitudes["Moon"])
|
||
except (KeyError, TypeError, ValueError):
|
||
vimshottari_timeline = None
|
||
|
||
# BUG-1759: source-version Narayana (as the report / card); blocked -> layer unavailable, no fallback.
|
||
narayana_periods = None
|
||
try:
|
||
narayana_periods = _narayana_periods(ascendant_index, planet_longitudes)
|
||
except (KeyError, TypeError, ValueError):
|
||
narayana_periods = None
|
||
if narayana_periods is None:
|
||
blocked_layers.append("Narayana")
|
||
# 原值：明确选中 functional_profile=FUNCTIONAL_PROFILE_BPHS（原产品表）。
|
||
# 新值：默认档（PVR 表 30），传月亮盈亏。
|
||
# 原因：产品 2026-11-08 批准；占星师第五轮裁定 1、3；与报告/卡口径一致（BUG-1759）。
|
||
functional_profile = functional_benefics.derive_functional_benefic_malefic(
|
||
chart["ascendant"].get("sign"),
|
||
moon_paksha=functional_benefics.moon_paksha_from_planets(chart.get("planets")),
|
||
)
|
||
narayana_antardasha_by_md = _narayana_antardasha_table(narayana_periods, planet_longitudes)
|
||
|
||
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,
|
||
"functional_profile": functional_profile,
|
||
"narayana_antardasha_by_md": narayana_antardasha_by_md,
|
||
}
|
||
|
||
|
||
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)
|
||
contexts = [{**build_candidate_static_context(request, candidate), **position}
|
||
for candidate, position in zip(candidate_datetimes, positions)]
|
||
_assert_single_profile_run(contexts)
|
||
return contexts
|
||
|
||
|
||
def _assert_single_profile_run(contexts: Sequence[dict[str, Any]]) -> str | None:
|
||
"""K3 (占星师 2026-11-08 第五轮逐项答复): one scoring run, one functional table."""
|
||
return functional_benefics.assert_single_functional_profile(
|
||
[context.get("functional_profile") for context in contexts], run="rectification")
|
||
|
||
|
||
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] = []
|
||
layer_points: dict[str, dict[str, float]] = {}
|
||
|
||
for event in independent_events(request["events"])[0]:
|
||
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,
|
||
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"),
|
||
context.get("narayana_antardasha_by_md"),
|
||
)
|
||
except (KeyError, TypeError, ValueError):
|
||
missing_layers.append("Narayana_MD_AD")
|
||
continue
|
||
if narayana[0] is None or narayana[1] is None:
|
||
# BUG-1759: a blocked source-version step lands here (mandatory layer missing), never the old algorithm.
|
||
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,
|
||
functional=context.get("functional_profile"),
|
||
))
|
||
# Comparability policy (占星师 2026-11-08 验收清单 §6): points of each layer actually computed.
|
||
event_layers = layer_points.setdefault(event["id"], {"dasha_varga_core": evidence[-1]["points"]})
|
||
transit_rules = _controlled_transit_rules(
|
||
request,
|
||
event,
|
||
ascendant_index,
|
||
DOMAIN_CONFIG[event["domain"]][1],
|
||
context.get("_transit_chart_cache"),
|
||
)
|
||
transit_points = 0.25 * len(transit_rules) * precision_weight(event["precision"])
|
||
event_layers["controlled_transit"] = transit_points
|
||
if transit_rules:
|
||
evidence[-1]["rule_ids"].extend(transit_rules)
|
||
evidence[-1]["points"] = round(evidence[-1]["points"] + transit_points, 4)
|
||
av_rules, av_points = _ashtakavarga_auxiliary(
|
||
chart,
|
||
ascendant_index,
|
||
DOMAIN_CONFIG[event["domain"]][1],
|
||
context.get("ashtakavarga_result"),
|
||
)
|
||
if context.get("ashtakavarga_result") is not None:
|
||
event_layers["Ashtakavarga"] = av_points * precision_weight(event["precision"])
|
||
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 context.get("shadbala_result") is not None:
|
||
event_layers["Shadbala"] = shadbala_points * precision_weight(event["precision"])
|
||
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]
|
||
)),
|
||
"layer_points": layer_points,
|
||
}
|
||
|
||
|
||
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)
|
||
# A repeated fact (same domain, date and precision) is the same test, not independent support (§6).
|
||
events, duplicate_ids = independent_events(request["events"])
|
||
leave_one_out = _leave_one_event_out(rows, events)
|
||
result = adjudicate_candidate_rows(
|
||
rows,
|
||
event_count=len(events),
|
||
domain_count=len({event["domain"] for event in events}),
|
||
request_fingerprint=fingerprint,
|
||
canonical_input_hash=input_hash,
|
||
calculation_contract=input_contract,
|
||
leave_one_event_out=leave_one_out,
|
||
)
|
||
result["duplicate_event_ids"] = duplicate_ids
|
||
return result
|
||
|
||
|
||
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)
|
||
_assert_single_profile_run(contexts)
|
||
transit_chart_cache: dict[tuple[Any, ...], dict[str, Any]] = {}
|
||
return [
|
||
_candidate_row(request, {**context, "_transit_chart_cache": transit_chart_cache})
|
||
for context in contexts
|
||
]
|