fix(ashtottari): source arcs and a shifted sub-period clock (BUG-1475, upstream 0712f98f)

Rahu uses the circular unique start. Output keys are unchanged. Upstream arc and clock tests passed.

Co-Authored-By: Grok (agent loop)

scripts/ashtottari_dasha.py

@@ -51,9 +51,9 @@ NAKSHATRAS = [
     "Uttara Bhadrapada", "Revati"
 ]
 
-# Candidate 28-nakshatra cycle used by the PL9 sample-admitted path.
-# This preserves the sample's Mula -> Mercury balance note while leaving the
-# legacy Moon-whitelist rule available for research comparison.
+# PVR Integrated Approach Table 39 angular arcs over the 27 named stars.
+# This profile includes the reduced Saturn arc and circular Rahu arc.
+# Applicability and child-period profiles remain separate choices.
 _CANDIDATE_NAKSHATRA_STARTING_LORDS = {
     "Ardra": "Sun",
     "Punarvasu": "Sun",
@@ -171,7 +171,12 @@ def _candidate_lord_nakshatra_span(starting_planet: str) -> tuple[int, int] | tu
     ]
     if not matching:
         return None, None
-    return matching[0], matching[-1]
+    # Table 39 has a circular Rahu arc: Uttara Bhadrapada through Bharani.
+    # min/max would incorrectly turn that four-star arc into the full zodiac.
+    starts = [idx for idx in matching if (idx - 1) % 27 not in matching]
+    if len(starts) != 1:
+        return None, None
+    return starts[0], starts[0] + len(matching) - 1
 
 
 def _years_to_balance_parts(years: float) -> dict:
