fix(yoga): Pancha Mahapurusha has one main detector; combust/afflicted only reduced (T1, BUG-1207)

Astrologer ruling 甲5. scripts/pancha_mahapurusha.py is the only detector: kendras from
the Lagna; retrograde no longer cancels; combustion (shared orb table), malefic
conjunction and malefic graha drishti (Sun 7, Mars 4/7/8, Saturn 3/7/10) mark the
yoga reduced with reasons; D9 debilitation still cancels when navamsa_sign is given
(current behaviour, ruling silent, asked next round). _detect_yogas now passes the
Sun's longitude (combustion was never checked). mahapurusha_*, *_own, moolatrikona_*,
hamsa_trikona (Jupiter 1/5/9, not classical) and bvr_maalavya_yoga are reference_only;
yogas_doshas.calc_pancha_mahapurusha_yoga is kept as reference_yogas only. The rule
engine appends main-detector rows; cmd_yoga rows now carry lon + retrograde so the
report and card get the same combustion verdict. Card label lists 部分减弱 reasons.

Changed assertions (three columns in PROGRESS): self-conjunction tests (malefic in
same sign: invalid -> reduced), run_all t31/t32 (combust/retro: invalid -> reduced /
full), run_all t54 (evaluated rules >= 400 -> evaluated + reference_only >= 400).

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N4f2nya58RoRu4yEmJgRGE

scripts/jyotish_api_server.py

@@ -7118,11 +7118,13 @@ class JyotishAPIHandler(BaseHTTPRequestHandler, VedastroEvidenceMixin, SynastryM
         yogas = []
         KENDRA = {1,4,7,10}
         try:
-            from pancha_mahapurusha import detect_pancha_mahapurusha
-            pmc = detect_pancha_mahapurusha(planets)
-            for y in pmc:
-                if y['is_valid']:
-                    yogas.append({'name': y['name'], 'planets': [y['planet']], 'category': 'PMC'})
+            # BUG-1207: the one Pancha Mahapurusha detector, with the Sun's
+            # longitude for combustion; reduced yogas stay formed.
+            from main_yoga_detectors import pancha_mahapurusha_rows
+            sun_lon = (planets.get('Sun') or {}).get('lon')
+            for y in pancha_mahapurusha_rows(planets, sun_degree=sun_lon):
+                yogas.append({'name': y['name'], 'planets': y['planets'], 'category': 'PMC',
+                              'reductions': y['reductions']})
         except Exception as e:
             import logging
             logging.warning(f"[api_server] pancha_mahapurusha detection failed: {e}")

scripts/jyotish_engine.py

@@ -12536,7 +12536,12 @@ def cmd_yoga(args):
                         "house": pd["house"],
                         "degree": pd.get("degree_in_sign", pd.get("degree")),
                     }
+                    # Main yoga detectors (BUG-1207) read the longitude and the
+                    # retrograde flag (combustion orb), as the consultation path does.
+                    if pd.get("retrograde") is not None:
+                        planets[pname]["retrograde"] = bool(pd.get("retrograde"))
                     if pd.get("degree_raw") is not None:
+                        planets[pname]["lon"] = pd.get("degree_raw")
                         planet_lons_for_varga[pname] = pd.get("degree_raw")
             try:
                 from varga import calc_all_vargas

scripts/main_yoga_detectors.py

@@ -13,6 +13,10 @@ English reader edition).
 
 - Neecha Bhanga (BUG-1205, 乙14): `yogas_doshas.calc_nicha_bhanga_raj_yoga`;
   only the cancellation conditions that hold are listed, never called Raja Yoga.
