Files
Jyotisha/scripts/narayana_dasha.py
T
Jesse_Chen f224146348
Independent Staging Quality Gate / validate (push) Has been cancelled
Independent Staging Quality Gate / publish (push) Has been cancelled
feat(upstream): merge a6f47abd engine, MCP, and orchestrator
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>
2026-09-03 20:28:57 +08:00

1002 lines
38 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Narayana DashaRishi Dasha / Padakrama Dasha)计算模块
Jyotish Vedic Astrology Skill — v6.0.20
Narayana Dasha 是 Parashara 传授的 Rishi-based 大运系统,与 Vimshottari 互补:
- Vimshottari: Nakshatra-based, 120年固定周期
- Narayana: Rashi-based, 从 Lagna 起按黄道序推进,周期可变
算法(Lagna-based variant, BPHS Chapter 48:
1. 从 Lagna 星座开始
2. 每个星座的大运年数 = 从该星座数到其守护星所在星座的步数(含起点,不含终点)
3. 若守护星在本星座 → 12年
4. 按黄道顺序推进 12 星座,然后循环
5. Antardasha 按各星座年数比例分配
依赖: 需要 planet_lons 和 houses 数据(可从引擎传入)
"""
from typing import Dict, List, Optional, Tuple
import math
# ── 常量 ──────────────────────────────────────────────────────────
SIGNS = ['Aries','Taurus','Gemini','Cancer','Leo','Virgo',
'Libra','Scorpio','Sagittarius','Capricorn','Aquarius','Pisces']
SIGN_LORDS = {
'Aries':'Mars','Taurus':'Venus','Gemini':'Mercury','Cancer':'Moon',
'Leo':'Sun','Virgo':'Mercury','Libra':'Venus','Scorpio':'Mars',
'Sagittarius':'Jupiter','Capricorn':'Saturn','Aquarius':'Saturn','Pisces':'Jupiter',
}
SIGN_INDEX = {s:i for i,s in enumerate(SIGNS)}
# Planet name → sign index lookup
PLANET_SIGN_INDEX = {
'Sun': 4, 'Moon': 3, 'Mars': (0,7), 'Mercury': (2,5),
'Jupiter': (8,11), 'Venus': (1,6), 'Saturn': (9,10),
'Rahu': 10, 'Ketu': 7, # traditional assignments
}
_NARAYANA_ODD_FOOTED_SIGNS = {0, 1, 2, 6, 7, 8}
_NARAYANA_DURATION_PROFILES = {
"legacy_forward_v0",
"jaimini_odd_footed_v1",
"jaimini_odd_footed_dignity_v2",
}
_NARAYANA_GENERAL_SEQUENCE = {
0: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
1: [1, 8, 3, 10, 5, 0, 7, 2, 9, 4, 11, 6],
2: [2, 10, 6, 5, 1, 9, 8, 4, 0, 11, 7, 3],
3: [3, 2, 1, 0, 11, 10, 9, 8, 7, 6, 5, 4],
4: [4, 9, 2, 7, 0, 5, 10, 3, 8, 1, 6, 11],
5: [5, 9, 1, 2, 6, 10, 11, 3, 7, 8, 0, 4],
6: [6, 7, 8, 9, 10, 11, 0, 1, 2, 3, 4, 5],
7: [7, 2, 9, 4, 11, 6, 1, 8, 3, 10, 5, 0],
8: [8, 4, 0, 11, 7, 3, 2, 10, 6, 5, 1, 9],
9: [9, 8, 7, 6, 5, 4, 3, 2, 1, 0, 11, 10],
10: [10, 3, 8, 1, 6, 11, 4, 9, 2, 7, 0, 5],
11: [11, 3, 7, 0, 4, 8, 1, 5, 9, 2, 6, 10],
}
_NARAYANA_RATH_GENERAL_SEQUENCE = {
0: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
1: [1, 8, 3, 10, 5, 0, 7, 2, 9, 4, 11, 6],
2: [2, 10, 6, 5, 1, 9, 8, 4, 0, 11, 7, 3],
3: [3, 2, 1, 0, 11, 10, 9, 8, 7, 6, 5, 4],
4: [4, 9, 2, 7, 0, 5, 10, 3, 8, 1, 6, 11],
5: [5, 9, 1, 2, 6, 10, 11, 3, 7, 8, 0, 4],
6: [6, 7, 8, 9, 10, 11, 0, 1, 2, 3, 4, 5],
7: [7, 2, 9, 4, 11, 6, 1, 8, 3, 10, 5, 0],
8: [8, 4, 0, 11, 7, 3, 2, 10, 6, 5, 1, 9],
9: [9, 8, 7, 6, 5, 4, 3, 2, 1, 0, 11, 10],
10: [10, 3, 8, 1, 6, 11, 4, 9, 2, 7, 0, 5],
11: [11, 3, 7, 8, 0, 4, 5, 9, 1, 2, 6, 10],
}
_NARAYANA_SEQUENCE_PROFILES = {
"legacy_zodiacal_v0",
"jaimini_general_order_v1",
"rath_general_table10_v1",
"rath_saturn_table11_v1",
}
_NARAYANA_SEED_PROFILES = {"legacy_lagna_v0", "jaimini_lagna_seventh_strength_v1"}
_NARAYANA_DUAL_LORD_PROFILES = {"legacy_primary_lord_v0", "jaimini_dual_lord_source_v1"}
_NARAYANA_ANTARDASHA_PROFILES = {
"legacy_weighted_v0",
"parashara_equal_v1",
"pl9_observed_stronger_sign_v1",
"pl9_observed_direction_v1",
"pl9_observed_direction_tie_break_v1",
}
_NARAYANA_AD_ORDER_TABLE13 = {
0: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
1: [1, 0, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2],
2: [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 0, 1],
3: [3, 2, 1, 0, 11, 10, 9, 8, 7, 6, 5, 4],
4: [4, 5, 6, 7, 8, 9, 10, 11, 0, 1, 2, 3],
5: [5, 4, 3, 2, 1, 0, 11, 10, 9, 8, 7, 6],
6: [6, 7, 8, 9, 10, 11, 0, 1, 2, 3, 4, 5],
7: [7, 6, 5, 4, 3, 2, 1, 0, 11, 10, 9, 8],
8: [8, 9, 10, 11, 0, 1, 2, 3, 4, 5, 6, 7],
9: [9, 8, 7, 6, 5, 4, 3, 2, 1, 0, 11, 10],
10: [10, 11, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9],
11: [11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0],
}
_NARAYANA_AD_KETU_TABLE14 = {
0: [0, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
1: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 0],
2: [2, 1, 0, 11, 10, 9, 8, 7, 6, 5, 4, 3],
3: [3, 4, 5, 6, 7, 8, 9, 10, 11, 0, 1, 2],
4: [4, 3, 2, 1, 0, 11, 10, 9, 8, 7, 6, 5],
5: [5, 6, 7, 8, 9, 10, 11, 0, 1, 2, 3, 4],
6: [6, 5, 4, 3, 2, 1, 0, 11, 10, 9, 8, 7],
# Source/OCR line repeats Cancer/Gemini at the tail. Keep the normalized
# reversal rule here; the ledger records the printed anomaly separately.
7: [7, 8, 9, 10, 11, 0, 1, 2, 3, 4, 5, 6],
8: [8, 7, 6, 5, 4, 3, 2, 1, 0, 11, 10, 9],
9: [9, 10, 11, 0, 1, 2, 3, 4, 5, 6, 7, 8],
10: [10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0, 11],
11: [11, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
}
_NARAYANA_AD_ORDER_PROFILES = {
"rath_table13_v1",
"rath_ketu_table14_v1",
"rath_saturn_table11_v1",
}
_MOVABLE_SIGNS = {0, 3, 6, 9}
_FIXED_SIGNS = {1, 4, 7, 10}
_NARAYANA_DIGNITY = {
"Sun": {"exalted": {0}, "debilitated": {6}},
"Moon": {"exalted": {1}, "debilitated": {7}},
"Mars": {"exalted": {9}, "debilitated": {3}},
"Mercury": {"exalted": {5}, "debilitated": {11}},
"Jupiter": {"exalted": {3}, "debilitated": {9}},
"Venus": {"exalted": {11}, "debilitated": {5}},
"Saturn": {"exalted": {6}, "debilitated": {0}},
}
_NARAYANA_NODE_MOOLATRIKONA = {"Rahu": {5}, "Ketu": {7}}
_NARAYANA_NODE_OWN_SIGNS = {"Rahu": {10}, "Ketu": {7}}
# =========================================================================
# 核心计算
# =========================================================================
def _get_planet_sign(planet_name: str, planet_lons: Dict[str, float]) -> Optional[int]:
"""获取行星所在星座索引"""
lon = planet_lons.get(planet_name)
if lon is None:
return None
return int(lon / 30) % 12
def _count_signs_forward(src_idx: int, dest_idx: int) -> int:
"""从 src 数到 dest(含起点,不含终点),顺黄道方向。
若 src == dest → 12 年(守护星在本星座)"""
if src_idx == dest_idx:
return 12
count = (dest_idx - src_idx + 12) % 12
return count # 已含起点
def build_narayana_sign_sequence(
seed_sign_idx: int,
*,
profile: str = "legacy_zodiacal_v0",
) -> List[int]:
"""Return a first-cycle sign order for an explicit Narayana sequence profile."""
if profile not in _NARAYANA_SEQUENCE_PROFILES:
raise ValueError(f"unsupported Narayana sequence profile: {profile}")
if not 0 <= seed_sign_idx < 12:
raise ValueError("seed sign index must be in 0..11")
if profile == "legacy_zodiacal_v0":
return [(seed_sign_idx + offset) % 12 for offset in range(12)]
if profile == "rath_general_table10_v1":
return list(_NARAYANA_RATH_GENERAL_SEQUENCE[seed_sign_idx])
if profile == "rath_saturn_table11_v1":
return [(seed_sign_idx + offset) % 12 for offset in range(12)]
return list(_NARAYANA_GENERAL_SEQUENCE[seed_sign_idx])
def build_narayana_antardasha_order_sequence(
start_sign_idx: int,
*,
profile: str = "rath_table13_v1",
) -> List[int]:
"""Return Sanjay Rath source-table Antardasha sign order."""
if profile not in _NARAYANA_AD_ORDER_PROFILES:
raise ValueError(f"unsupported Narayana Antardasha order profile: {profile}")
if not 0 <= start_sign_idx < 12:
raise ValueError("start sign index must be in 0..11")
if profile == "rath_table13_v1":
return list(_NARAYANA_AD_ORDER_TABLE13[start_sign_idx])
if profile == "rath_ketu_table14_v1":
return list(_NARAYANA_AD_KETU_TABLE14[start_sign_idx])
return [(start_sign_idx + offset) % 12 for offset in range(12)]
def build_narayana_pratyantardasha_order_sequence(
start_sign_idx: int,
*,
profile: str = "rath_inherit_ad_table13_v1",
) -> List[int]:
"""Return source-attested PD order by inheriting the Rath AD rolling rule."""
if profile == "rath_inherit_ad_table13_v1":
return build_narayana_antardasha_order_sequence(
start_sign_idx,
profile="rath_table13_v1",
)
if profile == "rath_inherit_ketu_ad_table14_v1":
return build_narayana_antardasha_order_sequence(
start_sign_idx,
profile="rath_ketu_table14_v1",
)
raise ValueError(f"unsupported Narayana Pratyantardasha order profile: {profile}")
def subdivide_narayana_equal_order(
parent_period: Dict,
*,
start_sign_idx: int,
order_profile: str,
level: str,
parent_key: str,
parent_name: str,
) -> List[Dict]:
"""Split a Narayana parent period into twelve equal source-profile rows."""
if not 0 <= start_sign_idx < 12:
raise ValueError("start sign index must be in 0..11")
total_years = float(parent_period.get("years", 0))
parent_start = float(parent_period.get("start_age", 0))
parent_end = float(parent_period.get("end_age", parent_start + total_years))
if level == "PD":
sequence = build_narayana_pratyantardasha_order_sequence(
start_sign_idx,
profile=order_profile,
)
else:
sequence = build_narayana_antardasha_order_sequence(
start_sign_idx,
profile=order_profile,
)
rows = []
for index, sign_idx in enumerate(sequence):
start_age = parent_start + total_years * index / 12
end_age = parent_end if index == 11 else parent_start + total_years * (index + 1) / 12
rows.append({
"level": level,
"sign": SIGNS[sign_idx],
"sign_idx": sign_idx,
"lord": SIGN_LORDS[SIGNS[sign_idx]],
"years": round(max(0.0, end_age - start_age), 6),
"start_age": round(start_age, 6),
"end_age": round(end_age, 6),
"sequence_index": index,
"order_profile": order_profile,
"start_sign_source": "explicit_source_profile_start_sign",
parent_key: parent_name,
})
return rows
def _has_narayana_sequence_exception(
seed_sign_idx: int,
planet_lons: Dict[str, float],
) -> bool:
return any(
_get_planet_sign(planet, planet_lons) == seed_sign_idx
for planet in ("Saturn", "Ketu")
)
def _rasi_aspects(source_sign_idx: int, target_sign_idx: int) -> bool:
"""Return Jaimini Rasi Drishti between two signs."""
if source_sign_idx in _MOVABLE_SIGNS:
return target_sign_idx in _FIXED_SIGNS and target_sign_idx != (source_sign_idx + 1) % 12
if source_sign_idx in _FIXED_SIGNS:
return target_sign_idx in _MOVABLE_SIGNS and target_sign_idx != (source_sign_idx - 1) % 12
return target_sign_idx in {2, 5, 8, 11} and target_sign_idx != source_sign_idx
def select_narayana_seed_sign(
lagna_sign_idx: int,
planet_lons: Dict[str, float],
) -> Dict:
"""Select the stronger of Lagna and seventh using the first two rule levels.
A tie is deliberately returned as blocked: later classical strength rules are
not yet implemented and must not be silently substituted with a preference.
"""
if not 0 <= lagna_sign_idx < 12:
raise ValueError("lagna sign index must be in 0..11")
candidate_indices = (lagna_sign_idx, (lagna_sign_idx + 6) % 12)
candidates = {}
for sign_idx in candidate_indices:
lord = SIGN_LORDS[SIGNS[sign_idx]]
occupants = sorted(
planet
for planet in planet_lons
if _get_planet_sign(planet, planet_lons) == sign_idx
)
aspect_factors = []
for planet in dict.fromkeys(("Jupiter", "Mercury", lord)):
planet_sign_idx = _get_planet_sign(planet, planet_lons)
if planet_sign_idx is not None and _rasi_aspects(planet_sign_idx, sign_idx):
aspect_factors.append(planet)
candidates[sign_idx] = {
"sign": SIGNS[sign_idx],
"lord": lord,
"occupants": occupants,
"occupant_count": len(occupants),
"aspect_factors": aspect_factors,
"aspect_factor_count": len(aspect_factors),
}
lagna_candidate = candidates[lagna_sign_idx]
seventh_sign_idx = (lagna_sign_idx + 6) % 12
seventh_candidate = candidates[seventh_sign_idx]
lagna_score = (lagna_candidate["occupant_count"], lagna_candidate["aspect_factor_count"])
seventh_score = (seventh_candidate["occupant_count"], seventh_candidate["aspect_factor_count"])
if lagna_score == seventh_score:
return {
"status": "blocked",
"reason": "later_narayana_sign_strength_rules_required",
"profile": "jaimini_lagna_seventh_strength_v1",
"lagna_sign_idx": lagna_sign_idx,
"seventh_sign_idx": seventh_sign_idx,
"candidates": candidates,
}
selected_sign_idx = lagna_sign_idx if lagna_score > seventh_score else seventh_sign_idx
return {
"status": "selected",
"profile": "jaimini_lagna_seventh_strength_v1",
"selected_sign_idx": selected_sign_idx,
"selected_sign": SIGNS[selected_sign_idx],
"selection_levels": ["occupant_count", "jupiter_mercury_or_lord_rasi_aspect"],
"candidates": candidates,
}
def calculate_narayana_duration(
sign_idx: int,
lord_sign_idx: int,
*,
profile: str = "legacy_forward_v0",
) -> int:
"""Return a Narayana sign-period duration under an explicit rule profile."""
if profile not in _NARAYANA_DURATION_PROFILES:
raise ValueError(f"unsupported Narayana duration profile: {profile}")
if not 0 <= sign_idx < 12 or not 0 <= lord_sign_idx < 12:
raise ValueError("sign indices must be in 0..11")
if sign_idx == lord_sign_idx:
return 12
if profile == "legacy_forward_v0":
return _count_signs_forward(sign_idx, lord_sign_idx)
if sign_idx in _NARAYANA_ODD_FOOTED_SIGNS:
return (lord_sign_idx - sign_idx) % 12
return (sign_idx - lord_sign_idx) % 12
def calculate_narayana_period_years(
sign_idx: int,
lord: str,
lord_sign_idx: int,
*,
profile: str = "legacy_forward_v0",
) -> tuple[int, int]:
"""Return a period length and dignity adjustment under an explicit profile."""
base_years = calculate_narayana_duration(sign_idx, lord_sign_idx, profile=profile)
if profile != "jaimini_odd_footed_dignity_v2":
return base_years, 0
dignity = _NARAYANA_DIGNITY.get(lord, {})
adjustment = 1 if lord_sign_idx in dignity.get("exalted", set()) else 0
if lord_sign_idx in dignity.get("debilitated", set()):
adjustment = -1
return max(0, min(12, base_years + adjustment)), adjustment
def _narayana_dual_lord_strength(
lord: str,
lord_sign_idx: int,
planet_lons: Dict[str, float],
) -> tuple[int, int, int]:
occupants = sum(
1 for planet in planet_lons if _get_planet_sign(planet, planet_lons) == lord_sign_idx
)
sign_lord = SIGN_LORDS[SIGNS[lord_sign_idx]]
aspect_factors = sum(
1
for planet in dict.fromkeys(("Jupiter", "Mercury", sign_lord))
if (planet_sign_idx := _get_planet_sign(planet, planet_lons)) is not None
and _rasi_aspects(planet_sign_idx, lord_sign_idx)
)
dignity = 2 if lord_sign_idx in _NARAYANA_NODE_MOOLATRIKONA.get(lord, set()) else 0
if lord_sign_idx in _NARAYANA_NODE_OWN_SIGNS.get(lord, set()):
dignity = max(dignity, 1)
return occupants, aspect_factors, dignity
def _resolve_narayana_lord(
sign_idx: int,
planet_lons: Dict[str, float],
*,
dual_lord_profile: str,
) -> str:
if dual_lord_profile not in _NARAYANA_DUAL_LORD_PROFILES:
raise ValueError(f"unsupported Narayana dual-lord profile: {dual_lord_profile}")
primary_lord = SIGN_LORDS[SIGNS[sign_idx]]
dual_lords = {7: ("Mars", "Ketu"), 10: ("Saturn", "Rahu")}.get(sign_idx)
if dual_lords is None or dual_lord_profile == "legacy_primary_lord_v0":
return primary_lord
first, second = dual_lords
first_sign = _get_planet_sign(first, planet_lons)
second_sign = _get_planet_sign(second, planet_lons)
if first_sign is None or second_sign is None:
return primary_lord
if first_sign == sign_idx and second_sign != sign_idx:
return second
if second_sign == sign_idx and first_sign != sign_idx:
return first
if first_sign == second_sign:
return first
first_strength = _narayana_dual_lord_strength(first, first_sign, planet_lons)
second_strength = _narayana_dual_lord_strength(second, second_sign, planet_lons)
if first_strength > second_strength:
return first
if second_strength > first_strength:
return second
first_years = calculate_narayana_duration(sign_idx, first_sign, profile="jaimini_odd_footed_v1")
second_years = calculate_narayana_duration(sign_idx, second_sign, profile="jaimini_odd_footed_v1")
return first if first_years >= second_years else second
def calc_narayana_mahadasha(
lagna_sign_idx: int,
planet_lons: Dict[str, float],
start_year: float = 0.0,
) -> List[Dict]:
"""
计算 Narayana Dasha 大运序列(第1周期)。
参数:
lagna_sign_idx: Lagna 星座索引 (0-11)
planet_lons: 行星经度字典 {planet_name: longitude_deg}
start_year: 起始年份偏移(默认 0 = 出生时)
返回:
list of dasha periods, each: {
'sign': str, 'sign_idx': int, 'lord': str,
'years': int, 'start_age': float, 'end_age': float,
}
"""
periods = []
cum_years = start_year
for i in range(12):
sign_idx = (lagna_sign_idx + i) % 12
sign = SIGNS[sign_idx]
lord = SIGN_LORDS[sign]
# 获取 lord 所在的星座
lord_sign_idx = _get_planet_sign(lord, planet_lons)
if lord_sign_idx is None:
# fallback: 如果找不到 lord 位置,用 lord 的 Moolatrikona 或自身星座
lord_sign_idx = sign_idx # 保守假设:lord 在自己星座
years = _count_signs_forward(sign_idx, lord_sign_idx)
periods.append({
'sign': sign,
'sign_idx': sign_idx,
'lord': lord,
'lord_in_sign': SIGNS[lord_sign_idx],
'lord_sign_idx': lord_sign_idx,
'years': years,
'count_from_to': f'{sign}({sign_idx})→{SIGNS[lord_sign_idx]}({lord_sign_idx})',
'start_age': round(cum_years, 2),
'end_age': round(cum_years + years, 2),
})
cum_years += years
return periods
def calc_narayana_antardasha(
mahadasha_periods: List[Dict],
md_sign_idx: int,
*,
planet_lons: Optional[Dict[str, float]] = None,
profile: str = "legacy_weighted_v0",
dual_lord_profile: str = "legacy_primary_lord_v0",
md_period: Optional[Dict] = None,
) -> List[Dict]:
"""
计算给定 Mahadasha 的 Antardasha 子周期。
参数:
mahadasha_periods: calc_narayana_mahadasha 的返回值
md_sign_idx: Mahadasha 星座索引
返回:
list of antardasha periods
"""
# 找到对应的大运
md = None
for p in mahadasha_periods:
if p['sign_idx'] == md_sign_idx:
md = p
break
if md is None:
return []
if profile == "legacy_weighted_v0" and len({p['sign_idx'] for p in mahadasha_periods}) != len(mahadasha_periods):
raise ValueError("legacy_weighted_v0 does not support repeated MD signs across cycles")
if profile in {"parashara_equal_v1", "pl9_observed_stronger_sign_v1", "pl9_observed_direction_v1", "pl9_observed_direction_tie_break_v1"}:
if not planet_lons:
raise ValueError(f"{profile} requires planet_lons")
selection = select_narayana_seed_sign(md_sign_idx, planet_lons)
if selection["status"] != "selected":
raise ValueError(selection["reason"])
stronger_sign_idx = selection["selected_sign_idx"]
lord = _resolve_narayana_lord(stronger_sign_idx, planet_lons, dual_lord_profile=dual_lord_profile)
if profile in {"parashara_equal_v1", "pl9_observed_direction_v1", "pl9_observed_direction_tie_break_v1"}:
start_sign_idx = _get_planet_sign(lord, planet_lons)
if start_sign_idx is None:
raise ValueError(f"{profile} requires stronger-sign lord longitude")
start_sign_source = "lord_of_stronger_dasha_or_seventh_sign"
evidence_status = "sourced" if profile == "parashara_equal_v1" else "observed_pl9_control_case_only"
else:
start_sign_idx = stronger_sign_idx
start_sign_source = "pl9_observed_stronger_dasha_or_seventh_sign"
evidence_status = "parameter_sensitive"
zodiacal_direction = md_sign_idx % 2 == 0
if profile in {"pl9_observed_direction_v1", "pl9_observed_direction_tie_break_v1"}:
zodiacal_direction = md_sign_idx in {3, 4, 9, 10}
return _subdivide_narayana_period_equal(
parent_period=md,
start_sign_idx=start_sign_idx,
zodiacal_direction=zodiacal_direction,
stronger_sign_idx=stronger_sign_idx,
stronger_sign_lord=lord,
profile=profile,
start_sign_source=start_sign_source,
evidence_status=evidence_status,
)
return _subdivide_narayana_period(
mahadasha_periods=mahadasha_periods,
parent_period=md,
start_sign_idx=md_sign_idx,
parent_key='parent_md',
parent_name=SIGNS[md_sign_idx],
)
def _subdivide_narayana_period_equal(
*,
parent_period: Dict,
start_sign_idx: int,
zodiacal_direction: bool,
stronger_sign_idx: int,
stronger_sign_lord: str,
profile: str,
start_sign_source: str,
evidence_status: str,
) -> List[Dict]:
"""Apply BPHS's twelve equal-sign Antardasha rule for a Rashi Dasha."""
total_years = float(parent_period.get("years", 0))
parent_start = float(parent_period.get("start_age", 0))
parent_end = float(parent_period.get("end_age", parent_start + total_years))
direction = 1 if zodiacal_direction else -1
sub_periods = []
for index in range(12):
sign_idx = (start_sign_idx + direction * index) % 12
start_age = parent_start + total_years * index / 12
end_age = parent_end if index == 11 else parent_start + total_years * (index + 1) / 12
sub_periods.append({
"sign": SIGNS[sign_idx],
"sign_idx": sign_idx,
"lord": SIGN_LORDS[SIGNS[sign_idx]],
"years": round(max(0.0, end_age - start_age), 4),
"start_age": round(start_age, 4),
"end_age": round(end_age, 4),
"parent_md": parent_period.get("sign", SIGNS[parent_period["sign_idx"]]),
"sequence_index": index,
"antardasha_profile": profile,
"start_sign_source": start_sign_source,
"evidence_status": evidence_status,
"stronger_sign_idx": stronger_sign_idx,
"stronger_sign_lord": stronger_sign_lord,
"direction": "zodiacal" if zodiacal_direction else "reverse",
})
return sub_periods
def calc_narayana_pratyantardasha(
mahadasha_periods: List[Dict],
antardasha_period: Dict,
) -> List[Dict]:
"""
计算给定 Antardasha 的 Pratyantardasha 子周期。
返回的 start_age / end_age 与 Mahadasha、Antardasha 使用同一条绝对年龄轴,
方便 get_current_narayana_dasha 直接定位当前周期。
"""
if not antardasha_period:
return []
sign_idx = antardasha_period.get('sign_idx')
if sign_idx is None:
return []
return _subdivide_narayana_period(
mahadasha_periods=mahadasha_periods,
parent_period=antardasha_period,
start_sign_idx=sign_idx,
parent_key='parent_ad',
parent_name=antardasha_period.get('sign', SIGNS[sign_idx]),
)
def _subdivide_narayana_period(
mahadasha_periods: List[Dict],
parent_period: Dict,
start_sign_idx: int,
parent_key: str,
parent_name: str,
) -> List[Dict]:
"""按 Narayana 星座年数权重切分父周期,返回绝对年龄轴上的子周期。"""
if not mahadasha_periods:
return []
period_by_sign = {p['sign_idx']: p for p in mahadasha_periods}
denominator = sum(float(p.get('years', 0)) for p in mahadasha_periods)
if denominator <= 0:
return []
total_years = float(parent_period.get('years', 0))
parent_start = float(parent_period.get('start_age', 0))
parent_end = float(parent_period.get('end_age', parent_start + total_years))
if len(period_by_sign) == 12:
weighted_sequence = [period_by_sign[(start_sign_idx + i) % 12] for i in range(12)]
else:
start_pos = next(
(i for i, p in enumerate(mahadasha_periods) if p.get('sign_idx') == start_sign_idx),
0,
)
weighted_sequence = mahadasha_periods[start_pos:] + mahadasha_periods[:start_pos]
sub_periods = []
cum = parent_start
sequence_len = len(weighted_sequence)
for i, weighted_period in enumerate(weighted_sequence):
sign_idx = weighted_period['sign_idx']
if i == sequence_len - 1:
end_age = parent_end
else:
sub_years_raw = total_years * float(weighted_period.get('years', 0)) / denominator
end_age = cum + sub_years_raw
start_age = cum
years = max(0.0, end_age - start_age)
sub_periods.append({
'sign': SIGNS[sign_idx],
'sign_idx': sign_idx,
'lord': SIGN_LORDS[SIGNS[sign_idx]],
'years': round(years, 4),
'start_age': round(start_age, 4),
'end_age': round(end_age, 4),
parent_key: parent_name,
'sequence_index': i,
})
cum = end_age
return sub_periods
def get_current_narayana_dasha(
mahadasha_periods: List[Dict],
current_age: float,
) -> Dict:
"""
获取当前年龄对应的 Narayana Dasha 周期(Mahadasha + Antardasha)。
返回:
{
'md': {...}, # 当前大运
'ad': {...}, # 当前小运
'pd': {...}, # 当前节运(Pratyantara,简化)
'remaining_years': float,
}
"""
result = {'md': None, 'ad': None, 'pd': None, 'remaining_years': 0}
# 处理多年期(可能跨多个周期)
total_cycle = sum(p['years'] for p in mahadasha_periods)
if total_cycle == 0:
return result
cycle_start = min(float(p.get('start_age', 0)) for p in mahadasha_periods)
age_in_cycle = ((current_age - cycle_start) % total_cycle) + cycle_start
# 找当前 MD
for p in mahadasha_periods:
if p['start_age'] <= age_in_cycle < p['end_age']:
result['md'] = {
'sign': p['sign'],
'sign_idx': p['sign_idx'],
'lord': p['lord'],
'years': p['years'],
'start_age': p['start_age'],
'end_age': p['end_age'],
}
result['remaining_years'] = round(p['end_age'] - age_in_cycle, 2)
# 计算 AD
ads = calc_narayana_antardasha(mahadasha_periods, p['sign_idx'])
for ad in ads:
if ad['start_age'] <= age_in_cycle < ad['end_age']:
result['ad'] = {
'sign': ad['sign'],
'sign_idx': ad['sign_idx'],
'lord': ad['lord'],
'years': ad['years'],
'start_age': ad['start_age'],
'end_age': ad['end_age'],
}
pds = calc_narayana_pratyantardasha(mahadasha_periods, ad)
for pd in pds:
if pd['start_age'] <= age_in_cycle < pd['end_age']:
result['pd'] = {
'sign': pd['sign'],
'sign_idx': pd['sign_idx'],
'lord': pd['lord'],
'years': pd['years'],
'start_age': pd['start_age'],
'end_age': pd['end_age'],
}
break
break
break
return result
def build_narayana_dense_boundaries(
mahadasha_periods: List[Dict],
*,
planet_lons: Optional[Dict[str, float]] = None,
antardasha_profile: str = "legacy_weighted_v0",
dual_lord_profile: str = "legacy_primary_lord_v0",
) -> List[Dict]:
"""Return schema-stable MD/AD Narayana boundaries on the absolute age axis."""
rows: List[Dict] = []
for md_index, md in enumerate(mahadasha_periods, start=1):
md_direction = md.get("direction")
rows.append({
"level": "MD",
"sign": md.get("sign"),
"sign_idx": md.get("sign_idx"),
"lord": md.get("lord"),
"start_age": md.get("start_age"),
"end_age": md.get("end_age"),
"duration_years": md.get("years"),
"duration_days": round(float(md.get("years", 0)) * 365.25, 2),
"direction": md_direction or md.get("sequence_profile"),
"seed_sign": SIGNS[md.get("seed_sign_idx", md.get("sign_idx", 0))],
"profile": {
"duration_profile": md.get("duration_profile"),
"sequence_profile": md.get("sequence_profile"),
"seed_profile": md.get("seed_profile"),
"dual_lord_profile": md.get("dual_lord_profile"),
"antardasha_profile": antardasha_profile,
},
"source_formula_status": "local_md_profiled_v1",
"period_key": md.get("period_key"),
"sequence_index": md_index - 1,
})
try:
antardashas = calc_narayana_antardasha(
mahadasha_periods,
md["sign_idx"],
planet_lons=planet_lons,
profile=antardasha_profile,
dual_lord_profile=dual_lord_profile,
md_period=md,
)
ad_status = "local_ad_profiled_v1"
except ValueError as exc:
rows.append({
"level": "AD",
"sign": None,
"start_age": md.get("start_age"),
"end_age": md.get("end_age"),
"duration_years": None,
"duration_days": None,
"direction": None,
"seed_sign": md.get("sign"),
"profile": {
"duration_profile": md.get("duration_profile"),
"sequence_profile": md.get("sequence_profile"),
"seed_profile": md.get("seed_profile"),
"dual_lord_profile": dual_lord_profile,
"antardasha_profile": antardasha_profile,
},
"source_formula_status": "blocked_antardasha_profile_error",
"blocked_reason": str(exc),
"parent_md": md.get("sign"),
"parent_period_key": md.get("period_key"),
})
continue
for ad_index, ad in enumerate(antardashas, start=1):
rows.append({
"level": "AD",
"sign": ad.get("sign"),
"sign_idx": ad.get("sign_idx"),
"lord": ad.get("lord"),
"start_age": ad.get("start_age"),
"end_age": ad.get("end_age"),
"duration_years": ad.get("years"),
"duration_days": round(float(ad.get("years", 0)) * 365.25, 2),
"direction": ad.get("direction") or md.get("sequence_profile"),
"seed_sign": ad.get("start_sign_source") or md.get("sign"),
"profile": {
"duration_profile": md.get("duration_profile"),
"sequence_profile": md.get("sequence_profile"),
"seed_profile": md.get("seed_profile"),
"dual_lord_profile": dual_lord_profile,
"antardasha_profile": antardasha_profile,
},
"source_formula_status": ad.get("evidence_status", ad_status),
"parent_md": md.get("sign"),
"parent_period_key": md.get("period_key"),
"sequence_index": ad_index - 1,
})
return rows
def narayana_dasha_full_report(
lagna_sign_idx: int,
planet_lons: Dict[str, float],
current_age: float = 0,
birth_year: int = 0,
duration_profile: str = "legacy_forward_v0",
sequence_profile: str = "legacy_zodiacal_v0",
seed_profile: str = "legacy_lagna_v0",
dual_lord_profile: str = "legacy_primary_lord_v0",
antardasha_profile: str = "legacy_weighted_v0",
cycle_count: int = 1,
) -> Dict:
"""
Narayana Dasha 完整报告。
参数:
lagna_sign_idx: Lagna 星座索引 (0-11)
planet_lons: 行星经度字典
current_age: 当前年龄
birth_year: 出生年份
返回:
dict with mahadasha_sequence, current_dasha, total_cycle_years
"""
result = {}
# 1. 完整大运序列
mahadasha = calc_narayana_mahadasha(lagna_sign_idx, planet_lons)
total_cycle = sum(p['years'] for p in mahadasha)
result['mahadasha_sequence'] = mahadasha
result['total_cycle_years'] = total_cycle
result['lagna_sign'] = SIGNS[lagna_sign_idx]
result['lagna_sign_idx'] = lagna_sign_idx
result['duration_profile'] = duration_profile
result['sequence_profile'] = sequence_profile
result['seed_profile'] = seed_profile
result['dual_lord_profile'] = dual_lord_profile
result['antardasha_profile'] = antardasha_profile
result['cycle_count'] = cycle_count
result['seed_selection'] = (
select_narayana_seed_sign(lagna_sign_idx, planet_lons)
if seed_profile == "jaimini_lagna_seventh_strength_v1"
else {
"status": "selected",
"profile": "legacy_lagna_v0",
"selected_sign_idx": lagna_sign_idx,
"selected_sign": SIGNS[lagna_sign_idx],
}
)
result['dense_boundaries'] = build_narayana_dense_boundaries(
mahadasha,
planet_lons=planet_lons,
antardasha_profile=antardasha_profile,
dual_lord_profile=dual_lord_profile,
)
result['dense_boundary_schema'] = "narayana_dense_boundaries.v1"
# 2. 当前大运
if current_age > 0:
curr = get_current_narayana_dasha(mahadasha, current_age)
result['current_dasha'] = curr
# 当前日期(如果有出生年份)
if birth_year > 0:
curr_year = birth_year + int(current_age)
result['current_year'] = curr_year
result['current_age'] = current_age
# 3. 简要解读
result['interpretation'] = _interpret_narayana(result, current_age)
return result
def _interpret_narayana(result: Dict, current_age: float) -> List[str]:
"""Narayana Dasha 简要解读"""
lines = []
md_seq = result.get('mahadasha_sequence', [])
if md_seq:
total = sum(p['years'] for p in md_seq)
lines.append(f"Narayana Dasha 完整周期: {total} 年(12 星座 × 可变年数)")
lines.append(f"起运星座: {result.get('lagna_sign', '?')}Lagna")
curr = result.get('current_dasha', {})
md = curr.get('md')
if md:
lord_cond = _check_lord_condition(md['sign_idx'], result.get('lagna_sign_idx', 0))
lines.append(f"当前 Narayana Mahadasha: {md['sign']}(守护星 {md['lord']}{md['years']}年)")
lines.append(f" 剩余: {curr.get('remaining_years', 0):.1f}")
lines.append(f" {lord_cond}")
ad = curr.get('ad')
if ad:
lines.append(f"当前 Antardasha: {ad['sign']}(守护星 {ad['lord']}{ad['years']}年)")
# 与 Vimshottari 互补提示
lines.append("")
lines.append("【与 Vimshottari 互补解读提示】")
lines.append("Narayana Dasha 的星座主题与 Vimshottari 的行星主题形成互补。")
lines.append("两者一致 → 事件确定性高;两者矛盾 → 混合影响,需看具体宫位。")
return lines
def _check_lord_condition(sign_idx: int, lagna_idx: int) -> str:
"""检查当前星座守护星与 Lagna 的关系"""
house_from_lagna = (sign_idx - lagna_idx + 12) % 12 + 1
house_labels = {
1: 'Lagna(自我/身体)', 2: '2宫(财富/家庭)', 3: '3宫(兄弟/努力)',
4: '4宫(家庭/房产)', 5: '5宫(子女/创意)', 6: '6宫(健康/竞争)',
7: '7宫(婚姻/合作)', 8: '8宫(转型/遗产)', 9: '9宫(信仰/长途)',
10: '10宫(事业/地位)', 11: '11宫(收益/社交)', 12: '12宫(支出/灵性)',
}
return f" 从 Lagna 算 = {house_from_lagna}{house_labels.get(house_from_lagna, '')}"
# =========================================================================
# CLI 测试入口
# =========================================================================
if __name__ == '__main__':
print("Narayana DashaRishi Dasha)模块 v6.0.20")
print()
# 测试数据:用户星盘 Le Asc, Saturn MD
test_lagna = 4 # Leo
test_planets = {
'Sun': 19.07, 'Moon': 325.64, 'Mars': 98.12, 'Mercury': 32.56,
'Jupiter': 97.08, 'Venus': 55.01, 'Saturn': 306.96,
'Rahu': 342.27, 'Ketu': 162.27,
}
test_age = 33.13 # 2026-06-04 年龄
print(f"测试: Leo Asc, age={test_age}")
print(f"行星经度: { {k: f'{v:.1f}' for k,v in test_planets.items()} }")
print()
result = narayana_dasha_full_report(test_lagna, test_planets, test_age, 1990)
print("=== 大运序列 ===")
for p in result['mahadasha_sequence']:
print(f" {p['sign']:>12s} ({p['lord']:>7s} in {p['lord_in_sign']:>12s}): "
f"{p['years']:2d}年 [{p['start_age']:5.1f}{p['end_age']:5.1f}] "
f"{p['count_from_to']}")
print(f"\n总周期: {result['total_cycle_years']}")
print("\n=== 当前 Dasha ===")
curr = result.get('current_dasha', {})
md = curr.get('md')
if md:
print(f" MD: {md['sign']} ({md['lord']}) {md['years']}")
print(f" 剩余: {curr['remaining_years']}")
ad = curr.get('ad')
if ad:
print(f" AD: {ad['sign']} ({ad['lord']}) {ad['years']}")
print("\n=== 解读 ===")
for line in result.get('interpretation', []):
print(line)