@@ -225,11 +230,12 @@ def _calc_candidate_balance_and_birth_chain(moon_longitude: float, starting_plan
     if start_nak_idx is None or end_nak_idx is None:
         return None, None
 
-    one_star = 360.0 / 27.0
-    segment_start = start_nak_idx * one_star
-    segment_end = (end_nak_idx + 1) * one_star
+    segment_start = start_nak_idx * 40.0 / 3.0
+    segment_end = (end_nak_idx + 1) * 40.0 / 3.0
     longitude = float(moon_longitude) % 360.0
-    if not (segment_start <= longitude <= segment_end):
+    if segment_end > 360 and longitude < segment_start:
+        longitude += 360
+    if not (segment_start <= longitude < segment_end):
         return None, None
 
     span = segment_end - segment_start
@@ -276,7 +282,7 @@ def _calc_candidate_balance_and_birth_chain(moon_longitude: float, starting_plan
     balance = {
         "planet": starting_planet,
         "segment_start_nakshatra": NAKSHATRAS[start_nak_idx],
-        "segment_end_nakshatra": NAKSHATRAS[end_nak_idx],
+        "segment_end_nakshatra": NAKSHATRAS[end_nak_idx % 27],
         "segment_span_nakshatras": end_nak_idx - start_nak_idx + 1,
         "display_calendar": "360_day_traditional",
         **_years_to_balance_parts_360(remaining_years),
@@ -314,25 +320,36 @@ def _timeline_start_with_birth_balance(
     return birth_dt - timedelta(days=elapsed_years * 365.25), True
 
 
-def _current_period_from_shifted_majors(major_periods: list, now: datetime) -> dict | None:
-    if not major_periods:
+def _shift_period_tree(value, shift: timedelta):
+    """Copy a period tree and move every stored instant by the same cycle shift."""
+    if isinstance(value, list):
+        return [_shift_period_tree(item, shift) for item in value]
+    if isinstance(value, dict):
+        shifted = {}
+        for key, item in value.items():
+            if key in ("start_date", "end_date") and isinstance(item, str):
+                shifted[key] = (datetime.fromisoformat(item) + shift).isoformat()
+            else:
+                shifted[key] = _shift_period_tree(item, shift)
+        return shifted
+    return value
+
+
+def _current_period_on_cycle(major_periods: list, birth_dt: datetime, now: datetime) -> dict | None:
+    """Second and later cycles move child dates with the parent. Birth row stays put."""
+    if not major_periods or now < birth_dt:
         return None
-    first = datetime.fromisoformat(major_periods[0]["start_date"])
-    last = datetime.fromisoformat(major_periods[-1]["end_date"])
-    span = last - first
-    if span.total_seconds() <= 0 or now < first:
-        return None
-    cycles = int((now - first).total_seconds() // span.total_seconds())
+    cycle_duration = timedelta(days=TOTAL_CYCLE * 365.25)
+    cycle_index = (now - birth_dt) // cycle_duration
+    shift = cycle_index * cycle_duration
     for period in major_periods:
-        start = datetime.fromisoformat(period["start_date"]) + cycles * span
-        end = datetime.fromisoformat(period["end_date"]) + cycles * span
+        start = datetime.fromisoformat(period["start_date"]) + shift
+        end = datetime.fromisoformat(period["end_date"]) + shift
         if start <= now < end:
-            current = period.copy()
-            current["start_date"] = start.isoformat()
-            current["end_date"] = end.isoformat()
-            current["elapsed_years"] = (now - start).days / 365.25
-            current["remaining_years"] = (end - now).days / 365.25
-            current["cycle_number"] = cycles + 1
+            current = _shift_period_tree(period, shift)
+            current["elapsed_years"] = (now - start).total_seconds() / (86400 * 365.25)
+            current["remaining_years"] = (end - now).total_seconds() / (86400 * 365.25)
+            current["cycle_number"] = cycle_index + 1
             return current
     return None
 
@@ -467,40 +484,22 @@ def calculate_ashtottari_dasha(birth_info: dict) -> dict:
             idx for idx, lord in enumerate(DASHA_SEQUENCE) if lord["planet"] == starting_planet
         )
 
-    timeline_start, balance_shifted = _timeline_start_with_birth_balance(
+    timeline_start, _balance_shifted = _timeline_start_with_birth_balance(
         birth_dt, moon_longitude, starting_planet,
     )
     major_periods = _build_major_periods(start_lord_idx, timeline_start)
     antardasha_periods, pratyantardasha_periods = _attach_recursive_subperiods(major_periods)
 
-    # Determine current period. Major periods repeat every 108 years; older natives
-    # should still return a current period instead of None after the first cycle.
-    # A shifted timeline is read from the period dates. An unshifted timeline keeps
-    # the previous birth-anchored walk so charts without a Moon longitude stay put.
-    now = datetime.now()
-    current_period = None
-    if balance_shifted:
-        current_period = _current_period_from_shifted_majors(major_periods, now)
-    else:
-        age_years = max((now - birth_dt).days / 365.25, 0)
-        current_in_cycle = age_years % TOTAL_CYCLE
-        cumulative = 0.0
-        for p in major_periods:
-            years = p["years"]
-            if cumulative <= current_in_cycle < cumulative + years:
-                cycle_start = birth_dt + timedelta(days=(age_years - current_in_cycle + cumulative) * 365.25)
-                cycle_end = cycle_start + timedelta(days=years * 365.25)
-                current_period = p.copy()
-                current_period["start_date"] = cycle_start.isoformat()
-                current_period["end_date"] = cycle_end.isoformat()
-                current_period["elapsed_years"] = (now - cycle_start).days / 365.25
-                current_period["remaining_years"] = (cycle_end - now).days / 365.25
-                current_period["cycle_number"] = int(age_years // TOTAL_CYCLE) + 1
-                break
-            cumulative += years
+    # Later cycles repeat every 108 years. Child dates move with the parent.
+    # The birth row in major stays on the original timeline.
+    now = birth_info.get("as_of") or birth_info.get("today") or datetime.now()
+    if isinstance(now, str):
+        now = datetime.fromisoformat(now)
+    current_period = _current_period_on_cycle(major_periods, birth_dt, now)
 
     result = {
         "applicable": True,
+        "reference_date": now.isoformat(),
         "major": major_periods,
         "antardasha": antardasha_periods,
         "pratyantardasha": pratyantardasha_periods,

scripts/cmd_nakshatra_adv.py

@@ -39,14 +39,15 @@ def _compute_chart_from_args(args: Any) -> tuple:
     planets = chart.get('planets', {})
     planet_lons = {}
     for pn, pd in planets.items():
-        if isinstance(pd, dict) and 'degree' in pd:
-            planet_lons[pn] = pd['degree']
+        if isinstance(pd, dict) and ('degree_raw' in pd or 'degree' in pd):
+            planet_lons[pn] = pd.get('degree_raw', pd.get('degree'))
 
     if not planet_lons:
         return chart, asc_idx, jd, ayanamsa, {}, 0, 0, 0
 
     moon_lon = planet_lons.get('Moon', 0)
-    moon_nak_idx = int(moon_lon / (360.0 / 27)) % 27
+    from nakshatra_advanced import find_nakshatra
+    moon_nak_idx = find_nakshatra(moon_lon)['nakshatra_idx']
     asc_sign_idx = asc_idx if asc_idx is not None else 0
 
     return chart, asc_idx, jd, ayanamsa, planet_lons, moon_lon, moon_nak_idx, asc_sign_idx
@@ -174,15 +175,15 @@ def cmd_nakshatra_dasha(args: Any) -> Dict[str, Any]:
 
     # 构建 birth_date_str
     birth_date_str = f"{args.year}-{args.month:02d}-{args.day:02d}"
+    birth_moment = datetime(args.year, args.month, args.day, args.hour, args.minute,
+                            int(getattr(args, 'second', 0) or 0))
 
     # 年龄
     age = getattr(args, 'age', None)
     if age is None:
-        try:
-            birth_date = datetime(args.year, args.month, args.day)
-            age = (datetime.now() - birth_date).days / 365.25
-        except:
-            age = 30.0
+        requested_date = getattr(args, 'transit_date', None)
+        observation = datetime.fromisoformat(requested_date) if requested_date else datetime.now()
+        age = (observation - birth_moment).total_seconds() / (86400 * 365.25)
 
     mode = getattr(args, 'mode', 'all') or 'all'
 
@@ -198,7 +199,7 @@ def cmd_nakshatra_dasha(args: Any) -> Dict[str, Any]:
         return asht
 
     if mode == 'vimshottari':
-        return calc_nakshatra_dasha_breakdown(moon_lon, planet_lons, birth_date_str, age)
+        return calc_nakshatra_dasha_breakdown(moon_lon, planet_lons, birth_moment.isoformat(), age)
 
     if mode == 'overlay':
         # 需要过境数据：获取当前日期或 --transit-date

scripts/nakshatra_dasha.py

@@ -25,6 +25,11 @@ try:
         NAK_LIST, NAK_NAMES, NAK_LORDS, NAK_YEARS,
         find_nakshatra, calc_tara_bala,
     )
+    from ashtottari_dasha import (
+        _candidate_nakshatra_starting_lord,
+        _calc_candidate_balance_and_birth_chain,
+    )
+    from ashtottari_rule_profiles import RULE_FAMILY_RAHU_NOT_IN_KENDRA
 except ImportError:
     import sys, os
     sys.path.insert(0, os.path.dirname(__file__))
@@ -32,6 +37,11 @@ except ImportError:
         NAK_LIST, NAK_NAMES, NAK_LORDS, NAK_YEARS,
         find_nakshatra, calc_tara_bala,
     )
+    from ashtottari_dasha import (
+        _candidate_nakshatra_starting_lord,
+        _calc_candidate_balance_and_birth_chain,
+    )
+    from ashtottari_rule_profiles import RULE_FAMILY_RAHU_NOT_IN_KENDRA
 
 # ============================================================================
 # Ashtottari Dasha 数据表
@@ -131,18 +141,13 @@ def calc_ashtottari_dasha(
     moon_nak = nak_data['nakshatra']
     moon_nak_idx = nak_data['nakshatra_idx']
 
-    # 起始守护星
-    starting_lord = NAK_TO_ASHTOTTARI_START.get(moon_nak, 'Sun')
-
-    # 计算起始点的已用/剩余比例
-    nak_span = 360.0 / 27
-    nak_start = moon_nak_idx * nak_span
-    deg_in_nak = moon_longitude - nak_start
-    used_ratio = deg_in_nak / nak_span
-    remaining_ratio = 1 - used_ratio
-
-    start_lord_years = ASHTOTTARI_YEARS.get(starting_lord, 6)
-    first_remaining = start_lord_years * remaining_ratio
+    # PVR Table 39 arc, not the Vimshottari nakshatra lord.
+    starting_lord = _candidate_nakshatra_starting_lord(moon_nak_idx)
+    balance, _birth_chain = _calc_candidate_balance_and_birth_chain(moon_longitude, starting_lord)
+    if balance is None:
+        raise ValueError('Moon longitude does not match the selected Ashtottari arc')
+    start_lord_years = ASHTOTTARI_YEARS[starting_lord]
+    first_remaining = balance['years']
 
     # 构建大运序列
     start_idx = ASHTOTTARI_ORDER.index(starting_lord)
@@ -153,10 +158,10 @@ def calc_ashtottari_dasha(
     sequence.append({
         'lord': starting_lord,
         'full_years': start_lord_years,
-        'remaining_years': round(first_remaining, 4),
-        'elapsed_years': round(start_lord_years - first_remaining, 4),
+        'remaining_years': first_remaining,
+        'elapsed_years': start_lord_years - first_remaining,
         'cumulative_from_birth': 0.0,
-        'cumulative_to_birth': round(first_remaining, 4),
+        'cumulative_to_birth': first_remaining,
         'is_first': True,
     })
     cumulative += first_remaining
@@ -172,8 +177,8 @@ def calc_ashtottari_dasha(
             'full_years': years,
             'remaining_years': years,
             'elapsed_years': 0,
-            'cumulative_from_birth': round(cumulative, 4),
-            'cumulative_to_birth': round(cumulative + years, 4),
+            'cumulative_from_birth': cumulative,
+            'cumulative_to_birth': cumulative + years,
             'is_first': False,
         })
         cumulative += years
@@ -183,10 +188,7 @@ def calc_ashtottari_dasha(
             break
 
     # 出生日期
-    try:
-        birth_date = datetime.strptime(birth_date_str, '%Y-%m-%d')
-    except:
-        birth_date = datetime(2000, 1, 1)
+    birth_date = datetime.fromisoformat(birth_date_str)
 
     # 为大运添加日期范围
     for item in sequence:
@@ -209,8 +211,10 @@ def calc_ashtottari_dasha(
         'rahu_house': rahu_house,
         'moon_nakshatra': moon_nak,
         'starting_lord': starting_lord,
+        'balance_method': 'PVR Table 39 arc fraction',
+        'applicability_rule_family': RULE_FAMILY_RAHU_NOT_IN_KENDRA,
         'total_years': ASHTOTTARI_TOTAL,
-        'first_remaining_years': round(first_remaining, 4),
+        'first_remaining_years': first_remaining,
         'sequence': sequence,
         'note': ('Ashtottari 作为 Vimshottari 的补充系统，'
                  '当 Rahu 不在 Kendra 时提供替代视角。'
@@ -245,12 +249,12 @@ def calc_current_ashtottari(
     if current is None:
         return {'error': f'无法定位年龄 {age} 在 Ashtottari 序列中'}
 
-    years_elapsed_in_md = age - current['cumulative_from_birth']
+    years_elapsed_in_md = age - current['cumulative_from_birth'] + current.get('elapsed_years', 0.0)
     years_remaining = current['cumulative_to_birth'] - age
-    progress = years_elapsed_in_md / current['remaining_years'] if current['remaining_years'] > 0 else 0
+    progress = years_elapsed_in_md / current['full_years'] if current['full_years'] > 0 else 0
 
-    # 计算 Antardasha 子周期（简化：按比例分配）
-    antardasha_sequence = _calc_ashtottari_antardasha(current['lord'], current['remaining_years'])
+    # Divide the original full mahadasha, not the clipped balance at birth.
+    antardasha_sequence = _calc_ashtottari_antardasha(current['lord'], current['full_years'])
 
     # 定位当前 Antardasha
     ad_cumulative = 0
@@ -261,7 +265,7 @@ def calc_current_ashtottari(
             current_ad = {
                 **ad,
                 'progress_in_ad': (years_elapsed_in_md - ad_cumulative) / ad['years'] if ad['years'] > 0 else 0,
-                'remaining_years': round(ad_end - years_elapsed_in_md, 4),
+                'remaining_years': ad_end - years_elapsed_in_md,
             }
             break
         ad_cumulative = ad_end
@@ -281,19 +285,18 @@ def calc_current_ashtottari(
 
 
 def _calc_ashtottari_antardasha(md_lord: str, md_years: float) -> List[Dict]:
-    """计算 Ashtottari Antardasha 子周期"""
+    """PVR section 17.2.2: next lord first, mahadasha lord last."""
     md_start_idx = ASHTOTTARI_ORDER.index(md_lord)
     total = sum(ASHTOTTARI_YEARS.values())
 
     result = []
     for i in range(8):
-        ad_lord = ASHTOTTARI_ORDER[(md_start_idx + i) % 8]
+        ad_lord = ASHTOTTARI_ORDER[(md_start_idx + i + 1) % 8]
         ad_years_total = ASHTOTTARI_YEARS[ad_lord]
-        # 按比例缩放
         ad_years_in_md = md_years * ad_years_total / total
         result.append({
             'lord': ad_lord,
-            'years': round(ad_years_in_md, 4),
+            'years': ad_years_in_md,
             'months': round(ad_years_in_md * 12, 1),
         })
 

tests/test_upstream_sync9_ashtottari_reference_clock.py

@@ -0,0 +1,34 @@
+from datetime import datetime, timedelta
+
+from scripts.ashtottari_dasha import calculate_ashtottari_dasha
+
+
+def calculate(reference):
+    return calculate_ashtottari_dasha({
+        'birth_datetime':'2000-01-01T12:34:56',
+        'moon_nakshatra_index':3, 'is_shukla_paksha':False,
+        'as_of':reference.isoformat()})
+
+
+def test_exact_boundary_uses_next_period_and_preserves_seconds():
+    birth=datetime(2000,1,1,12,34,56)
+    first=calculate(birth)['major'][0]
+    boundary=datetime.fromisoformat(first['end_date'])
+    assert calculate(boundary-timedelta(microseconds=1))['current']['planet']==first['planet']
+    assert calculate(boundary)['current']['planet']!=first['planet']
+    assert calculate(birth-timedelta(seconds=1))['current'] is None
+
+
+def test_second_cycle_children_move_with_their_parent():
+    birth=datetime(2000,1,1,12,34,56)
+    shift=timedelta(days=108*365.25)
+    result=calculate(birth+shift)
+    current=result['current']
+    assert current['cycle_number']==2
+    assert current['start_date']==(birth+shift).isoformat()
+    assert current['antardasha'][0]['start_date']==current['start_date']
+    assert current['antardasha'][-1]['end_date']==current['end_date']
+    for ad in current['antardasha']:
+        assert ad['pratyantardasha'][0]['start_date']==ad['start_date']
+        assert ad['pratyantardasha'][-1]['end_date']==ad['end_date']
+    assert result['major'][0]['start_date']==birth.isoformat()

tests/test_upstream_sync9_ashtottari_source_arcs.py

@@ -0,0 +1,64 @@
+"""PVR Integrated Approach Table 39 and Example 59 (file pages 239–240)."""
+import pytest
+from scripts.ashtottari_dasha import _calc_candidate_balance_and_birth_chain
+
+
+@pytest.mark.parametrize('lord,start,end,years', [
+    ('Sun',200/3,120.,6.), ('Moon',120.,160.,15.),
+    ('Mars',160.,640/3,8.), ('Mercury',640/3,760/3,17.),
+    ('Saturn',760/3,880/3,10.), ('Jupiter',880/3,1000/3,19.),
+    ('Rahu',1000/3,1160/3,12.), ('Venus',80/3,200/3,21.),
+])
+def test_all_eight_published_arcs(lord,start,end,years):
+    at_start,_=_calc_candidate_balance_and_birth_chain(start,lord)
+    at_middle,_=_calc_candidate_balance_and_birth_chain((start+end)/2,lord)
+    assert at_start['years']==pytest.approx(years)
+    assert at_middle['years']==pytest.approx(years/2)
+
+
+@pytest.mark.parametrize('longitude,lord,years', [
+    (144.,'Moon',6.), (0.,'Rahu',6.), (360.,'Rahu',6.),
+    (1000/3,'Rahu',12.), (20.,'Rahu',1.5), (350.,'Rahu',8.25),
+])
+def test_published_arc_balance_including_zero_wrap(longitude,lord,years):
+    balance,_=_calc_candidate_balance_and_birth_chain(longitude,lord)
+    assert balance is not None
+    assert balance['years']==pytest.approx(years)
+    if lord=='Rahu':
+        assert balance['segment_start_nakshatra']=='Uttara Bhadrapada'
+        assert balance['segment_end_nakshatra']=='Bharani'
+        assert balance['segment_span_nakshatras']==4
+
+
+def test_arc_end_belongs_to_next_lord():
+    assert _calc_candidate_balance_and_birth_chain(80/3,'Rahu')==(None,None)
+
+
+def test_auxiliary_entrypoint_uses_pvr_example_not_vimshottari_lords():
+    from scripts.nakshatra_dasha import calc_ashtottari_dasha
+    result=calc_ashtottari_dasha(144.,'2000-01-01',2)
+    assert result['starting_lord']=='Moon'
+    assert result['first_remaining_years']==6.
+    assert [row['lord'] for row in result['sequence'][:4]]==['Moon','Mars','Mercury','Saturn']
+    assert result['balance_method']=='PVR Table 39 arc fraction'
+    assert result['applicability_rule_family']=='rahu_not_in_kendra'
+
+
+def test_auxiliary_entrypoint_rejects_invalid_birth_date():
+    from scripts.nakshatra_dasha import calc_ashtottari_dasha
+    with pytest.raises(ValueError):
+        calc_ashtottari_dasha(144.,'not-a-date',2)
+
+
+def test_auxiliary_birth_balance_keeps_original_child_clock():
+    from scripts.nakshatra_dasha import calc_ashtottari_dasha,calc_current_ashtottari
+    data=calc_ashtottari_dasha(144.,'2000-01-01',2)
+    current=calc_current_ashtottari(data,0.)
+    assert current['years_elapsed_in_md']==9.
+    assert current['mahadasha_progress']==0.6
+    assert current['current_antardasha']['lord']=='Rahu'
+    assert current['current_antardasha']['remaining_years']==pytest.approx(1/6)
+    assert [r['lord'] for r in current['antardasha_sequence']]==[
+        'Mars','Mercury','Saturn','Jupiter','Rahu','Venus','Sun','Moon']
+    assert sum(r['years'] for r in current['antardasha_sequence'])==pytest.approx(15.)
+    assert calc_current_ashtottari(data,6.)['current_mahadasha']=='Mars'