feat(upstream): merge a6f47abd engine, MCP, and orchestrator
Independent Staging Quality Gate / validate (push) Has been cancelled
Independent Staging Quality Gate / publish (push) Has been cancelled

Three-way-merge calculation modules and pl9-export into the product fork while keeping commercial API routes, Raman ayanamsa, and the consultation contract as a keypath superset.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Jesse_Chen
2026-09-03 20:28:57 +08:00
parent 45d132588f
commit f224146348
30 changed files with 17151 additions and 248 deletions
+226 -4
View File
@@ -10,9 +10,20 @@ from zoneinfo import ZoneInfo
import swisseph as swe
try:
from western_chart_engine import _ASPECTS, _PLANETS, _birth_zone, _longitude, _orb_for, _point, build_tropical_natal_chart
from western_chart_engine import _ASPECTS, _PLANETS, _RULERS, _SIGNS, _birth_zone, _longitude, _orb_for, _point, build_tropical_natal_chart
except ImportError: # pragma: no cover - package import path
from scripts.western_chart_engine import _ASPECTS, _PLANETS, _birth_zone, _longitude, _orb_for, _point, build_tropical_natal_chart
from scripts.western_chart_engine import _ASPECTS, _PLANETS, _RULERS, _SIGNS, _birth_zone, _longitude, _orb_for, _point, build_tropical_natal_chart
_PLANETARY_YEARS = {
"sun": 19,
"moon": 25,
"mercury": 20,
"venus": 8,
"mars": 15,
"jupiter": 12,
"saturn": 30,
}
def _target_jd(target_date: str, timezone: str | float | int) -> tuple[float, datetime]:
@@ -183,6 +194,34 @@ def calculate_secondary_progressions(*, target_date: str, **birth: Any) -> dict[
}
def calculate_tertiary_progressed_moon(*, target_date: str, **birth: Any) -> dict[str, Any]:
"""Calculate the tertiary progressed Moon using one day per sidereal lunar month."""
natal_chart = build_tropical_natal_chart(**birth)
target_jd, local = _target_jd(target_date, birth["timezone"])
birth_jd = _birth_jd(**birth)
elapsed_months = (target_jd - birth_jd) / 27.321661
progressed_jd = birth_jd + elapsed_months
planets = _progressed_planets(progressed_jd)
natal_points = {
**natal_chart["natal"]["planets"],
"ascendant": natal_chart["natal"]["angles"]["ascendant"],
"mc": natal_chart["natal"]["angles"]["mc"],
}
return {
"technique": "tertiary_progressed_moon",
"status": "partial",
"method": "one_ephemeris_day_per_sidereal_lunar_month",
"target_date": target_date,
"target_local_time": local.isoformat(),
"elapsed_sidereal_lunar_months": round(elapsed_months, 8),
"progressed_julian_day_ut": round(progressed_jd, 8),
"natal_moon_longitude": natal_chart["natal"]["planets"]["moon"]["longitude"],
"progressed_moon": planets["moon"],
"aspects": _cross_aspects({"moon": planets["moon"]}, natal_points),
"boundary": "Tertiary progressed Moon only; progressed house framework, duration, and interpretation remain separate audited layers.",
}
def calculate_solar_arc_directions(*, target_date: str, **birth: Any) -> dict[str, Any]:
"""Direct natal points by the true arc of the secondary progressed Sun."""
natal_chart = build_tropical_natal_chart(**birth)
@@ -355,6 +394,24 @@ def calculate_lunar_return(*, start_date: str, **birth: Any) -> dict[str, Any]:
}
def calculate_lunar_return_series(*, target_date: str, months: int = 1, **birth: Any) -> dict[str, Any]:
"""Return the next exact lunar return from each of the prior calendar-month anchors."""
if not 1 <= int(months) <= 12:
raise ValueError("months must be between 1 and 12")
target = datetime.fromisoformat(target_date[:10])
returns = []
for offset in range(int(months) - 1, -1, -1):
anchor = target - timedelta(days=31 * offset)
returns.append(calculate_lunar_return(start_date=anchor.date().isoformat(), **birth))
return {
"technique": "lunar_return_series",
"target_date": target_date,
"months": int(months),
"returns": returns,
"boundary": "Calendar-month anchor series; each item is a native exact lunar return calculation.",
}
def calculate_transit_duration_scan(*, start_date: str, end_date: str, max_days: int = 370, **birth: Any) -> dict[str, Any]:
"""Scan daily transit-to-natal aspect activity and group consecutive windows."""
start = datetime.fromisoformat(start_date)
@@ -400,6 +457,20 @@ def calculate_transit_duration_scan(*, start_date: str, end_date: str, max_days:
"end_date": final_date,
"min_orb": round(row["min_orb"], 6),
})
windows = sorted(windows, key=lambda row: (row["start_date"], row["min_orb"], row["transit_planet"]))
exact_hit_timeline = [
{
"layer": "transits",
"target": row["natal_point"],
"aspect": row["aspect"],
"window_start": row["start_date"],
"exact_date": row["start_date"] if row["start_date"] == row["end_date"] else row["end_date"],
"window_end": row["end_date"],
"transit_planet": row["transit_planet"],
"min_orb": row["min_orb"],
}
for row in windows
]
return {
"technique": "transit_duration_scan",
"status": "used",
@@ -408,7 +479,8 @@ def calculate_transit_duration_scan(*, start_date: str, end_date: str, max_days:
"end_date": end_date,
"days_scanned": days,
"daily_hits": daily_hits,
"windows": sorted(windows, key=lambda row: (row["start_date"], row["min_orb"], row["transit_planet"])),
"windows": windows,
"exact_hit_timeline": exact_hit_timeline,
"boundary": "Daily scan only; exact ingress/egress times require sub-daily root finding.",
}
@@ -467,19 +539,152 @@ def calculate_parans_status(*, target_date: str | None = None, **birth: Any) ->
}
def calculate_annual_profection(*, target_date: str, **birth: Any) -> dict[str, Any]:
"""Calculate annual profection house, sign, and yearly ruler from the tropical ascendant."""
natal_chart = build_tropical_natal_chart(**birth)
target_local = datetime.fromisoformat(target_date[:10])
birth_month = int(birth["month"])
birth_day = int(birth["day"])
years_elapsed = target_local.year - int(birth["year"])
if (target_local.month, target_local.day) < (birth_month, birth_day):
years_elapsed -= 1
asc_sign = natal_chart["natal"]["ascendant"]["sign"]
asc_index = _SIGNS.index(asc_sign)
profected_house = years_elapsed % 12 + 1
profected_sign = _SIGNS[(asc_index + years_elapsed) % 12]
year_lord = _RULERS[profected_sign]
year_lord_natal = natal_chart["natal"]["planets"][year_lord]
return {
"technique": "annual_profection",
"status": "used",
"target_date": target_date,
"years_elapsed": years_elapsed,
"natal_ascendant_sign": asc_sign,
"profected_house": profected_house,
"profected_sign": profected_sign,
"year_lord": year_lord,
"year_lord_natal_house": year_lord_natal.get("house"),
"year_lord_natal_sign": year_lord_natal.get("sign"),
"boundary": "Annual profection house/sign activation only; interpretation and event adjudication remain separate audited layers.",
}
def _lot_longitude(*, ascendant: float, sun: float, moon: float, is_day_chart: bool, lot: str) -> float:
lot_key = lot.lower()
if lot_key == "fortune":
raw = ascendant + (moon - sun) if is_day_chart else ascendant + (sun - moon)
elif lot_key == "spirit":
raw = ascendant + (sun - moon) if is_day_chart else ascendant + (moon - sun)
else:
raise ValueError("lot must be spirit or fortune")
return _longitude(raw)
def _period_years_for_sign(sign: str) -> int:
return _PLANETARY_YEARS[_RULERS[sign]]
def _build_release_periods(start_sign_index: int, start_date: datetime, levels: int) -> list[dict[str, Any]]:
periods: list[dict[str, Any]] = []
current_start = start_date
for offset in range(12):
sign = _SIGNS[(start_sign_index + offset) % 12]
years = _period_years_for_sign(sign)
current_end = current_start + timedelta(days=round(years * 365.242189))
period = {
"level": 1,
"sign": sign,
"years": years,
"start_date": current_start.date().isoformat(),
"end_date": (current_end - timedelta(days=1)).date().isoformat(),
}
if levels >= 2:
subperiods = []
sub_start = current_start
for sub_offset in range(12):
sub_sign = _SIGNS[(start_sign_index + offset + sub_offset) % 12]
sub_years = years * _period_years_for_sign(sub_sign) / 12.0
sub_end = sub_start + timedelta(days=round(sub_years * 365.242189))
subperiods.append({
"level": 2,
"sign": sub_sign,
"years": round(sub_years, 6),
"start_date": sub_start.date().isoformat(),
"end_date": (sub_end - timedelta(days=1)).date().isoformat(),
})
sub_start = sub_end
period["subperiods"] = subperiods
periods.append(period)
current_start = current_end
return periods
def calculate_zodiacal_release(*, target_date: str, lot: str = "spirit", levels: int = 2, **birth: Any) -> dict[str, Any]:
"""Calculate a bounded L1/L2 zodiacal-release timeline from Fortune or Spirit."""
if levels not in {1, 2}:
raise ValueError("levels must be 1 or 2")
natal_chart = build_tropical_natal_chart(**birth)
natal = natal_chart["natal"]
ascendant = natal["ascendant"]["longitude"]
sun = natal["planets"]["sun"]["longitude"]
moon = natal["planets"]["moon"]["longitude"]
is_day_chart = natal["planets"]["sun"]["house"] in {7, 8, 9, 10, 11, 12}
lot_longitude = _lot_longitude(
ascendant=ascendant,
sun=sun,
moon=moon,
is_day_chart=is_day_chart,
lot=lot,
)
lot_sign = _SIGNS[int(lot_longitude // 30)]
start_date = datetime(
int(birth["year"]),
int(birth["month"]),
int(birth["day"]),
)
periods = _build_release_periods(_SIGNS.index(lot_sign), start_date, levels)
active_period = next(
(
period
for period in periods
if period["start_date"] <= target_date <= period["end_date"]
),
None,
)
return {
"technique": "zodiacal_release",
"status": "partial",
"target_date": target_date,
"lot": lot.lower(),
"lot_longitude": round(lot_longitude, 6),
"lot_sign": lot_sign,
"day_night_basis": "day_chart" if is_day_chart else "night_chart",
"levels": levels,
"periods": periods,
"active_period": active_period,
"boundary": "L1/L2 sign-period release scaffold from the Lot of Spirit/Fortune only; loosing-of-the-bond, peak periods, angularity weighting, and interpretation remain separate audited layers.",
}
def build_timing_techniques(
*,
transit_date: str | None = None,
solar_return_year: int | None = None,
secondary_progression_date: str | None = None,
tertiary_progressed_moon_date: str | None = None,
solar_arc_date: str | None = None,
converse_secondary_progression_date: str | None = None,
converse_solar_arc_date: str | None = None,
midpoint_date: str | None = None,
lunar_return_start_date: str | None = None,
lunar_return_months: int = 1,
duration_scan_start_date: str | None = None,
duration_scan_end_date: str | None = None,
parans_date: str | None = None,
profection_date: str | None = None,
zodiacal_release_date: str | None = None,
zodiacal_release_lot: str = "spirit",
zodiacal_release_levels: int = 2,
**birth: Any,
) -> dict[str, Any]:
"""Materialize only the requested, independently auditable timing layers."""
@@ -492,6 +697,10 @@ def build_timing_techniques(
techniques["secondary_progressions"] = calculate_secondary_progressions(
target_date=secondary_progression_date, **birth
)
if tertiary_progressed_moon_date:
techniques["tertiary_progressed_moon"] = calculate_tertiary_progressed_moon(
target_date=tertiary_progressed_moon_date, **birth
)
if solar_arc_date:
techniques["solar_arc_directions"] = calculate_solar_arc_directions(target_date=solar_arc_date, **birth)
if converse_secondary_progression_date:
@@ -505,7 +714,11 @@ def build_timing_techniques(
if midpoint_date:
techniques["midpoints"] = calculate_midpoints(target_date=midpoint_date, **birth)
if lunar_return_start_date:
techniques["lunar_return"] = calculate_lunar_return(start_date=lunar_return_start_date, **birth)
techniques["lunar_return"] = calculate_lunar_return_series(
target_date=lunar_return_start_date,
months=int(lunar_return_months),
**birth,
)
if duration_scan_start_date and duration_scan_end_date:
techniques["transit_duration_scan"] = calculate_transit_duration_scan(
start_date=duration_scan_start_date,
@@ -514,4 +727,13 @@ def build_timing_techniques(
)
if parans_date:
techniques["parans"] = calculate_parans_status(target_date=parans_date, **birth)
if profection_date:
techniques["annual_profection"] = calculate_annual_profection(target_date=profection_date, **birth)
if zodiacal_release_date:
techniques["zodiacal_release"] = calculate_zodiacal_release(
target_date=zodiacal_release_date,
lot=zodiacal_release_lot,
levels=int(zodiacal_release_levels),
**birth,
)
return techniques