+- Pancha Mahapurusha (BUG-1207, 甲5): `pancha_mahapurusha.detect_pancha_mahapurusha`;
+  kendras from the Lagna only, retrograde does not cancel, combustion / malefic
+  conjunction or aspect only mark the yoga `reduced` (部分减弱), D9 debilitation
+  still cancels it (current behaviour; not in the ruling, asked next round).
 - Kemadruma (BUG-1206, 甲7): `yoga_expansion.detect_kemadruma`; the 2nd / 12th
   from the Moon count only Mars, Mercury, Jupiter, Venus, Saturn, and there is no
   kendra cancellation (planets in kendras, including only Rahu / Ketu there, do
@@ -29,9 +33,11 @@ from __future__ import annotations
 from typing import Any, Dict, List, Optional
 
 try:  # scripts/ on sys.path
+    from pancha_mahapurusha import _longitude, detect_pancha_mahapurusha
     from yoga_expansion import detect_kemadruma
     from yogas_doshas import calc_nicha_bhanga_raj_yoga
 except ImportError:  # package import (scripts.main_yoga_detectors)
+    from scripts.pancha_mahapurusha import _longitude, detect_pancha_mahapurusha
     from scripts.yoga_expansion import detect_kemadruma
     from scripts.yogas_doshas import calc_nicha_bhanga_raj_yoga
 
@@ -55,13 +61,11 @@ NEECHA_BHANGA_NOTE_EN = 'Conditions only; no stand-alone conclusion'
 
 
 def _lon(row: Any) -> Optional[float]:
-    if not isinstance(row, dict):
-        return None
-    for key in ('lon', 'degree_raw', 'longitude'):
-        value = row.get(key)
-        if isinstance(value, (int, float)):
-            return float(value) % 360.0
-    return None
+    """Sidereal longitude; same reading as the Pancha Mahapurusha detector
+    (`lon` / `degree_raw` / `longitude`, else `degree` as a longitude when
+    >= 30 or as a degree in `sign`), so the report rows (degree in sign) and
+    the API rows (longitudes) give the same D9 and the same verdict."""
+    return _longitude(row) if isinstance(row, dict) else None
 
 
 def navamsa_sign(lon: float) -> str:
@@ -144,6 +148,54 @@ def neecha_bhanga_rows(planets: Dict[str, Any]) -> List[Dict[str, Any]]:
     return rows
 
 
+# Names as the rule registry and the card closed list spell them.
+MAHAPURUSHA_ROWS = {
+    'Mars': ('Ruchaka Yoga', '鲁查卡格局'),
+    'Mercury': ('Bhadra Yoga', '巴德拉格局'),
+    'Jupiter': ('Hamsa Yoga', '汉萨格局'),
+    'Venus': ('Malavya Yoga', '马拉维亚格局'),
+    'Saturn': ('Sasa Yoga', '萨萨格局'),
+}
+EXALTATION_SIGN = {'Mars': 'Capricorn', 'Mercury': 'Virgo', 'Jupiter': 'Cancer', 'Venus': 'Pisces', 'Saturn': 'Libra'}
+
+
+def pancha_mahapurusha_rows(planets: Dict[str, Any], sun_degree: Optional[float] = None) -> List[Dict[str, Any]]:
+    """Formed Pancha Mahapurusha yogas (full or reduced) from the main detector."""
+    rows: List[Dict[str, Any]] = []
+    for item in detect_pancha_mahapurusha(planets or {}, sun_degree):
+        if not item.get('is_valid'):
+            continue
+        planet = item['planet']
+        name, name_cn = MAHAPURUSHA_ROWS[planet]
+        sign = item.get('sign')
+        exalted = EXALTATION_SIGN.get(planet) == sign
+        reductions = list(item.get('reductions') or [])
+        reductions_en = list(item.get('reductions_en') or [])
+        combination = f"{_planet_zh(planet)}在第{item.get('house')}宫{_sign_zh(sign)}（{'入旺' if exalted else '入庙'}）"
+        combination_en = f"{planet} in house {item.get('house')}, {sign} ({'exalted' if exalted else 'own sign'})"
+        if reductions:
+            combination += '；部分减弱：' + '；'.join(reductions)
+            combination_en += '; partly weakened: ' + '; '.join(reductions_en)
+        rows.append({
+            'name': name,
+            'name_cn': name_cn,
+            'category': 'mahapurusha',
+            'planets': [planet],
+            'combination': combination,
+            'combination_en': combination_en,
+            'reductions': reductions,
+            'reductions_en': reductions_en,
+            'effects': ['卓越才能', '领域领军', '人格魅力'],
+            'effects_en': ['Outstanding abilities', "Leader in one's field", 'Personal charisma'],
+            'strength': '部分减弱' if reductions else '强',
+            'strength_en': 'partly weakened' if reductions else 'strong',
+            'source': 'BPHS',
+            'rule_id': f'main_detector:pancha_mahapurusha:{planet.lower()}',
+            'detector': 'pancha_mahapurusha.detect_pancha_mahapurusha',
+        })
+    return rows
+
+
 def kemadruma_rows(planets: Dict[str, Any]) -> List[Dict[str, Any]]:
     """Kemadruma from `yoga_expansion.detect_kemadruma` (no kendra cancellation)."""
     result = detect_kemadruma(planets or {})
@@ -170,4 +222,4 @@ def kemadruma_rows(planets: Dict[str, Any]) -> List[Dict[str, Any]]:
 
 def main_detector_rows(planets: Dict[str, Any], ascendant: Optional[str] = None) -> List[Dict[str, Any]]:
     """Every main-detector row for one D1 chart, in a fixed order."""
-    return kemadruma_rows(planets) + neecha_bhanga_rows(planets)
+    return pancha_mahapurusha_rows(planets) + kemadruma_rows(planets) + neecha_bhanga_rows(planets)

scripts/pancha_mahapurusha.py

@@ -11,7 +11,10 @@ Pancha Mahapurusha Yoga（五王瑜伽）完整检测模块 v1.0
 - Malavya Yoga (Venus)
 - Shasha Yoga (Saturn)
 
-含完整的燃烧(Combustion)/逆行(Retrograde)/受克(Affliction)失效条件检测
+破格口径按占星师定稿 2026-10-03 甲5（BUG-1207）：逆行不破格；燃烧、与凶星
+同宫或被照只记「部分减弱」；D9 落陷判不成立（现状，待问）。本模块是五大人格的
+唯一主检测器，`yoga_rules.json` 同名规则与 `yogas_doshas.calc_pancha_mahapurusha_yoga`
+为 reference_only。
 """
 
 from typing import Dict, List, Optional
@@ -73,7 +76,7 @@ def _is_combust(planet: str, planet_degree: float, sun_degree: float, retrograde
 
 
 def _has_malefic_aspect(planet_sign: str, all_planets: Dict, planet: str = None) -> List[str]:
-    """检查是否有凶星合相/相位该行星"""
+    """Natural malefics in the same sign as the yoga planet (the planet itself excluded, BUG-1180)."""
     afflicting = []
     for pn, pd in all_planets.items():
         if pn == planet or pn not in NATURAL_MALEFICS:
@@ -84,21 +87,73 @@ def _has_malefic_aspect(planet_sign: str, all_planets: Dict, planet: str = None)
     return afflicting
 
 
+# Parashari whole-sign graha drishti, as aspects.calc_house_aspects: every graha
+# the 7th; Mars also 4 / 8, Saturn 3 / 10; Rahu / Ketu cast none.
+MALEFIC_DRISHTI = {'Sun': (7,), 'Mars': (4, 7, 8), 'Saturn': (3, 7, 10)}
+
+
+def _malefic_aspectors(planet_sign: str, all_planets: Dict, planet: str = None) -> List[str]:
+    """Natural malefics whose graha drishti falls on the yoga planet's sign."""
+    if planet_sign not in SIGNS:
+        return []
+    target = SIGNS.index(planet_sign)
+    casters = []
+    for pn, offsets in MALEFIC_DRISHTI.items():
+        pd = all_planets.get(pn)
+        if pn == planet or not isinstance(pd, dict) or pd.get('sign') not in SIGNS:
+            continue
+        origin = SIGNS.index(pd['sign'])
+        if any((origin + offset - 1) % 12 == target for offset in offsets):
+            casters.append(pn)
+    return casters
+
+
+def _longitude(pd: Dict) -> Optional[float]:
+    """Sidereal longitude of a row; `degree` may be a longitude or a degree in sign."""
+    for key in ('lon', 'degree_raw', 'longitude'):
+        value = pd.get(key)
+        if isinstance(value, (int, float)):
+            return float(value) % 360.0
+    degree = pd.get('degree')
+    if not isinstance(degree, (int, float)):
+        return None
+    if degree >= 30 or pd.get('sign') not in SIGNS:
+        return float(degree) % 360.0
+    return SIGNS.index(pd['sign']) * 30.0 + float(degree)
+
+
+MAHAPURUSHA_ORDER = ('Mars', 'Mercury', 'Jupiter', 'Venus', 'Saturn')
+PLANET_ZH = {'Sun': '太阳', 'Mars': '火星', 'Mercury': '水星', 'Jupiter': '木星', 'Venus': '金星',
+             'Saturn': '土星', 'Rahu': '罗睺', 'Ketu': '计都'}
+
+
 def detect_pancha_mahapurusha(planets: Dict, sun_degree: float = None) -> List[Dict]:
     """
-    检测所有Pancha Mahapurusha Yoga，含失效条件。
+    检测所有Pancha Mahapurusha Yoga（主检测器）。
+
+    占星师定稿 2026-10-03 甲5（BUG-1207）：
+    - 只从命宫（Lagna）角宫计算：行星在上升起第 1/4/7/10 宫且入庙或入旺；
+    - 逆行不破格；
+    - 燃烧不破格，只记「部分减弱」（`strength: reduced`）；
+    - 与凶星同宫或被凶星照只降低强度，不消灭格局；
+    - D9 落陷判不成立（`is_valid: False`）保留现状：只在传入 `navamsa_sign` 时检查，
+      产品路径不传；定稿未提及，待问占星师。
 
     Args:
-        planets: {planet: {'sign': str, 'house': int, 'degree': float, ...}}
-        sun_degree: 太阳黄道经度(用于燃烧检测)
+        planets: {planet: {'sign': str, 'house': int, 'lon'|'degree': float, 'navamsa_sign'?: str}}
+            `house` 从上升数；经度取 `lon` / `degree_raw` / `longitude`，否则 `degree`
+            （≥30 视为经度，否则按星座内度数）。
+        sun_degree: 太阳黄道经度（燃烧检测）；缺省时取 planets['Sun'] 的经度。
 
     Returns:
-        检测到的PMC Yoga列表
+        检测到的PMC Yoga列表（含 strength: full / reduced / cancelled）
     """
     yogas = []
     sun_deg = sun_degree
+    if sun_deg is None and isinstance(planets.get('Sun'), dict):
+        sun_deg = _longitude(planets['Sun'])
 
-    for planet in MAHAPURUSHA_PLANETS:
+    for planet in MAHAPURUSHA_ORDER:
         pd = planets.get(planet)
         if not pd:
             continue
@@ -115,49 +170,57 @@ def detect_pancha_mahapurusha(planets: Dict, sun_degree: float = None) -> List[D
         yoga_name = MAHAPURUSHA_NAMES[planet]
         effect = MAHAPURUSHA_EFFECTS.get(yoga_name.replace(" Yoga", ""), "")
 
-        # 检查失效条件
         cancellations = []
+        reductions = []
+        reductions_en = []
         is_valid = True
 
-        # 1. 燃烧检测
-        if sun_deg is not None:
-            planet_deg = pd.get('degree', 0)
-            if _is_combust(planet, planet_deg, sun_deg, bool(pd.get('retrograde'))):
-                profile = COMBUSTION_ORB.get(planet, {'direct': 10})
-                threshold = profile.get('retrograde' if pd.get('retrograde') else 'direct', profile.get('direct', 10))
-                cancellations.append(f"{planet}被太阳燃烧(orb < {threshold}°) — Yoga效力大减")
-                is_valid = False
+        # 1. 燃烧：部分减弱（容许度表与 jyotish_engine.COMBUSTION_ORBS 一致，
+        #    tests/test_combustion_orb_tables_agree.py）
+        planet_deg = _longitude(pd)
+        retro = bool(pd.get('retrograde'))
+        if sun_deg is not None and planet_deg is not None and _is_combust(planet, planet_deg, sun_deg, retro):
+            profile = COMBUSTION_ORB.get(planet, {'direct': 10})
+            threshold = profile.get('retrograde' if retro else 'direct', profile.get('direct', 10))
+            reductions.append(f"{PLANET_ZH[planet]}燃烧（距太阳 {threshold}° 以内）")
+            reductions_en.append(f"{planet} combust (within {threshold}° of the Sun)")
 
-        # 2. 逆行检测
-        if pd.get('retrograde', False):
-            cancellations.append(f"{planet}逆行 — Yoga效力减弱")
-            is_valid = False
+        # 2. 逆行：不破格，不减弱（定稿甲5）
 
-        # 3. 凶星受克
+        # 3. 凶星同宫 / 凶星照：只降低强度
         malefics = _has_malefic_aspect(sign, planets, planet)
         if malefics:
-            cancellations.append(f"{planet}受凶星合相: {', '.join(malefics)} — Yoga部分失效")
-            is_valid = False
+            reductions.append(f"与凶星同宫：{'、'.join(PLANET_ZH.get(m, m) for m in malefics)}")
+            reductions_en.append(f"conjunct malefic(s): {', '.join(malefics)}")
+        aspectors = _malefic_aspectors(sign, planets, planet)
+        if aspectors:
+            reductions.append(f"受凶星照：{'、'.join(PLANET_ZH.get(m, m) for m in aspectors)}")
+            reductions_en.append(f"aspected by malefic(s): {', '.join(aspectors)}")
 
-        # 4. Navamsa弱势（D9中落陷）
+        # 4. Navamsa弱势（D9中落陷）：不成立。现状保留——只在调用方给出
+        #    `navamsa_sign` 时检查；产品路径（chart.yogas、规则引擎、报告）都不传，
+        #    所以这一条目前不生效。定稿未提及，记为下轮问题，不在本单启用。
         navamsa_sign = pd.get('navamsa_sign', '')
         if navamsa_sign in OWN_SIGNS.get(planet, set()) or (planet in EXALTATION and EXALTATION[planet][0] == navamsa_sign):
             pass  # D9 also strong → yoga reinforced
         elif navamsa_sign:
-            # Check if debilitated in D9
-            deb_sign_idx = (SIGNS.index(EXALTATION[planet][0]) + 6) % 12 if planet in EXALTATION and planet != 'Rahu' and planet != 'Ketu' else -1
+            deb_sign_idx = (SIGNS.index(EXALTATION[planet][0]) + 6) % 12 if planet in EXALTATION else -1
             if deb_sign_idx >= 0 and navamsa_sign == SIGNS[deb_sign_idx]:
                 cancellations.append(f"{planet}在Navamsa落陷 — Yoga根基不稳")
                 is_valid = False
 
+        strength = 'cancelled' if not is_valid else ('reduced' if reductions else 'full')
         yogas.append({
             'name': yoga_name,
             'planet': planet,
             'sign': sign,
             'house': house,
             'is_valid': is_valid,
-            'strength': '完整' if is_valid else '削弱',
+            'strength': strength,
+            'strength_cn': {'full': '完整', 'reduced': '部分减弱', 'cancelled': '不成立'}[strength],
             'effect': effect,
+            'reductions': reductions,
+            'reductions_en': reductions_en,
             'cancellations': cancellations,
         })
 
@@ -175,13 +238,14 @@ def assess_pmc_strength(planets: Dict, sun_degree: float = None) -> Dict:
     valid = [y for y in yogas if y['is_valid']]
     invalid = [y for y in yogas if not y['is_valid']]
 
+    reduced = [y['name'] for y in valid if y['strength'] == 'reduced']
     if len(valid) >= 2:
         assessment = "多重大王瑜伽 — 命主在不同领域均有杰出天赋"
     elif len(valid) == 1:
         assessment = f"{valid[0]['name']} — 命主在特定领域有卓越潜力"
     elif len(invalid) > 0:
         st = [y['planet'] for y in invalid]
-        assessment = f"瑜伽潜质存在但因{', '.join(st)}受克而削弱"
+        assessment = f"瑜伽潜质存在但因{', '.join(st)}在 D9 落陷而不成立"
     else:
         assessment = "无Pancha Mahapurusha Yoga"
 
@@ -189,6 +253,7 @@ def assess_pmc_strength(planets: Dict, sun_degree: float = None) -> Dict:
         'method': 'Pancha Mahapurusha Yoga 完整检测 v1.0',
         'yogas': yogas,
         'valid_count': len(valid),
+        'reduced': reduced,
         'invalid_count': len(invalid),
         'assessment': assessment,
     }

scripts/yogas_doshas.py

@@ -827,7 +827,18 @@ def calc_all_yogas_doshas(planets_data: Dict, houses: Dict,
     # Yogas
     results['raj_yogas'] = calc_raj_yogas(planets_data, houses)
     results['dhana_yogas'] = calc_dhana_yogas(planets_data, houses)
-    results['pancha_mahapurusha'] = calc_pancha_mahapurusha_yoga(planets_data)
+    # BUG-1207 (甲5): Pancha Mahapurusha has one main detector
+    # (scripts/pancha_mahapurusha.py, via main_yoga_detectors). This older
+    # aggregate checks no affliction at all; its list is kept for reference only
+    # and is not counted as a formed yoga.
+    reference = calc_pancha_mahapurusha_yoga(planets_data)
+    results['pancha_mahapurusha'] = {
+        'status': 'reference_only',
+        'main_detector': 'scripts/pancha_mahapurusha.py',
+        'yogas': [],
+        'reference_yogas': reference['yogas'],
+        'summary': 'reference_only：五大人格以 scripts/pancha_mahapurusha.py 主检测器为准',
+    }
     results['nicha_bhanga_raj'] = calc_nicha_bhanga_raj_yoga(planets_data, houses)
 
     # Doshas