#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Narayana Dasha（Rishi 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 Dasha（Rishi 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)