root/Jyotisha
1
0
Fork 0
Code Issues 103 Pull Requests Actions Packages Projects Releases Wiki Activity

merge origin/staging into narrower delivery (只碰文档；显示漏段 BUG-1375 顺延为 BUG-1392, issue #77)

Browse Source 
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LutgzyJqNhPpZy1nFsJtN3
This commit is contained in:
Jesse_Chen
2026-10-19 09:15:55 +08:00
co-authored by Claude Opus 5.5
parent f67d266129 2130682d43
commit e7cab24c1a
169 changed files with 89684 additions and 782 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/_script_root/__init__.py
+15
View File
@@ -0,0 +1,15 @@
"""Put the repository root on sys.path for scripts copied from upstream.
Those modules optionally import ``_script_root`` before their other imports.
This package only adjusts the import path. It does not calculate charts.
It lives in a subdirectory so it does not consume a ``scripts/`` root slot.
"""
from __future__ import annotations
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
root_text = str(ROOT)
if root_text not in sys.path:
sys.path.insert(0, root_text)
scripts/api_heavy_compute_gate.py
+1
View File
@@ -45,6 +45,7 @@ HEAVY_COMPUTE_PATHS = frozenset(
"/api/vedastro/range_scan",
"/api/vedastro_gateway/run",
"/api/thematic_report",
"/api/synastry_full_dossier",
"/api/qizheng",
"/api/western",
"/api/ephemeris_events",
scripts/api_scratch_retention.py
+1
View File
@@ -44,6 +44,7 @@ OFFLINE_SCRATCH_NAMES = frozenset(
"consult_full_data_cache",
# 与 consult_full_data_cache 同走 get_or_build，每次调用先 prune_cache(name=…) 删两天前的文件，自己会清；含出生资料，保留期 2 天。
"report_reference_cache",
"synastry_full_dossier_cache",
"ocr_extract",
"three_engine_parity",
"public_release_deny_patterns.txt",
scripts/argala.py
+134 -14
View File
@@ -7,7 +7,7 @@ Argala（门闩）行星干预模块 v1.1
- 主Argala: 2/4/11宫
- Virodhargala: 12/10/3宫分别阻挡2/4/11宫
- 特殊Argala: 第3宫有2颗以上凶星
- 次级Argala: 5/9宫
- 次级：第5宫是 Argala，第9宫是它的 Virodhargala，不再把第9宫另算一条次级 Argala
- Argala Rajayoga分类: Poornargala / Tripadargala / Ardhargala / Padargala
MIT复用来源：
@@ -62,9 +62,71 @@ def classify_argala_rajayoga(argala_result: Dict) -> Dict:
return {'type': 'None', 'level': 0, 'cn': '无明显Argala'}
def _pair_status(planets, blockers):
if not planets:
return 'absent'
if len(planets) == len(blockers):
return 'strength_comparison_required'
return 'effective' if len(planets) > len(blockers) else 'obstructed'
def _compare_pair(planets, blockers, sign_idx, block_idx, chart, include_nodes):
"""Equal occupancy uses the two Narayana levels already in select_narayana_seed_sign.
Later strength layers are not ported. A remaining tie stays unresolved.
"""
status = _pair_status(planets, blockers)
if status != 'strength_comparison_required':
return status, {'status': 'not_required', 'basis': 'occupant_count'}
metadata = dict(
profile='rath_argala_role_contacts_primary_lords_v1',
source='Rath Jaimini Upadesa Sutra course, source p363',
claim_boundary='First two levels only; traditional primary lords; later ties unresolved.',
)
missing = sorted((NATURAL_BENEFICS | NATURAL_MALEFICS) - set(chart))
if include_nodes:
missing += sorted(NODES - set(chart))
if missing:
return status, dict(metadata, status='blocked', reason='missing_classical_planet_signs', missing=missing)
try:
from scripts.narayana_dasha import select_narayana_seed_sign
except ModuleNotFoundError:
from narayana_dasha import select_narayana_seed_sign
classical = NATURAL_BENEFICS | NATURAL_MALEFICS | (NODES if include_nodes else set())
decision = select_narayana_seed_sign(
sign_idx,
{p: (s % 12) * 30.0 for p, s in chart.items() if p in classical},
comparison_sign_idx=block_idx,
)
decision['implementation_profile'] = decision.pop('profile')
decision.update(metadata)
if decision['status'] == 'selected':
status = 'effective' if decision['selected_sign_idx'] == sign_idx else 'obstructed'
else:
decision['reason'] = 'later_argala_sign_strength_rules_required'
return status, decision
_VERDICT_LABEL = {
'strength_comparison_required': '待定',
'special_argala_adjudication_required': '待定',
'mixed': '吉凶交错',
'supported': '支持',
'obstructed': '受阻',
'neutral': '中性',
}
def calc_argala_for_reference(ref_sign_idx: int, planet_sign_indices: Dict[str, int], include_nodes: bool = False) -> Dict:
"""计算单一参考点的完整Argala。"""
# Ketu in the reference sign reverses counting whether or not nodes count as occupants.
direction = -1 if planet_sign_indices.get('Ketu') == ref_sign_idx else 1
def reference_house(offset):
return (ref_sign_idx + direction * (offset - 1)) % 12
result = {
'source_reference_packets': ['science_of_light_v1_argala_reference_points'],
'ref_sign': SIGNS[ref_sign_idx],
'ref_sign_idx': ref_sign_idx,
'primary': {},
@@ -73,15 +135,20 @@ def calc_argala_for_reference(ref_sign_idx: int, planet_sign_indices: Dict[str,
'argala_count': 0,
'virodhargala_count': 0,
'net_result': 'neutral',
'counting_direction': 'antizodiacal' if direction == -1 else 'zodiacal',
'rule_source': 'PVR Integrated Approach §§10.5–10.7, source pages 117–118',
}
for h in sorted(PRIMARY_ARGALA):
argala_sign = _house_from(ref_sign_idx, h)
argala_sign = reference_house(h)
planets = _planets_in_sign(argala_sign, planet_sign_indices, include_nodes)
block_h = VIRODHARGALA_MAP[h]
block_sign = _house_from(ref_sign_idx, block_h)
block_sign = reference_house(block_h)
blockers = _planets_in_sign(block_sign, planet_sign_indices, include_nodes)
effective = len(planets) > len(blockers)
status, comparison = _compare_pair(
planets, blockers, argala_sign, block_sign, planet_sign_indices, include_nodes,
)
effective = None if status == 'strength_comparison_required' else status == 'effective'
result['primary'][f'H{h}'] = {
'house_from_reference': h,
'sign': SIGNS[argala_sign],
@@ -91,14 +158,16 @@ def calc_argala_for_reference(ref_sign_idx: int, planet_sign_indices: Dict[str,
'blocked_by_sign': SIGNS[block_sign],
'blockers': blockers,
'effective': effective,
'effect': _argala_effect(h, planets, blockers),
'comparison_status': status,
'strength_comparison': comparison,
'effect': _argala_effect(h, planets, blockers, status),
}
if planets:
result['argala_count'] += 1
if blockers:
result['virodhargala_count'] += 1
h3_sign = _house_from(ref_sign_idx, 3)
h3_sign = reference_house(3)
h3_planets = _planets_in_sign(h3_sign, planet_sign_indices, include_nodes)
h3_malefics = [p for p in h3_planets if p in NATURAL_MALEFICS]
result['specific'] = {
@@ -109,25 +178,56 @@ def calc_argala_for_reference(ref_sign_idx: int, planet_sign_indices: Dict[str,
'malefics': h3_malefics,
'effective': len(h3_malefics) >= 2,
'rule': '第3宫有2颗以上天然凶星形成特殊Argala',
'method_status': 'parameter_sensitive_two_malefic_threshold',
'obstruction_rule': 'Once formed, special third-house Argala has no obstruction (Rath source p363). Formation threshold remains profile-dependent.',
}
if result['specific']['effective']:
result['argala_count'] += 1
for h in sorted(SECONDARY_ARGALA):
sign_idx = _house_from(ref_sign_idx, h)
sign_idx = reference_house(h)
planets = _planets_in_sign(sign_idx, planet_sign_indices, include_nodes)
result['secondary'][f'H{h}'] = {
'house_from_reference': h,
'sign': SIGNS[sign_idx],
'sign_idx': sign_idx,
'planets': planets,
'role': 'argala' if h == 5 else 'virodhargala',
}
if result['argala_count'] > result['virodhargala_count']:
fifth, ninth = result['secondary']['H5'], result['secondary']['H9']
secondary_status, comparison = _compare_pair(
fifth['planets'], ninth['planets'], fifth['sign_idx'], ninth['sign_idx'],
planet_sign_indices, include_nodes,
)
fifth.update(
blocked_by_house=9,
blocked_by_sign=ninth['sign'],
strength_comparison=comparison,
blockers=ninth['planets'],
comparison_status=secondary_status,
effective=None if secondary_status == 'strength_comparison_required' else secondary_status == 'effective',
)
ninth['obstructs_house'] = 5
statuses = [row['comparison_status'] for row in result['primary'].values()]
statuses.append(secondary_status)
if 'strength_comparison_required' in statuses:
result['net_result'] = 'strength_comparison_required'
elif 'effective' in statuses and 'obstructed' in statuses:
result['net_result'] = 'mixed'
elif 'effective' in statuses:
result['net_result'] = 'supported'
elif result['argala_count'] < result['virodhargala_count']:
elif 'obstructed' in statuses:
result['net_result'] = 'obstructed'
if result['specific']['effective']:
result['net_result'] = 'special_argala_adjudication_required'
result['verdict_label'] = _VERDICT_LABEL[result['net_result']]
result['rajayoga_classification'] = classify_argala_rajayoga(result)
result['rajayoga_classification']['method_status'] = 'legacy_occupancy_heuristic'
result['rajayoga_classification']['claim_boundary'] = (
'Occupancy label only; not an adjudication of obstruction or a verified Rajayoga.'
)
return result
@@ -167,24 +267,38 @@ def _summarize_argala(houses: Dict) -> Dict:
supported = []
obstructed = []
strongest = []
unresolved = []
mixed = []
for house_key, data in houses.items():
if data['net_result'] == 'supported':
supported.append(house_key)
elif data['net_result'] == 'obstructed':
obstructed.append(house_key)
level = data.get('rajayoga_classification', {}).get('level', 0)
strongest.append((house_key, level, data.get('rajayoga_classification', {}).get('type')))
elif data['net_result'] == 'mixed':
mixed.append(house_key)
elif str(data['net_result']).endswith('_required'):
unresolved.append(house_key)
if house_key not in unresolved:
level = sum(row.get('effective') is True for row in data['primary'].values())
level += int(data['secondary']['H5'].get('effective') is True)
strongest.append((house_key, level, 'effective_pair_count'))
strongest.sort(key=lambda x: x[1], reverse=True)
return {
summary = {
'supported_houses': supported,
'obstructed_houses': obstructed,
'top_argala_houses': [{'house': h, 'level': lvl, 'type': typ} for h, lvl, typ in strongest[:5] if lvl > 0],
'supported_count': len(supported),
'obstructed_count': len(obstructed),
'unresolved_houses': unresolved,
'mixed_houses': mixed,
'ranking_basis': 'effective_pair_count_not_classical_rajayoga_strength',
}
if unresolved:
summary['unresolved_label'] = '待定'
return summary
def _argala_effect(house: int, planets: List[str], blockers: List[str]) -> str:
def _argala_effect(house: int, planets: List[str], blockers: List[str], status=None) -> str:
if not planets:
return '无行星形成此类Argala'
base = {
@@ -192,7 +306,13 @@ def _argala_effect(house: int, planets: List[str], blockers: List[str]) -> str:
4: '内在稳定、住所、母亲、教育与幸福感支持',
11: '收益、人脉、愿望实现与社会网络支持',
}.get(house, '一般支持')
if blockers and len(blockers) >= len(planets):
if blockers and len(blockers) == len(planets):
if status == 'effective':
return f'{base}，双方数量相等，所选强弱口径判为有效'
if status == 'obstructed':
return f'{base}，双方数量相等，所选强弱口径判为受阻'
return f'待定：{base}，双方数量相等，须比较强弱后裁决'
if blockers and len(blockers) > len(planets):
return f'{base}，但被{len(blockers)}个Virodhargala阻挡'
if blockers:
return f'{base}，有轻度阻挡但仍可能有效'
scripts/ashtakavarga.py
+55 -21
View File
@@ -162,6 +162,8 @@ PLANET_WEIGHTS = {
'Sun': 5, 'Moon': 5, 'Mars': 8, 'Mercury': 5,
'Jupiter': 10, 'Venus': 7, 'Saturn': 5
}
# PVR §12.7 表 28。处女座是 6。不改用 Prasna Marga 的 5。
RASHI_PINDA_WEIGHTS = (7, 10, 8, 4, 10, 6, 7, 8, 9, 5, 11, 12)
# 行星列表
SEVEN_PLANETS = ['Sun', 'Moon', 'Mars', 'Mercury', 'Jupiter', 'Venus', 'Saturn']
@@ -431,13 +433,17 @@ def calc_prastara_av(planets: Dict, asc_sign_idx: int) -> Dict:
def calc_yoga_pinda(bav_results: Dict, planets: Dict, asc_sign_idx: int) -> Dict:
"""
计算 Yoga Pinda / Shodhya Pinda 汇总。
"""Yoga Pinda from the same Trikona and Ekadhipatya rows as the Sodhita table.
本项目沿用既有 v2.1 口径：取行星本星座的 BAV bindu，分别乘以星座权重与行星权重；
两者相加作为 Yoga Pinda。该函数把原本嵌在 calc_ashtakavarga() 内部的结果提升为
一等契约，便于 API、前端与测试直接引用。
PVR §12.7 tables 28–29: twelve reduced signs times rashi weights (Virgo 6),
plus the seven planets' weights at the signs they occupy. A full seven-row
chart reuses ``sodhita_chain`` so the report cannot show two reductions.
"""
try:
from ashtakavarga_shodhana import occupied_signs, sodhita_chain
except ImportError:
from scripts.ashtakavarga_shodhana import occupied_signs, sodhita_chain
source_sign_idx = {}
for pname in SEVEN_PLANETS:
if pname in planets:
@@ -445,27 +451,51 @@ def calc_yoga_pinda(bav_results: Dict, planets: Dict, asc_sign_idx: int) -> Dict
if sign in SIGNS:
source_sign_idx[pname] = SIGNS.index(sign)
source_sign_idx['Lagna'] = asc_sign_idx
note = (
'PVR §12.7 表 28–29：十二座削减值乘星座权重（处女座 6），'
'再加七曜所在座的行星权重。与 Sodhita 表同一份削减。'
'同值且一占一空时按 Raman 15(c) 清空空宫。'
)
if any(pname not in source_sign_idx for pname in SEVEN_PLANETS):
return {
'method': 'Yoga Pinda', 'version': '2.0',
'profile': 'pvr_trikona_ekadhipatya_tables28_29',
'planets': {}, 'summary': {}, 'all_valid': False,
'validation_note': note + ' 七曜星座不全，不算 Graha Pinda。',
}
available = {}
for pname, row in bav_results.items():
bindus = row.get('bindus') if isinstance(row, dict) else None
if isinstance(bindus, list) and len(bindus) == 12:
available[pname] = {'bindus': bindus}
if set(SEVEN_PLANETS).issubset(available):
chain = sodhita_chain(available, planets)
trikona_rows = chain['trikona_bav']
reduced_rows = chain['sodhita_bav']
else:
trikona_rows = {
pname: calc_trikona_shodhana(row['bindus'])['reduced_scores']
for pname, row in available.items()
}
reduced_rows = calc_ekadhipatya_shodhana(trikona_rows, occupied_signs(planets))['rows']
rows = {}
all_valid = True
for pname in SEVEN_PLANETS:
bav_row = bav_results.get(pname, {})
bindus = bav_row.get('bindus', [])
reduced = reduced_rows.get(pname)
own_sign = source_sign_idx.get(pname)
if own_sign is None or len(bindus) != 12:
bindus = (available.get(pname) or {}).get('bindus')
if reduced is None or own_sign is None or bindus is None:
all_valid = False
continue
bindu_at_own = bindus[own_sign]
sign_weight = 4
planet_weight = PLANET_WEIGHTS.get(pname, 5)
rashi_pinda = bindu_at_own * sign_weight
graha_pinda = bindu_at_own * planet_weight
rashi_pinda = sum(value * weight for value, weight in zip(reduced, RASHI_PINDA_WEIGHTS))
graha_pinda = sum(
reduced[source_sign_idx[body]] * PLANET_WEIGHTS[body] for body in SEVEN_PLANETS
)
rows[pname] = {
'sign': SIGNS[own_sign],
'bindu_at_own_sign': bindu_at_own,
'rashi_weight': sign_weight,
'planet_weight': planet_weight,
'bindu_at_own_sign': bindus[own_sign],
'rashi_pinda': rashi_pinda,
'graha_pinda': graha_pinda,
'yoga_pinda': rashi_pinda + graha_pinda,
@@ -475,10 +505,14 @@ def calc_yoga_pinda(bav_results: Dict, planets: Dict, asc_sign_idx: int) -> Dict
strongest = ranked[0][0] if ranked else None
weakest = ranked[-1][0] if ranked else None
total = sum(item['yoga_pinda'] for item in rows.values())
return {
'method': 'Yoga Pinda',
'version': '1.0',
'version': '2.0',
'profile': 'pvr_trikona_ekadhipatya_tables28_29',
'rashi_weights': list(RASHI_PINDA_WEIGHTS),
'planet_weights': dict(PLANET_WEIGHTS),
'trikona_rows': trikona_rows,
'sodhita_rows': reduced_rows,
'planets': rows,
'summary': {
'strongest_planet': strongest,
@@ -487,7 +521,7 @@ def calc_yoga_pinda(bav_results: Dict, planets: Dict, asc_sign_idx: int) -> Dict
'average_yoga_pinda': round(total / len(rows), 2) if rows else 0,
},
'all_valid': all_valid and len(rows) == len(SEVEN_PLANETS),
'validation_note': 'Yoga Pinda = Rashi Pinda + Graha Pinda；当前复用既有 v2.1 Shodhya Pinda 权重口径。',
'validation_note': note,
}
@@ -569,5 +603,5 @@ def calc_ekadhipatya_shodhana(
'occupied_signs': sorted(occupied_signs),
'audit': audit,
'status': 'executed',
'claim_boundary': 'Applied after Trikona Shodhana; source profile follows PyJHora/PVR rule ordering.',
'claim_boundary': 'Applied after Trikona Shodhana. One occupied and one vacant sign with the same value clears the vacant sign (Raman 15(c)).',
}
scripts/ashtakoot.py
+2 -1
View File
@@ -364,7 +364,7 @@ def _planet_dignity_level(planet_name: str, sign: str) -> str:
def calc_kuja_dosha(chart: Dict) -> Dict:
"""计算完整的 Kuja Dosha（火星凶相分析）"""
"""旧版 Kuja Dosha（五星合分）。新功能不得调用；婚恋合盘用 marriage_match。"""
planets = chart.get("planets", {})
dosha_scores_high = {
@@ -408,6 +408,7 @@ def calc_kuja_dosha(chart: Dict) -> Dict:
def match_kuja_dosha(male_score: float, female_score: float) -> Dict:
"""旧版：按性别比较火星冲克分数。新功能不得调用。"""
diff = male_score - female_score
if abs(diff) <= 5:
return {"result": "good", "description": "Kuja Dosha balanced."}
scripts/avastha_calculator.py
+170 -47
View File
@@ -19,6 +19,60 @@ Avastha Calculator - 行星状态计算器
from typing import Dict, List, Tuple
from enum import Enum
import json
import math
from bisect import bisect_right
from pathlib import Path
with (Path(__file__).resolve().parents[1] / 'references' / 'shat_bala_constants.json').open(encoding='utf-8') as _handle:
NATURAL_RELATIONSHIPS = json.load(_handle)['natural_relationships']
SIGNS = ['Aries', 'Taurus', 'Gemini', 'Cancer', 'Leo', 'Virgo',
'Libra', 'Scorpio', 'Sagittarius', 'Capricorn', 'Aquarius', 'Pisces']
def calculate_pvr_shayanadi(planet, longitude, moon_longitude, asc_longitude, hours_since_sunrise):
"""PVR §15.4.4/Table 36; running ghati, not elapsed fractional ghatis.
Activity intensity requires the name's Sanskrit first-sound index and is
deliberately not guessed from an English display name.
"""
order = ('Sun', 'Moon', 'Mars', 'Mercury', 'Jupiter', 'Venus', 'Saturn', 'Rahu', 'Ketu')
if planet not in order:
raise ValueError('Unsupported Shayanadi planet')
values = (longitude, moon_longitude, asc_longitude, hours_since_sunrise)
if any(isinstance(value, bool) or not math.isfinite(float(value)) for value in values):
raise ValueError('Finite numeric Shayanadi inputs required')
if hours_since_sunrise < 0:
raise ValueError('Use the preceding sunrise, not a future sunrise')
longitude, moon_longitude, asc_longitude = (float(value) % 360 for value in values[:3])
boundaries = [i * 40 / 3 for i in range(1, 27)]
c = bisect_right(boundaries, longitude) + 1
p = order.index(planet) + 1
a = bisect_right([i * 10 / 3 for i in range(1, 9)], longitude % 30) + 1
m = bisect_right(boundaries, moon_longitude) + 1
g = math.floor(hours_since_sunrise * 2.5) + 1
l = int(asc_longitude // 30) + 1
total = c * p * a + m + g + l
index = (total - 1) % 12 + 1
states = ('Shayana', 'Upaveshana', 'Netrapani', 'Prakashana', 'Gamana', 'Agamana',
'Sabha', 'Agama', 'Bhojana', 'Nrityalipsa', 'Kautuka', 'Nidra')
return {
'state': states[index - 1],
'state_index': index,
'sign': SIGNS[int(longitude // 30)],
'meaning': 'Activity state from the PVR constellation, navamsa and running-ghati formula.',
'method_status': 'pvr_15_4_4_running_ghati',
'components': {
'planet_nakshatra': c,
'planet_index': p,
'navamsa_part': a,
'moon_nakshatra': m,
'running_ghati': g,
'lagna_sign_number': l,
'formula_sum': total,
},
'activity_intensity_status': 'requires_name_first_sound_index',
}
class BalaAvastha(Enum):
"""年龄状态（5种生命阶段）"""
@@ -39,11 +93,12 @@ class DeeptadiAvastha(Enum):
DEEPTA = "Deepta" # 光辉（Exalted）
SWASTHA = "Swastha" # 舒适（Own Sign）
MUDITA = "Mudita" # 快乐（Friend's Sign）
SHANTA = "Shanta" # 平静（Neutral Sign）
DINA = "Dina" # 沮丧（Enemy Sign）
VIKALA = "Vikala" # 残缺（Debilitated）
KHALA = "Khala" # 恶意（Combust）
KOPA = "Kopa" # 愤怒（Defeated in War）
SHANTA = "Shanta" # 平静（Friend's Sign）
DINA = "Dina" # 低落（Neutral Sign）
DUHKHITA = "Duhkhita" # 困苦（Enemy Sign）
VIKALA = "Vikala" # 残缺（Joined by malefics）
KHALA = "Khala" # 负担（Debilitated; Science of Light）
KOPA = "Kopa" # 愤怒（Combust）
GARVITA = "Garvita" # 骄傲（Victorious in War）
KSHUDITA = "Kshudita" # 饥饿（In Sandhi）
TRUSHITA = "Trushita" # 口渴（In Gandanta）
@@ -127,7 +182,7 @@ class AvasthaCalculator:
def calculate_all_avasthas(self, planet: str, sign: str, degree: float,
house: int, conjunctions: List[str] = None,
aspects: List[str] = None, is_combust: bool = False,
is_retrograde: bool = False) -> Dict[str, any]:
is_retrograde: bool = False, activity_context: Dict = None) -> Dict[str, any]:
"""
计算行星的所有Avastha状态
@@ -153,7 +208,7 @@ class AvasthaCalculator:
}
# 1. Bala Avastha（年龄状态）
result["avasthas"]["Bala"] = self._calculate_bala_avastha(degree)
result["avasthas"]["Bala"] = self._calculate_bala_avastha(degree, sign)
# 2. Jagrat Avastha（警觉状态）
result["avasthas"]["Jagrat"] = self._calculate_jagrat_avastha(planet, sign, house)
@@ -170,7 +225,34 @@ class AvasthaCalculator:
# 5. Shayanadi Avastha（姿态状态）
result["avasthas"]["Shayanadi"] = self._calculate_shayanadi_avastha(planet, sign)
for name in ('Deeptadi', 'Lajjitadi', 'Shayanadi'):
result['avasthas'][name]['method_status'] = 'simplified_heuristic'
if activity_context is not None and activity_context.get('unavailable_reason'):
# 极昼极夜等算不出日出时，PVR 公式缺一项。不退回元素启发式冒充，标成无法计算（BUG-1390）。
result['avasthas']['Shayanadi'] = {
'state': None,
'sign': sign,
'meaning': '当天没有可用的日出（极昼或极夜），姿态无法按 PVR 公式计算。',
'method_status': 'blocked',
'blocked_reason': activity_context['unavailable_reason'],
}
elif activity_context is not None:
result['avasthas']['Shayanadi'] = calculate_pvr_shayanadi(
planet,
SIGNS.index(sign) * 30 + degree,
activity_context['moon_longitude'],
activity_context['asc_longitude'],
activity_context['hours_since_sunrise'],
)
result['inputs'] = {
'conjunctions': conjunctions or [],
'aspects': aspects or [],
'is_combust': is_combust,
'is_retrograde': is_retrograde,
}
if activity_context is not None:
result['inputs']['activity_context'] = dict(activity_context)
# 6. 综合解读
result["interpretation"] = self._generate_interpretation(result["avasthas"])
@@ -179,11 +261,17 @@ class AvasthaCalculator:
return result
def _calculate_bala_avastha(self, degree: float) -> Dict:
def _calculate_bala_avastha(self, degree: float, sign: str = 'Aries') -> Dict:
"""计算Bala Avastha（年龄状态）"""
if not 0 <= degree < 30:
raise ValueError('Avastha degree must be within [0, 30)')
original_degree = degree
even_sign = SIGNS.index(sign) % 2 == 1
band = int(degree // 6)
degree = (4 - band if even_sign else band) * 6
if degree < 6:
state = BalaAvastha.INFANT
strength = 0.2
strength = 0.25
meaning = "婴儿期：行星力量微弱，潜力未开发，需要时间成长"
elif degree < 12:
state = BalaAvastha.YOUNG
@@ -195,40 +283,46 @@ class AvasthaCalculator:
meaning = "青年期：行星力量最强，充满活力，能够充分发挥作用"
elif degree < 24:
state = BalaAvastha.OLD
strength = 0.7
strength = None
meaning = "老年期：行星力量开始衰退，经验丰富但活力下降"
else:
state = BalaAvastha.DEAD
strength = 0.3
strength = 0.0
meaning = "死亡期：行星力量极弱，难以发挥作用，需要补救"
return {
"state": state.value,
"degree_range": f"{int(degree//6)*6}-{int(degree//6)*6+6}°",
"degree_range": f"{int(original_degree//6)*6}-{int(original_degree//6)*6+6}°",
"strength": strength,
"result_grade": 'negligible' if state == BalaAvastha.OLD else 'fraction',
"sign_parity": 'even' if even_sign else 'odd',
"method_status": 'bphs_age_sign_parity',
"meaning": meaning
}
def _calculate_jagrat_avastha(self, planet: str, sign: str, house: int) -> Dict:
"""计算Jagrat Avastha（警觉状态）"""
# 简化规则：
# Awake: 在角宫（1,4,7,10）
# Dreaming: 在续宫（2,5,8,11）
# Deep Sleep: 在果宫（3,6,9,12）
if house in [1, 4, 7, 10]:
if planet not in self.PLANET_SIGNS:
return {
'state': 'Not classified',
'house': house,
'method_status': 'node_dignity_not_assigned',
'meaning': '交点未指定尊贵度流派，不套用七曜警觉状态规则',
}
if self._is_own_sign(planet, sign) or self._is_exalted(planet, sign, 0):
state = JagratAvastha.AWAKE
meaning = "清醒状态：行星完全活跃，能够主动发挥作用，影响显著"
elif house in [2, 5, 8, 11]:
elif not self._is_enemy_sign(planet, sign) and not self._is_debilitated(planet, sign, 0):
state = JagratAvastha.DREAMING
meaning = "梦境状态：行星半活跃，作用不稳定，时强时弱"
else:
state = JagratAvastha.DEEP_SLEEP
meaning = "深睡状态：行星不活跃，难以发挥作用，影响微弱"
return {
"state": state.value,
"house": house,
"method_status": 'bphs_dignity_natural_relationships',
"meaning": meaning
}
@@ -236,29 +330,50 @@ class AvasthaCalculator:
is_combust: bool, aspects: List[str] = None) -> Dict:
"""计算Deeptadi Avastha（情绪状态）"""
aspects = aspects or []
# 检查入庙/落陷
if planet not in self.PLANET_SIGNS:
return {
'state': 'Not classified',
'meaning': '该口径未定义此天体的状态',
'aspect_effects': [],
'label_profile': 'science_of_light_deeptadi_natural_relationship_subset',
'claim_boundary': 'Classical seven-planet profile; no nodal dignity is assumed.',
}
conditions = {
'exalted': self._is_exalted(planet, sign, degree),
'debilitated': self._is_debilitated(planet, sign, degree),
'combust': bool(is_combust),
'own_sign': self._is_own_sign(planet, sign),
'natural_friend_sign': self._is_friend_sign(planet, sign),
'natural_enemy_sign': self._is_enemy_sign(planet, sign),
}
applicable_states = [label for key, label in (
('exalted', 'Deepta'), ('debilitated', 'Khala'), ('combust', 'Kopa'),
('own_sign', 'Swastha'), ('natural_friend_sign', 'Shanta'),
('natural_enemy_sign', 'Duhkhita')) if conditions[key]]
if not any(conditions[key] for key in ('own_sign', 'natural_friend_sign', 'natural_enemy_sign')):
applicable_states.append('Dina')
# 单标签顺序是兼容字段，无经典出处。并存条件全部留在 applicable_states。
if self._is_exalted(planet, sign, degree):
state = DeeptadiAvastha.DEEPTA
meaning = "光辉状态：行星处于最佳位置，能量充沛，带来积极结果"
elif self._is_debilitated(planet, sign, degree):
state = DeeptadiAvastha.VIKALA
meaning = "残缺状态：行星处于最弱位置，能量受损，带来挑战"
elif is_combust:
state = DeeptadiAvastha.KHALA
meaning = "恶意状态：行星被太阳燃烧，能量被压制，难以正常发挥"
meaning = "负担状态：行星落陷；采用 Science of Light 的 Khala 定义"
elif is_combust:
state = DeeptadiAvastha.KOPA
meaning = "愤怒状态：行星受太阳燃烧；对应 Kopa"
elif self._is_own_sign(planet, sign):
state = DeeptadiAvastha.SWASTHA
meaning = "舒适状态：行星在自己的星座，感觉自在，能够稳定发挥"
elif self._is_friend_sign(planet, sign):
state = DeeptadiAvastha.MUDITA
meaning = "快乐状态：行星在友好星座，得到支持，能够良好发挥"
elif self._is_enemy_sign(planet, sign):
state = DeeptadiAvastha.DINA
meaning = "沮丧状态：行星在敌对星座，受到压制，发挥受限"
else:
state = DeeptadiAvastha.SHANTA
meaning = "平静状态：行星在中性星座，不强不弱，平稳发挥"
meaning = "平静状态：行星在自然友星的星座；对应 Shanta"
elif self._is_enemy_sign(planet, sign):
state = DeeptadiAvastha.DUHKHITA
meaning = "困苦状态：行星在自然敌星的星座；对应 Duhkhita"
else:
state = DeeptadiAvastha.DINA
meaning = "低落状态：行星在自然中性星的星座；对应 Dina"
# 检查相位影响
aspect_effects = []
@@ -274,7 +389,15 @@ class AvasthaCalculator:
return {
"state": state.value,
"meaning": meaning,
"aspect_effects": aspect_effects
"aspect_effects": aspect_effects,
"label_profile": "science_of_light_deeptadi_natural_relationship_subset",
"conditions": conditions,
"applicable_states": applicable_states,
"legacy_single_label_note": "无经典出处",
"uncomputed_conditions": ["great_friend", "malefic_conjunction"],
"uncomputed_label": "未计算",
"source": "Science of Light, source p475; Khala is debility in this profile",
"claim_boundary": "Single-label priority is a legacy heuristic. Natural friendship only; great-friend and simultaneous conjunction states are not fully calculated. Khala follows Science of Light debility, not the alternative PVR malefic-sign definition.",
}
def _calculate_lajjitadi_avastha(self, planet: str, sign: str, house: int,
@@ -357,14 +480,14 @@ class AvasthaCalculator:
"""检查是否入庙"""
if planet in self.EXALTATION:
exalt_sign, exalt_degree = self.EXALTATION[planet]
return sign == exalt_sign and abs(degree - exalt_degree) < 5
return sign == exalt_sign
return False
def _is_debilitated(self, planet: str, sign: str, degree: float) -> bool:
"""检查是否落陷"""
if planet in self.DEBILITATION:
debil_sign, debil_degree = self.DEBILITATION[planet]
return sign == debil_sign and abs(degree - debil_degree) < 5
return sign == debil_sign
return False
def _is_own_sign(self, planet: str, sign: str) -> bool:
@@ -372,14 +495,14 @@ class AvasthaCalculator:
return sign in self.PLANET_SIGNS.get(planet, [])
def _is_friend_sign(self, planet: str, sign: str) -> bool:
"""检查是否在友好星座（简化版）"""
# 这里需要完整的行星友好关系表，暂时简化
return False
"""检查是否在自然友星星座。"""
lord = next((name for name, signs in self.PLANET_SIGNS.items() if sign in signs), None)
return lord in NATURAL_RELATIONSHIPS.get(planet, {}).get('friends', [])
def _is_enemy_sign(self, planet: str, sign: str) -> bool:
"""检查是否在敌对星座（简化版）"""
# 这里需要完整的行星敌对关系表，暂时简化
return False
"""检查是否在自然敌星星座。"""
lord = next((name for name, signs in self.PLANET_SIGNS.items() if sign in signs), None)
return lord in NATURAL_RELATIONSHIPS.get(planet, {}).get('enemies', [])
def _is_enemy_planet(self, planet1: str, planet2: str) -> bool:
"""检查两个行星是否敌对（简化版）"""
@@ -426,7 +549,7 @@ class AvasthaCalculator:
# Shayanadi Avastha
shayanadi = avasthas["Shayanadi"]
interpretation += f"5. 行为模式：{shayanadi['state']}（{shayanadi['sign']}）\n"
interpretation += f"5. 行为模式：{shayanadi['state'] or '无法计算'}（{shayanadi['sign']}）\n"
interpretation += f" {shayanadi['meaning']}\n"
return interpretation
@@ -437,7 +560,7 @@ class AvasthaCalculator:
# 根据Bala Avastha
bala_strength = avasthas["Bala"]["strength"]
if bala_strength < 0.5:
if bala_strength is not None and bala_strength < 0.5:
remedies.append(f"行星处于弱势年龄阶段，建议佩戴{planet}对应的宝石")
remedies.append(f"在{planet}主管的日子进行布施和祈祷")
@@ -447,7 +570,7 @@ class AvasthaCalculator:
# 根据Deeptadi Avastha
deeptadi_state = avasthas["Deeptadi"]["state"]
if deeptadi_state in ["Vikala", "Khala", "Dina"]:
if deeptadi_state in ["Vikala", "Khala", "Duhkhita", "Kopa"]:
remedies.append(f"行星处于{deeptadi_state}状态，需要特别补救")
remedies.append(f"念诵{planet}的Beeja Mantra（种子咒）")
scripts/bhava_bala.py
+34 -33
View File
@@ -35,36 +35,34 @@ SIGN_NATURE = {
# 宫位力量基于星座性质（BPHS标准，单位Virupas）
# 索引0-11对应第1-12宫
BHAVA_STRENGTH_FROM_SIGN_NATURE = {
'nara': [30, 20, 10, 20, 10, 20, 30, 20, 10, 20, 10, 20],
'chatushpada': [30, 20, 10, 20, 10, 20, 30, 20, 10, 20, 10, 20],
'jalachara': [20, 30, 10, 30, 10, 30, 20, 30, 10, 30, 10, 30],
'keeta': [10, 20, 30, 20, 30, 20, 10, 20, 30, 20, 30, 20],
'vanaspati': [20, 10, 20, 10, 20, 30, 20, 10, 20, 10, 20, 30],
'nara': [60, 50, 40, 30, 20, 10, 0, 10, 20, 30, 40, 50],
'chatushpada': [30, 20, 10, 0, 10, 20, 30, 40, 50, 60, 50, 40],
'jalachara': [30, 40, 50, 60, 50, 40, 30, 20, 10, 0, 10, 20],
'keeta': [0, 10, 20, 30, 40, 50, 60, 50, 40, 30, 20, 10],
}
# Sputa Drishti值表（基于角距离的连续相位力量）
def _get_sputa_drishti(angle_diff: float) -> float:
"""基于jxotishganit MIT算法，将精确角距离转换为Sputa Drishti值(0-60)"""
ad = angle_diff % 360
def _get_sputa_drishti(angle_diff: float, planet: str = 'Sun') -> float:
"""jyotishganit 镜像角公式。火、木、土在特殊相位上另有一段。不是本仓方向性 Shadbala。"""
ad = abs(angle_diff) % 360
if ad > 180:
ad = 360 - ad
if ad <= 1.0: return 60.0
if ad <= 3.0: return 57.0
if ad <= 7.0: return 51.0
if ad <= 10.0: return 45.0
if ad <= 15.0: return 40.5
if ad <= 20.0: return 30.0
if ad <= 25.0: return 22.5
if ad <= 30.0: return 15.0
if ad <= 40.0: return 12.0
if ad <= 50.0: return 9.0
if ad <= 60.0: return 6.0
if ad <= 75.0: return 4.5
if ad <= 90.0: return 3.0
if ad <= 120.0: return 2.25
if ad <= 150.0: return 1.5
return 0.0
if ad < 30:
strength = 0.0
elif ad < 60:
strength = (ad - 30) / 2
elif ad < 90:
strength = ad - 45
elif ad < 120:
strength = 30 + (120 - ad) / 2
elif ad < 150:
strength = 150 - ad
else:
strength = 2 * (ad - 150)
for special in {'Mars': (90, 210), 'Jupiter': (120, 240), 'Saturn': (60, 270)}.get(planet, ()):
distance = abs(ad - special)
if distance <= 15:
strength = max(strength, 60 * (1 - distance / 15))
return strength
def calc_bhava_adhipathi_bala(house_signs: List[str], planet_shadbala: Dict) -> List[float]:
@@ -109,8 +107,8 @@ def calc_bhava_dig_bala(house_signs: List[str], house_degrees: List[float]) -> L
# 处理双性星座
if 'first_half' in nature_key:
# Sagittarius前半=nara, Capricorn前半=jalachara
real_nature = 'nara' if 'Sagittarius' in nature_key else 'jalachara'
# Sagittarius 前半是人象，Capricorn 前半是四足，不是水象。
real_nature = 'nara' if 'Sagittarius' in nature_key else 'chatushpada'
elif 'second_half' in nature_key:
real_nature = 'chatushpada' if 'Sagittarius' in nature_key else 'jalachara'
else:
@@ -143,7 +141,7 @@ def calc_bhava_drik_bala(house_signs: List[str], house_degrees: List[float],
Returns:
12个宫位的drik bala值
"""
BENEFICS = ['Jupiter', 'Venus', 'Mercury']
BENEFICS = ['Jupiter', 'Venus', 'Moon']
MALEFICS = ['Saturn', 'Mars', 'Sun']
asc_sign_idx = SIGNS.index(asc_sign) if asc_sign in SIGNS else 0
@@ -175,7 +173,7 @@ def calc_bhava_drik_bala(house_signs: List[str], house_degrees: List[float],
dist_deg = abs(diff)
# Sputa Drishti值
sputa = _get_sputa_drishti(dist_deg)
sputa = _get_sputa_drishti(dist_deg, planet_name)
if planet_name in BENEFICS:
benefic_sputa += sputa
@@ -238,8 +236,11 @@ def calc_bhava_bala(house_signs: List[str], house_degrees: List[float],
})
return {
'method': 'Bhava Bala三元力量（基于jyotishganit MIT算法）',
'version': '1.0',
'method': 'jyotishganit_mirrored_equal_house_v2',
'version': '2.0',
'score_unit': 'Virupas',
'method_boundary': '具名实现：镜像相位、吉星固定为月亮木星金星、水星中性、上升等宫中点。不是本仓方向性 Shadbala，也不是经典定论。',
'claim_boundary': 'Bhava Bala 只是三分量原始分数，不能单独证明某项人生结果。',
'houses': houses,
'strongest_house': max(houses, key=lambda h: h['total'])['house'],
'weakest_house': min(houses, key=lambda h: h['total'])['house'],
@@ -281,7 +282,7 @@ def bhava_bala_from_chart(planets: Dict, asc_sign: str, asc_degree: float, shadb
house['lord'] = SIGN_LORDS.get(house['sign'])
house['total_rupas'] = round(house['total'] / 60.0, 2)
house.pop('level', None)
result['method'] = 'Bhava Bala 三分量：Bhavadhipati（宫主 Shadbala）+ Bhava Dig + Bhava Drishti（jyotishganit MIT 算法，等宫）'
result['method'] = 'jyotishganit_mirrored_equal_house_v2：宫主 Shadbala + Dig + Drishti。吉星为月亮、木星、金星，水星中性。不是经典定论。'
result['unit'] = 'virupa'
result['source'] = 'scripts/bhava_bala.py'
return result
scripts/classical_astrology.py
+13065
View File
File diff suppressed because it is too large Load Diff
scripts/compatibility_skill_readiness_dashboard.py
+1 -1
View File
@@ -61,7 +61,7 @@ def build() -> dict:
"partial",
"basic_safe_with_boundary",
"36-point compatibility can be exposed as one factor only; not deterministic marriage outcome.",
["references/oracle/ashtakoot_oracle_cases.json", "scripts/synastry.py"],
["references/oracle/ashtakoot_oracle_cases.json", "scripts/marriage_match.py"],
),
layer(
"mangal_dosha",
scripts/dasha_analyzer.py
+12 -6
View File
@@ -68,12 +68,18 @@ def fuzzy_match_nakshatra(name):
return None
def lon_to_nakshatra(moon_lon):
"""月亮黄经（0-360）→ Nakshatra + Pada"""
nak_span = 360 / 27
pada_span = nak_span / 4
idx = int(moon_lon / nak_span)
progress = (moon_lon % nak_span) / nak_span
pada = int((moon_lon % nak_span) / pada_span) + 1
"""月亮黄经（0-360）→ Nakshatra + Pada。序号与进度来自同一把尺子。"""
try:
from nakshatra_advanced import nakshatra_index_and_progress
except ImportError:
from scripts.nakshatra_advanced import nakshatra_index_and_progress
idx, progress = nakshatra_index_and_progress(moon_lon)
if progress == 0.0:
pada = 1
else:
nak_span = 360 / 27
pada_span = nak_span / 4
pada = int((float(moon_lon) % nak_span) / pada_span) + 1
return NAKSHATRA_LIST[idx], progress, pada
def build_dasha_timeline(birth_date_str, nak_info, nak_progress):
scripts/gulika.py
+4 -8
View File
@@ -89,13 +89,11 @@ def calculate_gulika(
) -> dict[str, Any]:
"""Return Gulika from local moment/location using Swiss sunrise and sunset."""
ayanamsa_name = normalize_ayanamsa_name(ayanamsa)
daynight = determine_daytime(moment, lat=lat, lon=lon, tz=tz)
daynight = determine_daytime(moment, lat=lat, lon=lon, tz=tz, rising="hindu")
is_day = bool(daynight["is_daytime"])
period = "day" if is_day else "night"
start_jd = daynight["sunrise_jd_ut"] if is_day else daynight["sunset_jd_ut"]
end_jd = daynight["sunset_jd_ut"] if is_day else daynight["sunrise_jd_ut"] + 1.0
if end_jd <= start_jd:
end_jd += 1.0
end_jd = daynight["sunset_jd_ut"] if is_day else daynight["next_sunrise_jd_ut"]
start_year, start_month, start_day, _ = swe.revjul(start_jd + float(tz) / 24.0)
period_weekday = datetime(int(start_year), int(start_month), int(start_day)).weekday()
if method == "saturn_part_start":
@@ -143,13 +141,11 @@ def _temporal_upagraha(
ayanamsa: str = DEFAULT_AYANAMSA_NAME,
) -> dict[str, Any]:
ayanamsa_name = normalize_ayanamsa_name(ayanamsa)
daynight = determine_daytime(moment, lat=lat, lon=lon, tz=tz)
daynight = determine_daytime(moment, lat=lat, lon=lon, tz=tz, rising="hindu")
is_day = bool(daynight["is_daytime"])
period = "day" if is_day else "night"
start_jd = daynight["sunrise_jd_ut"] if is_day else daynight["sunset_jd_ut"]
end_jd = daynight["sunset_jd_ut"] if is_day else daynight["sunrise_jd_ut"] + 1.0
if end_jd <= start_jd:
end_jd += 1.0
end_jd = daynight["sunset_jd_ut"] if is_day else daynight["next_sunrise_jd_ut"]
start_year, start_month, start_day, _ = swe.revjul(start_jd + float(tz) / 24.0)
weekday = datetime(int(start_year), int(start_month), int(start_day)).weekday()
ruler_table = DAY_RULERS if is_day else NIGHT_RULERS
scripts/jyotish_api_server.py
+16 -1
View File
@@ -2534,7 +2534,8 @@ def _load_local_module(module_name):
cached = _LOCAL_MODULE_CACHE.get(module_name)
if cached:
return cached
module_path = os.path.join(SCRIPTS_DIR, f'{module_name}.py')
relative = module_name.replace('.', os.sep)
module_path = os.path.join(SCRIPTS_DIR, f'{relative}.py')
spec = importlib.util.spec_from_file_location(f'_jyotish_local_{module_name}', module_path)
if not spec or not spec.loader:
raise ImportError(f'Cannot load local module: {module_name}')
@@ -10505,6 +10506,20 @@ class JyotishAPIHandler(BaseHTTPRequestHandler, JyotishComputeService):
elif path == '/api/synastry':
result = self._compute_synastry(body)
self._json(result)
elif path == '/api/marriage-match':
match_api = _load_local_module('marriage_match_service.api')
try:
result = match_api.compute_marriage_match(body)
except match_api.MarriageMatchInputError as exc:
raise BadRequest(str(exc)) from exc
self._json(result)
elif path == '/api/synastry_full_dossier':
match_api = _load_local_module('marriage_match_service.api')
try:
result = match_api.compute_synastry_full_dossier(self, body)
except match_api.MarriageMatchInputError as exc:
raise BadRequest(str(exc)) from exc
self._json(result)
elif path == '/api/dasha':
result = self._compute_dasha_system(body)
self._json(result)
scripts/jyotish_engine.py
+159 -41
View File
@@ -16,7 +16,7 @@
nakshatra-adv 高级Nakshatra分析（Tara Bala/Sub-Lord/兼容性）（v3.7新增）
argala Argala门闩系统（v3.7新增）
tajika Tajika/Varshaphala年运盘（v3.7新增）
synastry 合盘分析（Ashta Koota 36分制+Mangal Dosha）（v3.7新增）
marriage-match 婚恋合盘（两份盘快照；七项 35 分，火星冲克与婚姻宫压力分开列）
full-reading 全自动综合解盘（出生信息→全链路计算→完整报告）（v3.7.1新增）
celebrity 名人案例查询（15,807条数据+SQLite验证库）
db-stats 验证数据库统计
@@ -435,6 +435,52 @@ def _calc_pushkara_flags(planets):
return result
def _calc_lagna_vargottamamsa(report, varga_full):
chart = report.get('modules', {}).get('chart', {}) if isinstance(report, dict) else {}
ascendant = chart.get('ascendant', {}) if isinstance(chart, dict) else {}
d1_sign = ascendant.get('sign') if isinstance(ascendant, dict) else None
d9 = varga_full.get('D9_Navamsa', {}) if isinstance(varga_full, dict) else {}
d9_ascendant = d9.get('ascendant') if isinstance(d9, dict) else None
if not isinstance(d9_ascendant, dict) and isinstance(d9, dict):
d9_ascendant = d9.get('Ascendant', {})
d9_sign = d9_ascendant.get('sign') if isinstance(d9_ascendant, dict) else None
return {
'd1_sign': d1_sign,
'd9_sign': d9_sign,
'is_lagna_vargottamamsa': bool(d1_sign and d9_sign and d1_sign == d9_sign),
'source': 'raman_horoscope_v1_bounded_pp64',
}
def _summarize_pushkara_planets(flags):
"""Return the compact full-reading list without discarding raw flags."""
return [
f"{planet} ({data['sign']} {data['degree_in_sign']:.1f}°)"
for planet, data in flags.items()
if data.get('pushkara_navamsa') or data.get('pushkara_bhaga')
]
def _public_secondary_dasha_module(native):
"""Expose one native rashi dasha under the upstream full-reading key.
The period table and profile id stay ours. The outer status is what the
upstream synastry validator reads, and it is never a bare error object.
"""
if not isinstance(native, dict):
return {
'status': 'blocked',
'reason': 'native_rashi_dasha_missing',
'periods': [],
}
return {
'status': 'parameter_sensitive',
'profile_id': native.get('profile_id'),
'periods': list(native.get('periods') or []),
'native_result': native,
}
def _calc_sensitive_points(planets):
result = {}
try:
@@ -464,7 +510,7 @@ def _calc_sensitive_points(planets):
def _calc_dasha_sandhi(dasha_result, reference_date=None, orb_days=90):
ref = datetime.strptime(reference_date, "%Y-%m-%d") if reference_date else datetime.now()
ref = datetime.strptime(reference_date, "%Y-%m-%d") if reference_date else datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
sandhi = []
timeline = dasha_result.get('timeline', []) if isinstance(dasha_result, dict) else []
for md in timeline:
@@ -486,7 +532,11 @@ def _calc_dasha_sandhi(dasha_result, reference_date=None, orb_days=90):
for boundary_key in ['start', 'end']:
if boundary_key not in ad:
continue
bdt = datetime.strptime(ad[boundary_key], "%Y-%m-%d")
# 被出生裁掉的小运起点不是交接。用完整起点判断，第一大运自己的 start 仍是出生前的真实交接，不跳过。
boundary_date = ad[boundary_key]
if boundary_key == 'start' and ad.get('unclipped_start_datetime'):
boundary_date = datetime.fromisoformat(ad['unclipped_start_datetime']).strftime('%Y-%m-%d')
bdt = datetime.strptime(boundary_date, "%Y-%m-%d")
delta = (bdt - ref).days
if abs(delta) <= orb_days:
sandhi.append({
@@ -494,7 +544,7 @@ def _calc_dasha_sandhi(dasha_result, reference_date=None, orb_days=90):
'mahadasha_lord': md.get('lord'),
'lord': ad.get('lord'),
'boundary': boundary_key,
'date': ad.get(boundary_key),
'date': boundary_date,
'days_from_reference': delta,
'within_orb': True,
})
@@ -7250,9 +7300,9 @@ def render_pl9_markdown(packet: dict) -> str:
if shodhya_pinda:
out.extend([
'',
'#### Shodhya / Yoga Pinda（本地补充分值）',
'#### Shodhya / Yoga Pinda（PVR §12.7）',
'',
'本表透传本地 Bindu 权重计算的 Rashi/Graha/Yoga Pinda。不等同于 PL9 第 54 页的 Trikona / Ekadhipatya reduction 前后矩阵；该页逐阶段结构仍未闭环，所有行仅作 parameter_sensitive 补充数据。',
'本表与 Sodhita 表用同一份削减：先三方净化，再单宫主净化。同值且一占一空时按 Raman 15(c) 清空空宫。星座权重按 PVR 表 28（处女座 6），七曜权重按表 29。这是具名口径，不是经典定论。',
'',
'| 行星 | Rashi Pinda | Graha Pinda | Yoga Pinda | 本命星座 Bindu | 状态 |',
'|------|-------------|-------------|------------|----------------|------|',
@@ -12023,10 +12073,16 @@ def _load_strict_evidence_collector():
# ============================================================================
def _nakshatra_fields(sidereal_lon):
"""Nakshatra, pada and lord of a sidereal longitude — planets and the ascendant alike."""
nak_span = 360.0 / 27
from nakshatra_advanced import nakshatra_index_and_progress
lon = sidereal_lon % 360
pada = int((lon % nak_span) / (nak_span / 4)) + 1
name, lord, _ = NAKSHATRA_LIST[int(lon / nak_span) % 27]
idx, progress = nakshatra_index_and_progress(lon)
if progress == 0.0:
pada = 1
else:
nak_span = 360.0 / 27
pada = int((lon % nak_span) / (nak_span / 4)) + 1
idx = int(lon / nak_span) % 27
name, lord, _ = NAKSHATRA_LIST[idx]
return {"nakshatra": name, "nakshatra_pada": pada, "nakshatra_lord": lord}
@@ -12165,8 +12221,9 @@ def _build_pratyantar_timeline(md_lord, ad_lord, ad_start, ad_end, reference_dt=
def cmd_dasha(args):
nak_info = None; progress = 0.5
if args.moon_lon is not None:
ns = 360.0 / 27; idx = int(args.moon_lon / ns); progress = (args.moon_lon % ns) / ns
nak_info = NAKSHATRA_LIST[idx % 27]
from nakshatra_advanced import nakshatra_index_and_progress
idx, progress = nakshatra_index_and_progress(args.moon_lon)
nak_info = NAKSHATRA_LIST[idx]
elif args.nakshatra:
nl = args.nakshatra.lower().replace(" ", "").replace("-", "")
for n in NAKSHATRA_LIST:
@@ -12183,8 +12240,9 @@ def cmd_dasha(args):
moon_lon = moon.get("degree_raw")
if moon_lon is None:
return {"error": "无法计算Moon位置"}
ns = 360.0 / 27; idx = int(moon_lon / ns); progress = (moon_lon % ns) / ns
nak_info = NAKSHATRA_LIST[idx % 27]
from nakshatra_advanced import nakshatra_index_and_progress
idx, progress = nakshatra_index_and_progress(moon_lon)
nak_info = NAKSHATRA_LIST[idx]
nak_name, start_lord, start_years = nak_info
birthdate = args.birthdate or f"{args.year}-{args.month:02d}-{args.day:02d}"
@@ -14499,20 +14557,17 @@ def _tajika_annual_core(args, planet_lons, asc_si, age, mode='all'):
# ============================================================================
# 21. 合盘分析（v3.7新增）
# ============================================================================
def cmd_synastry(args):
def cmd_marriage_match(args):
try:
sys.path.insert(0, SCRIPT_DIR)
from synastry import calc_synastry
except ImportError as e:
return {"error": f"synastry模块导入失败: {e}"}
# 构建两人数据
p1 = {'moon_lon': args.moon1, 'gender': args.gender1 or 'M'}
p2 = {'moon_lon': args.moon2, 'gender': args.gender2 or 'F'}
if args.mars1 is not None: p1['mars_lon'] = args.mars1
if args.mars2 is not None: p2['mars_lon'] = args.mars2
if args.asc1 is not None: p1['asc_lon'] = args.asc1
if args.asc2 is not None: p2['asc_lon'] = args.asc2
return _build_response_envelope('synastry', calc_synastry(p1, p2), args=args, execution_status='executed')
from marriage_match_service.cli import run_marriage_match_cli
except ImportError as exc:
return {"error": f"marriage_match模块导入失败: {exc}"}
try:
result = run_marriage_match_cli(args)
except (OSError, ValueError, json.JSONDecodeError) as exc:
return {"error": str(exc)}
return _build_response_envelope('marriage-match', result, args=args, execution_status='executed')
# ============================================================================
@@ -15425,10 +15480,15 @@ def cmd_full_reading(args):
try:
moon_data = planets.get('Moon', {})
moon_lon = moon_data.get('degree_raw', moon_data.get('degree', 0))
nak_idx = int(moon_lon / (360 / 27)) % 27
from nakshatra_advanced import nakshatra_index_and_progress
nak_idx, nak_progress = nakshatra_index_and_progress(moon_lon)
if nak_progress != 0.0:
nak_idx = int(moon_lon / (360 / 27)) % 27
pada = int((moon_lon % (360/27)) / (360/108)) + 1
else:
pada = 1
nak_name = NAKSHATRA_LIST[nak_idx][0]
nak_lord = NAKSHATRA_LIST[nak_idx][1]
pada = int((moon_lon % (360/27)) / (360/108)) + 1
birthdate = f"{args.year}-{args.month:02d}-{args.day:02d}"
today_str = getattr(args, 'transit_date', None) or getattr(args, 'today', None) or datetime.now().strftime('%Y-%m-%d')
@@ -15551,7 +15611,9 @@ def cmd_full_reading(args):
report['errors'].append(f"yoga-context-injection: {yoga_ctx_e}")
report['modules']['vargottama'] = _calc_vargottama(planets, varga_result)
report['modules']['lagna_vargottamamsa'] = _calc_lagna_vargottamamsa(report, varga_result)
report['modules']['pushkara'] = _calc_pushkara_flags(planets)
report['modules']['pushkara_planets'] = _summarize_pushkara_planets(report['modules']['pushkara'])
report['modules']['sensitive_points'] = _calc_sensitive_points(planets)
except Exception as e:
report['errors'].append(f"varga-full: {e}")
@@ -15993,6 +16055,13 @@ def cmd_full_reading(args):
'reference_date': reference_date,
'reason': f'{module_name}_producer_failed:{e}',
}
# Upstream western modules import ``scripts.*``. Running this file as a
# script only puts ``scripts/`` on the path, and the copied files treat
# ``_script_root`` as optional. Keep the repository root available here
# instead of adding a local helper under ``scripts/``.
_repo_root = str(Path(__file__).resolve().parents[1])
if _repo_root not in sys.path:
sys.path.append(_repo_root)
try:
from unified_western_reading import build_unified_western_reading
@@ -16036,6 +16105,37 @@ def cmd_full_reading(args):
'status': 'blocked',
'reason': 'reference_date_missing',
}
try:
from punarphoo import detect_punarphoo
report['modules']['punarphoo'] = detect_punarphoo(planets)
except Exception as e:
report['modules']['punarphoo'] = {
'status': 'blocked',
'reason': f'punarphoo_producer_failed:{e}',
}
try:
from rashi_dashas_native import calculate_rashi_dashas
secondary_dashas = calculate_rashi_dashas(
_birth_datetime_from_args(args),
float(asc_deg),
planet_lons,
)
report['modules']['navamsha_dasha'] = _public_secondary_dasha_module(secondary_dashas.get('navamsha'))
except Exception as e:
report['modules']['navamsha_dasha'] = {
'status': 'blocked',
'reason': f'navamsha_dasha_producer_failed:{e}',
'periods': [],
}
# F2b: do not publish the current Niryaana Shoola table. It matches Shoola.
# The synastry validator marks this row not applicable, so the pair task stays open.
report['modules']['niryaana_shoola_dasha'] = {
'execution_status': 'blocked',
'reason': '口径更正中，见 TASK-niryaana-shoola-pvr-20261018',
}
_record_stage_timing(
stage_timings,
'dasha_and_core_varga_stack',
@@ -16352,6 +16452,20 @@ def cmd_full_reading(args):
# ── Step 10.5: Avasthas 行星状态 (v4.4.0) ──
try:
from avastha_calculator import AvasthaCalculator
from saham_daynight import SahamDayNightError, determine_daytime
try:
avastha_solar_day = determine_daytime(
_birth_datetime_from_args(args), lat=args.lat, lon=args.lon, tz=args.tz, rising='hindu')
except SahamDayNightError as exc:
# 极昼极夜没有日出：只让姿态一项标无法计算，其余四类照常（BUG-1390）。
avastha_activity_context = {'unavailable_reason': str(exc)}
else:
hours_since_sunrise = (avastha_solar_day['julian_day_ut'] - avastha_solar_day['sunrise_jd_ut']) * 24
avastha_activity_context = {
'moon_longitude': planet_lons['Moon'],
'asc_longitude': asc_deg,
'hours_since_sunrise': hours_since_sunrise,
}
avast_calc = AvasthaCalculator()
avast_result = {}
for pn, pd in planets.items():
@@ -16364,7 +16478,10 @@ def cmd_full_reading(args):
if isinstance(pd2, dict) and pd2.get('house') == house and p2 != pn]
avast_result[pn] = avast_calc.calculate_all_avasthas(
planet=pn, sign=sign, degree=deg_in_sign,
house=house, conjunctions=conjunctions)
house=house, conjunctions=conjunctions,
is_combust=bool(pd.get('combust', False)),
is_retrograde=bool(pd.get('retrograde', False)),
activity_context=avastha_activity_context)
report['modules']['avasthas'] = avast_result
except Exception as e:
report['errors'].append(f"avasthas: {e}")
@@ -16459,8 +16576,13 @@ def cmd_full_reading(args):
jaimini_data = report['modules'].get('jaimini', {})
chara_dasha_data = jaimini_data.get('chara_dasha', {}) if isinstance(jaimini_data, dict) else {}
moon_lon = planet_lons.get('Moon', 0)
moon_nakshatra_index = int(moon_lon / (360 / 27)) % 27
moon_pada = int((moon_lon % (360 / 27)) / (360 / 108)) + 1
from nakshatra_advanced import nakshatra_index_and_progress
moon_nakshatra_index, moon_nak_progress = nakshatra_index_and_progress(moon_lon)
if moon_nak_progress != 0.0:
moon_nakshatra_index = int(moon_lon / (360 / 27)) % 27
moon_pada = int((moon_lon % (360 / 27)) / (360 / 108)) + 1
else:
moon_pada = 1
tithi_number = int(((moon_lon - planet_lons.get('Sun', 0)) % 360) / 12) + 1
birth_dt = _birth_datetime_from_args(args)
ll_name = SIGN_LORDS[SIGNS[asc_idx]]
@@ -18217,16 +18339,12 @@ def main():
p.add_argument('--hour-from-sunrise', type=float, default=6.0,
help='从日出起算的小时数（默认6h约正午）')
# 21. synastry (v3.7新增)
p = sub.add_parser('synastry', help='合盘分析（Ashta Koota 36分制）')
p.add_argument('--moon1', type=float, required=True, help='Person1月亮黄经')
p.add_argument('--moon2', type=float, required=True, help='Person2月亮黄经')
p.add_argument('--mars1', type=float, default=None, help='Person1火星黄经')
p.add_argument('--mars2', type=float, default=None, help='Person2火星黄经')
p.add_argument('--asc1', type=float, default=None, help='Person1上升黄经')
p.add_argument('--asc2', type=float, default=None, help='Person2上升黄经')
p.add_argument('--gender1', default='M', help='Person1性别')
p.add_argument('--gender2', default='F', help='Person2性别')
# 21. marriage-match（上游 bde5d615；只收盘快照）
p = sub.add_parser('marriage-match', help='婚恋合盘（两份盘快照）')
p.add_argument('--person-a', required=True, help='Person A 盘快照 JSON 文件')
p.add_argument('--person-b', required=True, help='Person B 盘快照 JSON 文件')
p.add_argument('--profile', default='practitioner_consensus_v1',
choices=['practitioner_consensus_v1', 'raman_muhurtha_v1'])
# 22. full-reading (v3.7.1新增)
p = sub.add_parser('full-reading', help='全自动综合解盘（出生信息→全链路→完整报告）')
@@ -18371,7 +18489,7 @@ def main():
'varga-full': cmd_varga_full, 'aspects': cmd_aspects, 'jaimini': cmd_jaimini,
'nakshatra-adv': cmd_nakshatra_adv, 'nakshatra-dasha': cmd_nakshatra_dasha,
'nakshatra-full': cmd_nakshatra_full, 'argala': cmd_argala, 'tajika': cmd_tajika,
'synastry': cmd_synastry, 'solar-return': cmd_solar_return,
'marriage-match': cmd_marriage_match, 'solar-return': cmd_solar_return,
'narayana-dasha': cmd_narayana_dasha, 'muhurta': cmd_muhurta,
'full-reading': cmd_full_reading, 'pl9-export': cmd_pl9_export, 'prashna': cmd_prashna,
'double-transit-pac': cmd_double_transit_pac,
scripts/kp_system.py
+34 -33
View File
@@ -11,7 +11,10 @@ KP (Krishnamurti Paddhati) 占星系统模块
"""
import json
import math
from bisect import bisect_right
from datetime import datetime, timedelta
from fractions import Fraction
from pathlib import Path
from typing import Any, Dict, List, Tuple, Optional
@@ -44,6 +47,29 @@ VIMSHOTTARI_YEARS = dict(zip(KP_LORDS, VIMSHOTTARI_DURATION))
NAKSHATRA_SPAN = 360.0 / 27.0 # 13.333... degrees
def _angular_partition_starts():
"""Exact proportional starts, rounded once at the float API boundary."""
starts, owners = [], []
for star in range(27):
sub_start = Fraction(40 * star, 3)
for offset in range(9):
sub = (star + offset) % 9
sub_span = Fraction(40, 3) * VIMSHOTTARI_DURATION[sub] / 120
child_start = sub_start
for child_offset in range(9):
child = (sub + child_offset) % 9
starts.append(float(child_start))
owners.append((sub, child))
child_start += sub_span * VIMSHOTTARI_DURATION[child] / 120
sub_start += sub_span
return starts, owners
_KP_STARTS, _KP_OWNERS = _angular_partition_starts()
_STAR_STARTS = [float(Fraction(40 * i, 3)) for i in range(27)]
_PADA_STARTS = [float(Fraction(10 * i, 3)) for i in range(108)]
ROOT = Path(__file__).resolve().parents[1]
@@ -228,50 +254,25 @@ def get_kp_lords(degree: float) -> Dict:
Returns:
KP lords字典
"""
if isinstance(degree, bool) or not isinstance(degree, (int, float)) or not math.isfinite(degree):
raise ValueError('KP longitude must be a finite number')
deg = degree % 360
# 1. Sign lord
sign_index = int(deg // 30)
# 2. Nakshatra
nakshatra_index = int(deg // NAKSHATRA_SPAN) % 27
nakshatra_deg = deg % NAKSHATRA_SPAN
pada = int(nakshatra_deg // (NAKSHATRA_SPAN / 4)) + 1
# 3. Sublord & SubSubLord（KP核心算法）
# 将Vimshottari 120年周期按比例投影到度数上
deg_remainder = deg - 120 * int(deg / 120)
deg_cumulative = 0.0
for i in range(9):
deg_nl = NAKSHATRA_SPAN # 13.333... degrees per nakshatra
for j in range(i, i + 9):
j_mod = j % 9
deg_sl = deg_nl * VIMSHOTTARI_DURATION[j_mod] / 120.0
for k in range(j_mod, j_mod + 9):
k_mod = k % 9
deg_ss = deg_sl * VIMSHOTTARI_DURATION[k_mod] / 120.0
deg_cumulative += deg_ss
if deg_cumulative >= deg_remainder:
return {
'rasi_lord': SIGN_LORDS.get(SIGNS[sign_index], ''),
'sign': SIGNS[sign_index],
'nakshatra': NAKSHATRAS[nakshatra_index],
'nakshatra_lord': STAR_LORDS[nakshatra_index],
'pada': pada,
'sub_lord': KP_LORDS[j_mod],
'sub_sub_lord': KP_LORDS[k_mod],
}
# Fallback
# 2. Nakshatra. A shared boundary belongs to the subdivision that opens there.
nakshatra_index = bisect_right(_STAR_STARTS, deg) - 1
pada = (bisect_right(_PADA_STARTS, deg) - 1) % 4 + 1
sub, child = _KP_OWNERS[bisect_right(_KP_STARTS, deg) - 1]
return {
'rasi_lord': SIGN_LORDS.get(SIGNS[sign_index], ''),
'sign': SIGNS[sign_index],
'nakshatra': NAKSHATRAS[nakshatra_index],
'nakshatra_lord': STAR_LORDS[nakshatra_index],
'pada': pada,
'sub_lord': 'Unknown',
'sub_sub_lord': 'Unknown',
'sub_lord': KP_LORDS[sub],
'sub_sub_lord': KP_LORDS[child],
}
scripts/marriage_match.py
+467
View File
@@ -0,0 +1,467 @@
#!/usr/bin/env python3
"""User-facing romantic/marriage compatibility contract.
This module deliberately does not reuse the historical ``calc_kuja_dosha``
score. That function combines Mars with five pressure planets and contains a
male/female asymmetric comparison. It remains available for legacy reports,
but is not a valid default for a new relationship product.
"""
from __future__ import annotations
from copy import deepcopy
from typing import Any, Dict, Iterable, Mapping, Optional
from ashtakoot import (
FRIENDSHIP,
NADI,
SIGN_LORDS,
SIGNS,
calculate_ashtakoot,
get_nakshatra,
)
PRODUCT_VERSION = "marriage_match_product_v1"
DEFAULT_PROFILE = "practitioner_consensus_v1"
VALID_RELATIONSHIP_TYPES = {"romantic_or_marriage", "romantic", "marriage"}
KUJA_PLANET = "Mars"
PAPASAMYA_PLANETS = ("Mars", "Saturn", "Sun", "Rahu", "Ketu")
# These profiles are deliberately qualitative. The consensus did not support
# an invented severity percentage or a sex-specific tolerance threshold.
RULE_PROFILES = {
"practitioner_consensus_v1": {
"label": "Practitioner consensus v1",
"kuja_houses": (1, 2, 4, 7, 8, 12),
"reference_frames": ("lagna", "moon", "venus"),
"exact_house_sign_exceptions": {
2: {"Gemini", "Virgo"},
4: {"Aries", "Scorpio"},
7: {"Capricorn", "Cancer"},
8: {"Sagittarius", "Pisces"},
12: {"Taurus", "Libra"},
},
"leo_aquarius_exception": "variant_only",
"own_or_exalted": "mitigates",
"jupiter_contact": "mitigates",
"moon_contact": "mitigates",
"papasamya_planets": PAPASAMYA_PLANETS,
"gender_asymmetry": False,
"notes": [
"Kuja is Mars-only; Papasamya is a separately reported five-planet pressure layer.",
"Lagna, Moon and Venus are shown separately; they are not converted into an opaque weighted score.",
"Own/exalted Mars and Jupiter or Moon contact mitigate a signal in this default profile; they do not erase the raw placement.",
"Age, caste and sex-asymmetric comparison rules are not used for a user-facing conclusion.",
],
},
"raman_muhurtha_v1": {
"label": "B. V. Raman Muhurtha variant",
"kuja_houses": (2, 4, 7, 8, 12),
"reference_frames": ("lagna", "moon", "venus"),
"exact_house_sign_exceptions": {
2: {"Gemini", "Virgo"},
4: {"Aries", "Scorpio"},
7: {"Capricorn", "Cancer"},
8: {"Sagittarius", "Pisces"},
12: {"Taurus", "Libra"},
},
"leo_aquarius_exception": "cancels",
"own_or_exalted": "mitigates",
"jupiter_contact": "mitigates",
"moon_contact": "mitigates",
"papasamya_planets": PAPASAMYA_PLANETS,
"gender_asymmetry": False,
"notes": [
"Raman-style house set omits the first house. It is a selectable lineage, not the product default.",
"The traditional Leo/Aquarius exception is applied only in this variant profile.",
],
},
}
def available_profiles() -> Dict[str, Dict[str, Any]]:
"""Return a presentation-safe copy of available rule profiles."""
return deepcopy(RULE_PROFILES)
def _profile(profile_id: str) -> Dict[str, Any]:
if profile_id not in RULE_PROFILES:
raise ValueError("Unknown marriage-match rule profile: %s" % profile_id)
return RULE_PROFILES[profile_id]
def _lon(value: Any) -> Optional[float]:
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
return float(value) % 360.0
if isinstance(value, Mapping):
for key in ("longitude", "degree_raw", "degree", "lon"):
if isinstance(value.get(key), (int, float)):
return float(value[key]) % 360.0
return None
def _sign_index(value: Any) -> Optional[int]:
if isinstance(value, str) and value in SIGNS:
return SIGNS.index(value)
longitude = _lon(value)
return int(longitude // 30) if longitude is not None else None
def _planet_data(chart: Mapping[str, Any], planet: str) -> Mapping[str, Any]:
planets = chart.get("planets", {}) if isinstance(chart, Mapping) else {}
value = planets.get(planet, {}) if isinstance(planets, Mapping) else {}
return value if isinstance(value, Mapping) else {}
def _planet_lon(chart: Mapping[str, Any], planet: str) -> Optional[float]:
value = _planet_data(chart, planet)
longitude = _lon(value)
return longitude if longitude is not None else _lon(chart.get(planet + "_lon"))
def _reference_lon(chart: Mapping[str, Any], reference: str) -> Optional[float]:
if reference == "lagna":
for value in (chart.get("ascendant"), chart.get("lagna"), chart.get("asc_lon")):
longitude = _lon(value)
if longitude is not None:
return longitude
return None
return _planet_lon(chart, "Moon" if reference == "moon" else "Venus")
def _reference_sign(chart: Mapping[str, Any], reference: str) -> Optional[int]:
value = _reference_lon(chart, reference)
if value is not None:
return int(value // 30)
if reference == "lagna":
# Aries is index 0, which is a valid reference sign rather than a
# missing value. Do not use boolean fallback here.
ascendant_sign = _sign_index(chart.get("ascendant"))
return ascendant_sign if ascendant_sign is not None else _sign_index(chart.get("lagna"))
return _sign_index(_planet_data(chart, "Moon" if reference == "moon" else "Venus"))
def _planet_sign(chart: Mapping[str, Any], planet: str) -> Optional[int]:
lon = _planet_lon(chart, planet)
return int(lon // 30) if lon is not None else _sign_index(_planet_data(chart, planet))
def _house_from(reference_sign: Optional[int], planet_sign: Optional[int]) -> Optional[int]:
if reference_sign is None or planet_sign is None:
return None
return ((planet_sign - reference_sign) % 12) + 1
def _same_sign_contact(chart: Mapping[str, Any], planet_a: str, planet_b: str) -> bool:
a, b = _planet_sign(chart, planet_a), _planet_sign(chart, planet_b)
return a is not None and a == b
def _mars_dignity(chart: Mapping[str, Any]) -> str:
sign_idx = _planet_sign(chart, "Mars")
if sign_idx is None:
return "unknown"
sign = SIGNS[sign_idx]
if sign == "Capricorn":
return "exalted"
if sign in {"Aries", "Scorpio"}:
return "own"
if sign == "Cancer":
return "debilitated"
return "other"
def _frame_kuja(chart: Mapping[str, Any], reference: str, profile: Mapping[str, Any]) -> Dict[str, Any]:
mars_sign_idx = _planet_sign(chart, "Mars")
ref_sign_idx = _reference_sign(chart, reference)
house = _house_from(ref_sign_idx, mars_sign_idx)
frame = {
"reference": reference,
"reference_sign": SIGNS[ref_sign_idx] if ref_sign_idx is not None else None,
"mars_sign": SIGNS[mars_sign_idx] if mars_sign_idx is not None else None,
"mars_house": house,
"raw_signal": house in set(profile["kuja_houses"]) if house is not None else None,
"status": "not_available" if house is None else "not_triggered",
"mitigations": [],
}
if house is None or not frame["raw_signal"]:
return frame
mars_sign = frame["mars_sign"]
exact_exceptions = profile["exact_house_sign_exceptions"]
if mars_sign in exact_exceptions.get(house, set()):
frame["status"] = "cancelled_by_profile_exception"
frame["mitigations"].append("house_sign_exception")
return frame
if mars_sign in {"Leo", "Aquarius"}:
mode = profile["leo_aquarius_exception"]
if mode == "cancels":
frame["status"] = "cancelled_by_profile_exception"
frame["mitigations"].append("leo_aquarius_profile_exception")
return frame
if mode == "variant_only":
frame["mitigations"].append("leo_aquarius_exception_disputed")
frame["status"] = "active"
dignity = _mars_dignity(chart)
if dignity in {"own", "exalted"}:
frame["mitigations"].append("mars_%s" % dignity)
if _same_sign_contact(chart, "Mars", "Jupiter"):
frame["mitigations"].append("jupiter_conjunction")
if _same_sign_contact(chart, "Mars", "Moon"):
frame["mitigations"].append("moon_conjunction")
return frame
def analyze_kuja(chart: Mapping[str, Any], profile_id: str = DEFAULT_PROFILE) -> Dict[str, Any]:
"""Analyze Mars-only Kuja / Mangal Dosha across visible reference frames."""
profile = _profile(profile_id)
if _planet_sign(chart, "Mars") is None:
return {
"system": "Kuja / Mangal Dosha",
"scope": "Mars only",
"profile_id": profile_id,
"status": "not_available",
"reason": "Mars position is required.",
"frames": [],
}
frames = [_frame_kuja(chart, ref, profile) for ref in profile["reference_frames"]]
active = [frame for frame in frames if frame["status"] == "active"]
cancelled = [frame for frame in frames if frame["status"] == "cancelled_by_profile_exception"]
severity = {0: "none", 1: "mild", 2: "material", 3: "strong"}[len(active)]
return {
"system": "Kuja / Mangal Dosha",
"scope": "Mars only",
"profile_id": profile_id,
"status": "executed",
"active_reference_count": len(active),
"cancelled_reference_count": len(cancelled),
"qualitative_level": severity,
"frames": frames,
"boundary": "Reference count is a presentation label, not a probability or a compatibility score.",
}
def analyze_papasamya(chart: Mapping[str, Any], profile_id: str = DEFAULT_PROFILE) -> Dict[str, Any]:
"""Report five-planet pressure separately from Mars-only Kuja."""
profile = _profile(profile_id)
rows = []
for planet in profile["papasamya_planets"]:
planet_sign = _planet_sign(chart, planet)
if planet_sign is None:
rows.append({"planet": planet, "status": "not_available", "frames": []})
continue
frames = []
for reference in profile["reference_frames"]:
house = _house_from(_reference_sign(chart, reference), planet_sign)
frames.append({
"reference": reference,
"house": house,
"in_pressure_house": house in set(profile["kuja_houses"]) if house is not None else None,
})
rows.append({"planet": planet, "status": "executed", "frames": frames})
active = [
{"planet": row["planet"], "reference": frame["reference"], "house": frame["house"]}
for row in rows for frame in row.get("frames", []) if frame.get("in_pressure_house")
]
return {
"system": "Papasamya pressure",
"scope": "Mars, Saturn, Sun, Rahu and Ketu; separate from Kuja.",
"profile_id": profile_id,
"status": "executed" if any(row["status"] == "executed" for row in rows) else "not_available",
"planet_rows": rows,
"active_placements": active,
"boundary": "This is a descriptive pressure layer. It is not renamed as Mangal Dosha and is not reduced to a sex-weighted score.",
}
def compare_kuja(person_a: Mapping[str, Any], person_b: Mapping[str, Any], profile_id: str = DEFAULT_PROFILE) -> Dict[str, Any]:
"""Compare two Kuja results without sex-specific tolerance or scoring."""
a, b = analyze_kuja(person_a, profile_id), analyze_kuja(person_b, profile_id)
if a["status"] != "executed" or b["status"] != "executed":
outcome = "incomplete"
else:
difference = abs(a["active_reference_count"] - b["active_reference_count"])
outcome = "not_triggered" if not a["active_reference_count"] and not b["active_reference_count"] else (
"broadly_balanced" if difference <= 1 else "asymmetric"
)
return {
"comparison_method": "neutral_reference_count",
"gender_asymmetry_used": False,
"outcome": outcome,
"person_a": a,
"person_b": b,
"boundary": "Both-active is a traditional balancing consideration, not an automatic approval or cancellation.",
}
def _moon_lon(chart: Mapping[str, Any]) -> Optional[float]:
longitude = _planet_lon(chart, "Moon")
return longitude if longitude is not None else _lon(chart.get("moon_lon"))
def _are_natural_friends(lord_a: str, lord_b: str) -> bool:
if lord_a == lord_b:
return True
return lord_b in FRIENDSHIP.get(lord_a, {}).get("friend", []) and lord_a in FRIENDSHIP.get(lord_b, {}).get("friend", [])
def _ashtakoota_with_exceptions(
person_a: Mapping[str, Any],
person_b: Mapping[str, Any],
) -> Dict[str, Any]:
moon_a, moon_b = _moon_lon(person_a), _moon_lon(person_b)
if moon_a is None or moon_b is None:
return {"status": "not_available", "reason": "Both Moon longitudes are required for Ashtakoota."}
forward = calculate_ashtakoot(moon_a, moon_b)
reverse = calculate_ashtakoot(moon_b, moon_a)
# The historical Varna point encodes caste hierarchy. It cannot truthfully
# be relabelled as a modern value judgement, so the international product
# excludes it rather than deriving a replacement from lunar signs.
score_names = ("Vashya", "Tara", "Yoni", "GrahaMaitri", "Gana", "Bhakoot", "Nadi")
symmetric_scores = {
name: round(
(float(forward["scores"][name]) + float(reverse["scores"][name])) / 2.0,
2,
)
for name in score_names
}
international_core = {
"method": "International Ashtakoota core v1 (both input orders averaged; classical Varna excluded)",
"scores": symmetric_scores,
"total_score": round(sum(symmetric_scores.values()), 2),
"max_score": 35.0,
"match_percentage": round(sum(symmetric_scores.values()) / 35.0 * 100, 1),
"person_a_details": {
"moon_sign": SIGNS[int(moon_a // 30)],
"nakshatra": get_nakshatra(moon_a)["name"],
"nakshatra_pada": get_nakshatra(moon_a)["pada"],
},
"person_b_details": {
"moon_sign": SIGNS[int(moon_b // 30)],
"nakshatra": get_nakshatra(moon_b)["name"],
"nakshatra_pada": get_nakshatra(moon_b)["pada"],
},
"directionality": {
"status": "symmetric",
"rule": "all included Kutas are averaged across person_a -> person_b and person_b -> person_a",
},
"boundary": "This is a mutual-reference score, not a caste hierarchy or an automatic relationship verdict.",
}
sign_a, sign_b = SIGNS[int(moon_a // 30)], SIGNS[int(moon_b // 30)]
nak_a, nak_b = get_nakshatra(moon_a), get_nakshatra(moon_b)
bhakoot_exceptions = []
if symmetric_scores["Bhakoot"] == 0:
lord_a, lord_b = SIGN_LORDS[sign_a], SIGN_LORDS[sign_b]
if lord_a == lord_b:
bhakoot_exceptions.append("same_sign_lord")
elif _are_natural_friends(lord_a, lord_b):
bhakoot_exceptions.append("mutual_natural_friend_sign_lords")
nadi_exceptions = []
if symmetric_scores["Nadi"] == 0:
if nak_a["index"] == nak_b["index"] and nak_a["pada"] != nak_b["pada"]:
nadi_exceptions.append("same_nakshatra_different_pada")
if sign_a == sign_b and nak_a["index"] != nak_b["index"]:
nadi_exceptions.append("same_rashi_different_nakshatra")
return {
"status": "executed",
"international_core": international_core,
"shared_life_dialogue": {
"status": "not_scored",
"storage": "no_answers_collected",
"prompts": [
"What does a committed relationship mean to each of you?",
"How do you want to make decisions about money, home, and care responsibilities?",
"What boundaries, privacy expectations, and conflict-repair habits matter most?",
"How do you each imagine work, family, location, and personal autonomy over the next few years?",
],
"boundary": "These are voluntary discussion prompts, not a compatibility test or a score.",
},
"classical_varna": {
"status": "excluded_from_default",
"reason": "Historical caste hierarchy is not used in the international product or its default score.",
},
"person_a": {"moon_sign": sign_a, "nakshatra": nak_a},
"person_b": {"moon_sign": sign_b, "nakshatra": nak_b},
"exception_review": {
"bhakoot": {
"raw_score": symmetric_scores["Bhakoot"],
"candidate_cancellations": bhakoot_exceptions,
"status": "candidate_mitigation" if bhakoot_exceptions else "none",
},
"nadi": {
"raw_score": symmetric_scores["Nadi"],
"candidate_cancellations": nadi_exceptions,
"status": "candidate_mitigation" if nadi_exceptions else "none",
},
"yoni": {
"raw_score": symmetric_scores["Yoni"],
"status": "legacy_three_band_table",
"boundary": "The legacy 4/2/0 table remains visible. Intermediate 3/1 grades require a separately versioned source table before becoming a default score.",
},
},
}
def _structure_status(chart: Mapping[str, Any]) -> Dict[str, Any]:
keys = {"d1_relationship", "d9", "upapada", "darakaraka", "relationship_dasha"}
supplied = [key for key in keys if chart.get(key) is not None]
return {
"status": "partial" if supplied else "not_available",
"supplied_layers": sorted(supplied),
"required_for_complete_reading": ["D1 relationship structure", "D9", "UL", "DK", "relationship timing / Dasha"],
"boundary": "36-point matching and Dosha layers do not replace D1/D9/UL/DK or timing review.",
}
def build_marriage_match_dossier(
person_a: Mapping[str, Any],
person_b: Mapping[str, Any],
*,
relationship_type: str = "romantic_or_marriage",
profile_id: str = DEFAULT_PROFILE,
) -> Dict[str, Any]:
"""Build the stable, user-facing love/marriage compatibility payload.
Inputs are chart snapshots only. The caller is responsible for calculating
D1/D9/timing snapshots; this function never stores birth details.
"""
if relationship_type not in VALID_RELATIONSHIP_TYPES:
raise ValueError("This product supports romantic/marriage matching only.")
profile = _profile(profile_id)
kuja = compare_kuja(person_a, person_b, profile_id)
ashtakoota = _ashtakoota_with_exceptions(person_a, person_b)
return {
"product": PRODUCT_VERSION,
"relationship_type": relationship_type,
"rule_profile": {
"id": profile_id,
"label": profile["label"],
"notes": profile["notes"],
"gender_asymmetry_used": False,
},
"ashtakoota_36": ashtakoota,
"kuja_mangal_dosha": kuja,
"papasamya": {
"person_a": analyze_papasamya(person_a, profile_id),
"person_b": analyze_papasamya(person_b, profile_id),
},
"relationship_structure": {
"person_a": _structure_status(person_a),
"person_b": _structure_status(person_b),
},
"presentation": {
"headline": "Relationship compatibility is presented as parallel traditional layers, not one approval score.",
"decision_boundary": "Do not let the 36-point score, Kuja, or Papasamya alone determine a relationship conclusion.",
"not_in_scope": ["business partnerships", "family matching", "age-based cancellation", "sex-asymmetric tolerance"],
},
"audit": {
"kuja": "executed: Mars-only, Lagna/Moon/Venus frames where data is supplied",
"papasamya": "executed separately: five-planet pressure placement review",
"ashtakoota": ashtakoota["status"],
"d1_d9_ul_dk_timing": "input-dependent; no missing layer is fabricated",
},
}
scripts/marriage_match_service/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Local glue for the upstream marriage-match engine. The engine file stays at scripts/marriage_match.py."""
scripts/marriage_match_service/api.py
+106
View File
@@ -0,0 +1,106 @@
"""HTTP-facing marriage match and full synastry dossier.
The dossier endpoint caches on the two birth payloads plus ``GITHUB_SHA``,
the same scratch helper the report reference cache uses. A missing or
``unknown`` commit id skips the cache.
"""
from __future__ import annotations
import os
from typing import Any
from marriage_match import build_marriage_match_dossier
from marriage_match_service.full_dossier import RULES_VERSION, build_synastry_full_dossier
try:
from scripts import consult_full_data_cache as _cache
except ModuleNotFoundError: # pragma: no cover - direct scripts/ execution
import consult_full_data_cache as _cache
CACHE_NAME = "synastry_full_dossier_cache"
_BIRTH_KEYS = ("year", "month", "day", "hour", "minute", "lat", "lon", "tz")
class MarriageMatchInputError(ValueError):
"""The caller sent a shape this product does not accept."""
def compute_marriage_match(payload: dict[str, Any]) -> dict[str, Any]:
"""Snapshot-only product. Birth details are not accepted here."""
if not isinstance(payload, dict):
raise MarriageMatchInputError("request body must be an object")
person_a = payload.get("person_a")
person_b = payload.get("person_b")
if not isinstance(person_a, dict) or not isinstance(person_b, dict):
raise MarriageMatchInputError("person_a and person_b must be chart snapshots")
profile_id = payload.get("profile_id") or "practitioner_consensus_v1"
if not isinstance(profile_id, str) or not profile_id.strip():
raise MarriageMatchInputError("profile_id must be a string")
try:
return build_marriage_match_dossier(
person_a,
person_b,
profile_id=profile_id.strip(),
)
except ValueError as exc:
raise MarriageMatchInputError(str(exc)) from exc
def _birth(value: Any, label: str) -> dict[str, Any]:
if not isinstance(value, dict):
raise MarriageMatchInputError(f"{label} must be an object")
missing = [key for key in _BIRTH_KEYS if value.get(key) is None]
if missing:
raise MarriageMatchInputError(f"{label} is missing {', '.join(missing)}")
return value
def _birth_identity(body: dict[str, Any]) -> dict[str, Any]:
node = body.get("node_mode", body.get("nodeMode", "mean"))
return {
"year": body.get("year"),
"month": body.get("month"),
"day": body.get("day"),
"hour": body.get("hour"),
"minute": body.get("minute"),
"second": body.get("second", 0),
"lat": body.get("lat"),
"lon": body.get("lon"),
"tz": body.get("tz"),
"ayanamsa": body.get("ayanamsa", "lahiri"),
"node_mode": node or "mean",
}
def _commit_sha() -> str:
return os.environ.get("GITHUB_SHA") or ""
def compute_synastry_full_dossier(handler: Any, body: dict[str, Any]) -> dict[str, Any]:
if not isinstance(body, dict):
raise MarriageMatchInputError("request body must be an object")
person_a = _birth(body.get("person_a"), "person_a")
person_b = _birth(body.get("person_b"), "person_b")
sha = _commit_sha()
def build(_identity: dict[str, Any]) -> dict[str, Any]:
return build_synastry_full_dossier(handler, person_a, person_b)
if sha in ("", "unknown"):
return build({})
cached = _cache.get_or_build(
{
"person_a": _birth_identity(person_a),
"person_b": _birth_identity(person_b),
"github_sha": sha,
"rules_version": RULES_VERSION,
},
build,
cache_name=CACHE_NAME,
)
packet = cached["packet"]
if not isinstance(packet, dict):
raise MarriageMatchInputError("cached dossier was not an object")
return packet
scripts/marriage_match_service/cli.py
+20
View File
@@ -0,0 +1,20 @@
"""CLI for snapshot marriage matching. Birth details are not an input."""
from __future__ import annotations
import json
from pathlib import Path
from marriage_match import build_marriage_match_dossier
def run_marriage_match_cli(args) -> dict:
person_a = json.loads(Path(args.person_a).read_text(encoding="utf-8"))
person_b = json.loads(Path(args.person_b).read_text(encoding="utf-8"))
if not isinstance(person_a, dict) or not isinstance(person_b, dict):
raise ValueError("person snapshots must be JSON objects")
return build_marriage_match_dossier(
person_a,
person_b,
profile_id=getattr(args, "profile", None) or "practitioner_consensus_v1",
)
scripts/marriage_match_service/full_dossier.py
+70
View File
@@ -0,0 +1,70 @@
"""Glue around the upstream full-synastry skill.
Two birth payloads become two native full-readings, then the untouched
``build_full_synastry_dossier``. Raman is applied to the snapshots the skill
already extracted. This module does not edit those upstream files.
"""
from __future__ import annotations
import importlib.util
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Any, Callable
from marriage_match import build_marriage_match_dossier
ROOT = Path(__file__).resolve().parents[2]
DOSSIER_SCRIPT = (
ROOT
/ "skills"
/ "jyotish-full-synastry-reading"
/ "scripts"
/ "build_full_synastry_dossier.py"
)
RULES_VERSION = "bde5d615:marriage_match_product_v1"
RAMAN_PROFILE = "raman_muhurtha_v1"
ReadingFn = Callable[[dict[str, Any]], dict[str, Any]]
def _dossier_builder():
spec = importlib.util.spec_from_file_location(
"build_full_synastry_dossier",
DOSSIER_SCRIPT,
)
if spec is None or spec.loader is None:
raise RuntimeError("full synastry dossier script is missing")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module.build_full_synastry_dossier
def build_synastry_full_dossier(
handler: Any,
person_a: dict[str, Any],
person_b: dict[str, Any],
*,
read_chart: ReadingFn | None = None,
) -> dict[str, Any]:
"""Run two full-readings in parallel, then both marriage-match profiles."""
read = read_chart or handler._compute_full_reading_for_thematic
with ThreadPoolExecutor(max_workers=2) as pool:
future_a = pool.submit(read, person_a)
future_b = pool.submit(read, person_b)
reading_a = future_a.result()
reading_b = future_b.result()
dossier = _dossier_builder()(reading_a, reading_b)
snapshots = dossier.get("chart_snapshots") if isinstance(dossier, dict) else {}
snapshots = snapshots if isinstance(snapshots, dict) else {}
raman = build_marriage_match_dossier(
snapshots.get("person_a") or {},
snapshots.get("person_b") or {},
profile_id=RAMAN_PROFILE,
)
return {
"rules_version": RULES_VERSION,
"dossier": dossier,
"raman_marriage_match": raman,
}
scripts/nakshatra_advanced.py
+49 -12
View File
@@ -16,6 +16,14 @@ Nakshatra深度分析系统
from typing import Dict, List, Tuple, Optional
import math
from datetime import datetime, timedelta
from bisect import bisect_right
from fractions import Fraction
# Exact star openings. float(Fraction) is the grid cmd_dasha must share with
# lon_to_nakshatra; 360/27 and modulo disagree at several of these points.
_NAK_SPAN = Fraction(40, 3)
_NAK_STARTS = tuple(float(_NAK_SPAN * index) for index in range(27))
_NAK_SPAN_FLOAT = 360.0 / 27.0
NAK_LIST = [
("Ashwini","Ketu",7),("Bharani","Venus",20),("Krittika","Sun",6),
@@ -53,15 +61,42 @@ NAK_ELEMENT = {
}
def nakshatra_index_and_progress(longitude: float) -> Tuple[int, float]:
"""星宿序号和已用比例，由同一把尺子量出。
经度恰好等于某星宿起点时，这一点属于刚开始的那一宿，已用比例为 0。
其余经度保持原来的取模结果，避免改动非边界点。
"""
lon = float(longitude) % 360.0
idx = bisect_right(_NAK_STARTS, lon) - 1
if idx < 0:
idx = 0
elif idx > 26:
idx = 26
if lon == _NAK_STARTS[idx]:
return idx, 0.0
legacy_idx = int(lon / _NAK_SPAN_FLOAT) % 27
legacy_progress = (lon % _NAK_SPAN_FLOAT) / _NAK_SPAN_FLOAT
return legacy_idx, legacy_progress
def find_nakshatra(longitude: float) -> Dict:
"""从恒星黄道经度精确计算Nakshatra"""
nak_span = 360.0 / 27 # 每个星宿13.333...°
nak_idx = int(longitude / nak_span) % 27
nak_start = nak_idx * nak_span
deg_in_nak = longitude - nak_start
pada_span = nak_span / 4 # 每个Pada 3.333...°
pada = int(deg_in_nak / pada_span) + 1
deg_in_pada = deg_in_nak - (pada - 1) * pada_span
idx, _progress = nakshatra_index_and_progress(longitude)
lon = float(longitude) % 360.0
if lon == _NAK_STARTS[idx]:
nak_idx = idx
deg_in_nak = 0.0
pada = 1
deg_in_pada = 0.0
else:
nak_span = 360.0 / 27 # 每个星宿13.333...°
nak_idx = int(longitude / nak_span) % 27
nak_start = nak_idx * nak_span
deg_in_nak = longitude - nak_start
pada_span = nak_span / 4 # 每个Pada 3.333...°
pada = int(deg_in_nak / pada_span) + 1
deg_in_pada = deg_in_nak - (pada - 1) * pada_span
return {
'nakshatra': NAK_NAMES[nak_idx],
@@ -126,11 +161,13 @@ def calc_vimshottari_start(moon_longitude: float) -> Dict:
nak_lord = NAK_LORDS[nak_idx]
total_years = NAK_YEARS[nak_idx]
# 已用比例
nak_span = 360.0 / 27
nak_start = nak_idx * nak_span
deg_in_nak = moon_longitude - nak_start
used_ratio = deg_in_nak / nak_span
# 已用比例与星宿序号来自同一把尺子。起点处必须是 0，不能是「几乎走完」。
_, used_ratio = nakshatra_index_and_progress(moon_longitude)
if float(moon_longitude) % 360.0 != _NAK_STARTS[nak_idx]:
nak_span = 360.0 / 27
nak_start = nak_idx * nak_span
deg_in_nak = moon_longitude - nak_start
used_ratio = deg_in_nak / nak_span
remaining_ratio = 1 - used_ratio
# 第一个Mahadasha的剩余年数
scripts/nakshatra_dasha.py
+43 -96
View File
@@ -304,13 +304,6 @@ def _calc_ashtottari_antardasha(md_lord: str, md_years: float) -> List[Dict]:
# Vimshottari Nakshatra-level 拆解
# ============================================================================
# Vimshottari Dasha 顺序（9星循环）
try:
from dasha_common import VIMSHOTTARI_ORDER, VIMSHOTTARI_YEARS
except ModuleNotFoundError: # pragma: no cover - package import
from scripts.dasha_common import VIMSHOTTARI_ORDER, VIMSHOTTARI_YEARS
def calc_nakshatra_dasha_breakdown(
moon_longitude: float,
planet_lons: Dict[str, float],
@@ -338,87 +331,58 @@ def calc_nakshatra_dasha_breakdown(
moon_nak_data = find_nakshatra(moon_longitude)
moon_nak_idx = moon_nak_data['nakshatra_idx']
# 计算 Vimshottari 起始
nak = find_nakshatra(moon_longitude)
nak_idx = nak['nakshatra_idx']
nak_span = 360.0 / 27
nak_start = nak_idx * nak_span
deg_in_nak = moon_longitude - nak_start
used_ratio = deg_in_nak / nak_span
# 起始守护星和剩余年数
start_lord = NAK_LORDS[nak_idx]
start_total = VIMSHOTTARI_YEARS.get(start_lord, 7)
first_remaining = start_total * (1 - used_ratio)
# 构建大运序列
start_idx = VIMSHOTTARI_ORDER.index(start_lord)
sequence = []
cumulative = 0.0
# 首个（剩余）大运
sequence.append({
'lord': start_lord,
'years': round(first_remaining, 4),
'cumulative_from': 0.0,
'is_first': True,
})
cumulative += first_remaining
idx = (start_idx + 1) % 9
for _ in range(15): # 足够覆盖到120岁
lord = VIMSHOTTARI_ORDER[idx]
years = VIMSHOTTARI_YEARS[lord]
sequence.append({
'lord': lord,
'years': round(years, 4),
'cumulative_from': round(cumulative, 4),
'is_first': False,
})
cumulative += years
idx = (idx + 1) % 9
if cumulative > 120:
break
# 定位当前大运
current_md = None
current_md_idx = -1
for i, item in enumerate(sequence):
if item['cumulative_from'] <= age < (item['cumulative_from'] + item['years']):
current_md = item
current_md_idx = i
break
# 小运沿用 cmd_dasha：从完整大运真实起点排，再裁到出生之后。
# 把第一大运的剩余年数当成完整大运会把早期小运整段重切（BUG-1380）。
from types import SimpleNamespace
try:
from jyotish_engine import cmd_dasha
except ImportError:
from scripts.jyotish_engine import cmd_dasha
if not math.isfinite(age) or age < 0:
raise ValueError('age must be finite and nonnegative')
birth = datetime.fromisoformat(birth_date_str)
reference = birth + timedelta(days=age * 365.25)
packet = cmd_dasha(SimpleNamespace(
moon_lon=moon_longitude, nakshatra=None, pada=None,
birthdate=birth.date().isoformat(), year=birth.year, month=birth.month,
day=birth.day, hour=birth.hour, minute=birth.minute, second=birth.second,
lat=None, lon=None, tz=0, today=reference.date().isoformat(),
dasha_year_days=365.25))
current_md = next((row for row in packet['timeline']
if datetime.fromisoformat(row['start_datetime']) <= reference
< datetime.fromisoformat(row['end_datetime'])), None)
if current_md is None:
return {'error': f'无法定位年龄 {age} 在 Vimshottari 序列中'}
md_lord = current_md['lord']
md_years = current_md['years']
md_start_age = current_md['cumulative_from']
years_in_md = age - md_start_age
md_start = datetime.fromisoformat(current_md['start_datetime'])
md_end = datetime.fromisoformat(current_md['end_datetime'])
md_years = (md_end - md_start).total_seconds() / (86400 * 365.25)
md_start_age = (md_start - birth).total_seconds() / (86400 * 365.25)
years_in_md = (reference - md_start).total_seconds() / (86400 * 365.25)
# 大运守护星的 Nakshatra
md_lord_nak = None
if md_lord in planet_lons:
md_lord_nak = find_nakshatra(planet_lons[md_lord])
# 计算当前 Antardasha
antardasas = _calc_vimshottari_antardasha_full(md_lord, md_years)
# 定位当前 Antardasha
ad_cumulative = 0.0
antardasas = []
current_ad = None
for ad in antardasas:
ad_end = ad_cumulative + ad['years']
if ad_cumulative <= years_in_md < ad_end:
for ad in current_md['antardasha_timeline']:
ad_start = datetime.fromisoformat(ad['start_datetime'])
ad_end = datetime.fromisoformat(ad['end_datetime'])
years = (ad_end - ad_start).total_seconds() / (86400 * 365.25)
antardasas.append({'lord': ad['lord'], 'years': years})
if ad_start <= reference < ad_end:
nominal_start = datetime.fromisoformat(ad['unclipped_start_datetime'])
span = (ad_end - nominal_start).total_seconds()
current_ad = {
'lord': ad['lord'],
'years': ad['years'],
'progress': (years_in_md - ad_cumulative) / ad['years'] if ad['years'] > 0 else 0,
'remaining': round(ad_end - years_in_md, 4),
'years': round(years, 4),
'progress': (reference - nominal_start).total_seconds() / span if span else 0,
'remaining': round((ad_end - reference).total_seconds() / (86400 * 365.25), 4),
'start_datetime': ad['start_datetime'],
'end_datetime': ad['end_datetime'],
}
break
ad_cumulative = ad_end
# 当前 Antardasha 守护星的 Nakshatra
ad_lord_nak = None
@@ -451,8 +415,8 @@ def calc_nakshatra_dasha_breakdown(
md_years_per_pada = md_years_per_nak / 4
for ad in antardasas:
if md_years_per_nak > 0:
ad_nak = find_nakshatra(planet_lons.get(ad['lord'], 0))
if md_years_per_nak > 0 and ad['lord'] in planet_lons:
ad_nak = find_nakshatra(planet_lons[ad['lord']])
pada_windows.append({
'period': f"AD {ad['lord']}",
'nakshatra': ad_nak['nakshatra'],
@@ -470,6 +434,8 @@ def calc_nakshatra_dasha_breakdown(
return {
'system': 'Vimshottari Nakshatra-level Breakdown',
'period_source': 'jyotish_engine.cmd_dasha native timeline',
'reference_datetime': reference.isoformat(timespec='seconds'),
'moon_nakshatra': moon_nak_data['nakshatra'],
'moon_nakshatra_idx': moon_nak_idx,
'moon_pada': moon_nak_data['pada'],
@@ -490,25 +456,6 @@ def calc_nakshatra_dasha_breakdown(
}
def _calc_vimshottari_antardasha_full(md_lord: str, md_years: float) -> List[Dict]:
"""计算完整的 Vimshottari Antardasha（9个周期）"""
md_start_idx = VIMSHOTTARI_ORDER.index(md_lord)
total_years = sum(VIMSHOTTARI_YEARS.values()) # 120
result = []
for i in range(9):
ad_lord = VIMSHOTTARI_ORDER[(md_start_idx + i) % 9]
ad_years_total = VIMSHOTTARI_YEARS[ad_lord]
ad_years = md_years * ad_years_total / total_years
result.append({
'lord': ad_lord,
'years': round(ad_years, 4),
'months': round(ad_years * 12, 1),
})
return result
# ============================================================================
# Nakshatra Transit Overlay（过境星宿叠加大运）
# ============================================================================
scripts/narayana_dasha.py
+11 -5
View File
@@ -280,15 +280,21 @@ def _rasi_aspects(source_sign_idx: int, target_sign_idx: int) -> bool:
def select_narayana_seed_sign(
lagna_sign_idx: int,
planet_lons: Dict[str, float],
comparison_sign_idx: int | None = None,
) -> Dict:
"""Select the stronger of Lagna and seventh using the first two rule levels.
"""Select the stronger of two signs 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.
The default pair is Lagna and the seventh. Argala passes the obstructing
sign explicitly and reuses the same two levels. A tie is deliberately
returned as blocked: later classical strength rules are not 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)
other_sign_idx = (lagna_sign_idx + 6) % 12 if comparison_sign_idx is None else int(comparison_sign_idx)
if not 0 <= other_sign_idx < 12 or other_sign_idx == lagna_sign_idx:
raise ValueError("comparison sign index must be a different sign in 0..11")
candidate_indices = (lagna_sign_idx, other_sign_idx)
candidates = {}
for sign_idx in candidate_indices:
lord = SIGN_LORDS[SIGNS[sign_idx]]
@@ -312,7 +318,7 @@ def select_narayana_seed_sign(
}
lagna_candidate = candidates[lagna_sign_idx]
seventh_sign_idx = (lagna_sign_idx + 6) % 12
seventh_sign_idx = other_sign_idx
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"])
scripts/pl9_reader_export.py
+3
View File
@@ -1442,6 +1442,9 @@ def render_pl9_parity_markdown(packet: dict) -> str:
continue
def _state(name: str) -> str:
state = _dict(states.get(name))
applicable = state.get('applicable_states') if name == 'Deeptadi' else None
if isinstance(applicable, list) and applicable:
return _cell('、'.join(str(item) for item in applicable))
return _cell(state.get('state'))
lines.append(
f"| {_cell(_token(planet))} | {_state('Bala')} | {_state('Jagrat')} | {_state('Deeptadi')} | "
scripts/research/reported_offset_sweep.py
+5 -5
View File
@@ -30,11 +30,11 @@ from scripts.research.sealed_holdout_rerun import (
OFFSETS = (-30, -20, -15, -10, -8, -5, -3, 0, 3, 5, 8, 10, 15, 20, 30)
RADII = (15, 30, 60)
MINUTE_STEP = 1
# 原值: docs/research/reported_offset_ephemeris_nutation_2026_10_08.{freeze.json,json}
# 新值: docs/research/reported_offset_scoring_perf_2026_10_13.{freeze.json,json}
# 原因: BUG-1323～1325 改了冻结身份文件，输出不变，ERR-110 另存新记录
FREEZE = ROOT / "docs/research/reported_offset_scoring_perf_2026_10_13.freeze.json"
REPORT = ROOT / "docs/research/reported_offset_scoring_perf_2026_10_13.json"
# 原值: docs/research/reported_offset_scoring_perf_2026_10_13.{freeze.json,json}
# 新值: docs/research/reported_offset_upstream_sync8_2026_10_18.{freeze.json,json}
# 原因: sync8（BUG-1380～1389）改了 ashtakavarga / dasha_analyzer / narayana_dasha，输出不变，ERR-110 另存新记录
FREEZE = ROOT / "docs/research/reported_offset_upstream_sync8_2026_10_18.freeze.json"
REPORT = ROOT / "docs/research/reported_offset_upstream_sync8_2026_10_18.json"
LEGACY_REPORT = ROOT / "docs/research/reported_offset_2026_09_20.json"
scripts/research/sealed_holdout_rerun.py
+5 -5
View File
@@ -27,11 +27,11 @@ from scripts.minute_rectification_feature_facts_v4 import build_feature_fact_row
from scripts.minute_rectification_holdout_validator import validate
DATASET = ROOT / "references/real_case_calibration/minute_rectification_holdout_v3.json"
# 原值: docs/research/sealed_holdout_rerun_ephemeris_nutation_2026_10_08.{freeze.json,json}
# 新值: docs/research/sealed_holdout_rerun_scoring_perf_2026_10_13.{freeze.json,json}
# 原因: BUG-1323～1325 改了冻结身份文件，输出不变，ERR-110 另存新记录
FREEZE = ROOT / "docs/research/sealed_holdout_rerun_scoring_perf_2026_10_13.freeze.json"
REPORT = ROOT / "docs/research/sealed_holdout_rerun_scoring_perf_2026_10_13.json"
# 原值: docs/research/sealed_holdout_rerun_scoring_perf_2026_10_13.{freeze.json,json}
# 新值: docs/research/sealed_holdout_rerun_upstream_sync8_2026_10_18.{freeze.json,json}
# 原因: sync8（BUG-1380～1389）改了 ashtakavarga / dasha_analyzer / narayana_dasha，输出不变，ERR-110 另存新记录
FREEZE = ROOT / "docs/research/sealed_holdout_rerun_upstream_sync8_2026_10_18.freeze.json"
REPORT = ROOT / "docs/research/sealed_holdout_rerun_upstream_sync8_2026_10_18.json"
LEGACY_REPORT = ROOT / "docs/research/sealed_holdout_rerun_2026_09_20.json"
ARCHIVE = ROOT / "docs/research/history/rectification_pre_cross_midnight_2026_09_20"
PRODUCTION_FILES = [
scripts/saham_daynight.py
+62 -13
View File
@@ -12,23 +12,71 @@ class SahamDayNightError(ValueError):
pass
def determine_daytime(moment: datetime, *, lat: float, lon: float, tz: float) -> dict[str, Any]:
def determine_daytime(
moment: datetime,
*,
lat: float,
lon: float,
tz: float,
rising: str = "apparent",
) -> dict[str, Any]:
"""Day/night window.
``rising="apparent"`` is the historical disk-edge window. Rectification and
Panchanga rely on that default and must not pick up Hindu rising.
``rising="hindu"`` is disk-center, no refraction, with the real next
sunrise. Saham day/night, Gulika, and Shayanadi pass it explicitly.
"""
if rising not in ("apparent", "hindu"):
raise SahamDayNightError("unsupported sunrise convention")
if not -90 <= float(lat) <= 90 or not -180 <= float(lon) <= 180:
raise SahamDayNightError("invalid WGS84 latitude/longitude")
local = moment.replace(tzinfo=None)
jd = swe.julday(local.year, local.month, local.day, local.hour + local.minute / 60 + local.second / 3600 - float(tz))
geopos = (float(lon), float(lat), 0.0)
rise_status, rise = swe.rise_trans(jd - 1.0, swe.SUN, swe.CALC_RISE, geopos)
set_status, sunset = swe.rise_trans(jd - 1.0, swe.SUN, swe.CALC_SET, geopos)
if rise_status != 0 or set_status != 0:
raise SahamDayNightError("sunrise_or_sunset_unavailable_for_location_date")
sunrise_jd, sunset_jd = rise[0], sunset[0]
# Normalize the next daily events around the queried instant.
while sunrise_jd > jd:
sunrise_jd -= 1.0
while sunset_jd > jd:
sunset_jd -= 1.0
is_day = sunrise_jd <= jd < sunset_jd if sunrise_jd < sunset_jd else not (sunset_jd <= jd < sunrise_jd)
if rising == "apparent":
rise_status, rise = swe.rise_trans(jd - 1.0, swe.SUN, swe.CALC_RISE, geopos)
set_status, sunset = swe.rise_trans(jd - 1.0, swe.SUN, swe.CALC_SET, geopos)
if rise_status != 0 or set_status != 0:
raise SahamDayNightError("sunrise_or_sunset_unavailable_for_location_date")
sunrise_jd, sunset_jd = rise[0], sunset[0]
# Normalize the next daily events around the queried instant.
while sunrise_jd > jd:
sunrise_jd -= 1.0
while sunset_jd > jd:
sunset_jd -= 1.0
is_day = sunrise_jd <= jd < sunset_jd if sunrise_jd < sunset_jd else not (sunset_jd <= jd < sunrise_jd)
return {
"scope": "saham_daynight_swiss",
"status": "computed",
"is_daytime": is_day,
"julian_day_ut": jd,
"sunrise_jd_ut": sunrise_jd,
"sunset_jd_ut": sunset_jd,
"method": "swisseph.rise_trans",
"boundary": "Formula-specific +30 degree exceptions must be applied by the Saham rule layer, not inferred from house placement.",
}
flags = swe.FLG_SWIEPH | swe.FLG_TRUEPOS
def event(after, kind):
status, times = swe.rise_trans(after, swe.SUN, kind | swe.BIT_HINDU_RISING, geopos, flags=flags)
if status != 0:
raise SahamDayNightError("sunrise_or_sunset_unavailable_for_location_date")
return times[0]
# Consecutive rise/set events. Adding one Julian day to a sunrise is not
# the next sunrise, and it moves the night-segment endpoint.
midnight_jd = swe.julday(local.year, local.month, local.day, -float(tz))
sunrise_jd = event(midnight_jd, swe.CALC_RISE)
if sunrise_jd > jd:
midnight_jd -= 1.0
sunrise_jd = event(midnight_jd, swe.CALC_RISE)
next_sunrise_jd = event(midnight_jd + 1.0, swe.CALC_RISE)
sunset_jd = event(midnight_jd, swe.CALC_SET)
if not sunrise_jd <= jd < next_sunrise_jd or not sunrise_jd < sunset_jd < next_sunrise_jd:
raise SahamDayNightError("solar_day_window_unavailable")
is_day = sunrise_jd <= jd < sunset_jd
return {
"scope": "saham_daynight_swiss",
"status": "computed",
@@ -36,6 +84,7 @@ def determine_daytime(moment: datetime, *, lat: float, lon: float, tz: float) ->
"julian_day_ut": jd,
"sunrise_jd_ut": sunrise_jd,
"sunset_jd_ut": sunset_jd,
"method": "swisseph.rise_trans",
"next_sunrise_jd_ut": next_sunrise_jd,
"method": "swisseph.rise_trans_hindu_rising",
"boundary": "Formula-specific +30 degree exceptions must be applied by the Saham rule layer, not inferred from house placement.",
}
scripts/synastry.py
-262
View File
@@ -1,262 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Synastry（合盘）模块 — 基于 dashaflow (MIT License) 核心算法
16因子兼容性分析系统：
- Ashtakoot 8因子（36分制）
- 附加8 Kuta（Mahendra/Stree Deergha/Vedha/Kuja Dosha/Rajju等）
"""
from typing import Dict, List, Tuple, Any
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'
}
NAKSHATRAS = ['Ashwini', 'Bharani', 'Krittika', 'Rohini', 'Mrigashira', 'Ardra',
'Punarvasu', 'Pushya', 'Ashlesha', 'Magha', 'Purva Phalguni', 'Uttara Phalguni',
'Hasta', 'Chitra', 'Swati', 'Vishakha', 'Anuradha', 'Jyeshtha',
'Mula', 'Purva Ashadha', 'Uttara Ashadha', 'Shravana', 'Dhanishta', 'Shatabhisha',
'Purva Bhadrapada', 'Uttara Bhadrapada', 'Revati']
# =============================================================================
# 数据表 (来自 dashaflow MIT License)
# =============================================================================
YONI_ANIMALS = [
"Horse", "Elephant", "Sheep", "Serpent", "Serpent", "Dog", "Cat", "Sheep", "Cat",
"Rat", "Rat", "Cow", "Buffalo", "Tiger", "Buffalo", "Tiger", "Deer", "Deer",
"Dog", "Monkey", "Mongoose", "Monkey", "Lion", "Horse", "Lion", "Cow", "Elephant"
]
YONI_ENEMIES = {
"Horse": "Buffalo", "Buffalo": "Horse", "Elephant": "Lion", "Lion": "Elephant",
"Sheep": "Monkey", "Monkey": "Sheep", "Serpent": "Mongoose", "Mongoose": "Serpent",
"Dog": "Deer", "Deer": "Dog", "Cat": "Rat", "Rat": "Cat",
"Cow": "Tiger", "Tiger": "Cow"
}
GANA = [
"Deva", "Manushya", "Rakshasa", "Manushya", "Deva", "Manushya", "Deva", "Deva", "Rakshasa",
"Rakshasa", "Manushya", "Manushya", "Deva", "Rakshasa", "Deva", "Rakshasa", "Deva", "Rakshasa",
"Rakshasa", "Manushya", "Manushya", "Deva", "Rakshasa", "Rakshasa", "Manushya", "Manushya", "Deva"
]
NADI = [
"Adi", "Madhya", "Antya", "Antya", "Madhya", "Adi", "Adi", "Madhya", "Antya",
"Antya", "Madhya", "Adi", "Adi", "Madhya", "Antya", "Antya", "Madhya", "Adi",
"Adi", "Madhya", "Antya", "Antya", "Madhya", "Adi", "Adi", "Madhya", "Antya"
]
VARNA = {
"Cancer": 1, "Scorpio": 1, "Pisces": 1,
"Aries": 2, "Leo": 2, "Sagittarius": 2,
"Taurus": 3, "Virgo": 3, "Capricorn": 3,
"Gemini": 4, "Libra": 4, "Aquarius": 4
}
VASHYA_TYPE = {
"Aries": "Chatushpada", "Taurus": "Chatushpada",
"Leo": "Vanachara", "Sagittarius": "Chatushpada", "Capricorn": "Chatushpada",
"Gemini": "Manava", "Virgo": "Manava", "Libra": "Manava", "Aquarius": "Manava",
"Cancer": "Jalachara", "Pisces": "Jalachara",
"Scorpio": "Keet",
}
VEDHA_PAIRS = [(0,17),(1,16),(2,15),(3,14),(5,21),(6,20),(7,19),(8,18),(9,26),(10,25),(11,24),(12,23),(4,22)]
RAJJU_GROUPS = {
"Pada":{0,8,9,17,18,26}, "Kati":{1,7,10,16,25,19},
"Udara":{2,6,11,15,20,24}, "Kanta":{3,5,12,14,21,23}, "Sira":{4,13,22}
}
RAJJU_EFFECTS = {"Sira":"head—risk to husband","Kanta":"neck—risk to wife","Udara":"stomach—risk to children","Kati":"waist—poverty","Pada":"foot—wandering"}
NATURAL_FRIENDS = {
'Sun': ['Moon','Mars','Jupiter'], 'Moon': ['Sun','Mercury'],
'Mars': ['Sun','Moon','Jupiter'], 'Mercury': ['Sun','Venus'],
'Jupiter': ['Sun','Moon','Mars'], 'Venus': ['Mercury','Saturn'],
'Saturn': ['Mercury','Venus']
}
NATURAL_ENEMIES = {
'Sun': ['Saturn','Venus'], 'Moon': [], 'Mars': ['Mercury'],
'Mercury': ['Moon'], 'Jupiter': ['Mercury','Venus'],
'Venus': ['Sun','Moon'], 'Saturn': ['Sun','Moon','Mars']
}
EXALTATION = {'Sun':('Aries',10),'Moon':('Taurus',3),'Mars':('Capricorn',28),'Mercury':('Virgo',15),'Jupiter':('Cancer',5),'Venus':('Pisces',27),'Saturn':('Libra',20)}
DEBILITATION = {p:(SIGNS[(SIGNS.index(s)+6)%12],d) for p,(s,d) in EXALTATION.items()}
OWN_SIGNS = {'Sun':{'Leo'},'Moon':{'Cancer'},'Mars':{'Aries','Scorpio'},'Mercury':{'Gemini','Virgo'},'Jupiter':{'Sagittarius','Pisces'},'Venus':{'Taurus','Libra'},'Saturn':{'Capricorn','Aquarius'}}
_DOSHA_HOUSES = {2,4,7,8,12}
_HIGH_SEVERITY_HOUSES = {7,8}
_MARS_EXCEPTIONS = {2:{"Gemini","Virgo"},12:{"Taurus","Libra"},4:{"Aries","Scorpio"},7:{"Capricorn","Cancer"},8:{"Sagittarius","Pisces"}}
_MARS_EXEMPT_SIGNS = {"Aquarius","Leo"}
# =============================================================================
# Ashtakoot 8因子 (36分制)
# =============================================================================
def _calc_varna(m_sign, f_sign):
return 1.0 if VARNA[m_sign] <= VARNA[f_sign] else 0.0
def _calc_tara(m_nak, f_nak):
pts = 0.0
if (f_nak-m_nak)%9 not in (2,4,6): pts += 1.5
if (m_nak-f_nak)%9 not in (2,4,6): pts += 1.5
return pts
def _calc_yoni(m_nak, f_nak):
my, fy = YONI_ANIMALS[m_nak], YONI_ANIMALS[f_nak]
if my == fy: return 4.0
if YONI_ENEMIES.get(my) == fy: return 0.0
return 2.0
def _calc_graha_maitri(m_sign, f_sign):
ml, fl = SIGN_LORDS[m_sign], SIGN_LORDS[f_sign]
def _check(p1, p2):
if p1==p2: return 1.0
if p2 in NATURAL_FRIENDS.get(p1,[]): return 1.0
if p2 in NATURAL_ENEMIES.get(p1,[]): return 0.0
return 0.5
total = _check(ml,fl) + _check(fl,ml)
if total == 2.0: return 5.0
if total == 1.5: return 4.0
if total == 1.0: return 3.0
if total == 0.5: return 1.0
return 0.0
def _calc_gana(m_nak, f_nak):
mg, fg = GANA[m_nak], GANA[f_nak]
if mg == fg: return 6.0
if mg=="Deva" and fg=="Manushya": return 6.0
if mg=="Manushya" and fg=="Deva": return 5.0
if "Rakshasa" in (mg,fg) and "Manushya" in (mg,fg): return 0.0
if mg=="Rakshasa" and fg=="Deva": return 1.0
if fg=="Rakshasa" and mg=="Deva": return 0.0
return 0.0
def _calc_bhakoot(m_sign, f_sign):
mi, fi = SIGNS.index(m_sign), SIGNS.index(f_sign)
diff = (fi - mi) % 12 + 1
return 7.0 if diff in (1,7,3,11,4,10) else 0.0
def _calc_nadi(m_nak, f_nak):
return 8.0 if NADI[m_nak] != NADI[f_nak] else 0.0
def _calc_vedha(m_nak, f_nak):
for a,b in VEDHA_PAIRS:
if (m_nak==a and f_nak==b) or (m_nak==b and f_nak==a): return "bad"
return "good"
def _calc_rajju(m_nak, f_nak):
for group, indices in RAJJU_GROUPS.items():
if m_nak in indices and f_nak in indices:
return {"result":"bad","group":group,"effect":RAJJU_EFFECTS[group]}
return {"result":"good","group":None,"effect":""}
def _calc_mahendra(m_nak, f_nak):
cnt = ((m_nak-f_nak)%27)+1
return "good" if cnt in (4,7,10,13,16,19,22,25) else "bad"
def _calc_stree_deergha(m_nak, f_nak):
cnt = ((m_nak-f_nak)%27)+1
return "good" if cnt >= 9 else "bad"
# =============================================================================
# 主计算函数
# =============================================================================
def calc_ashtakoot(male_moon_degree: float, female_moon_degree: float,
male_chart: Dict = None, female_chart: Dict = None) -> Dict:
"""
计算16因子Ashtakoot兼容性匹配（基于dashaflow MIT算法）。
Args:
male_moon_degree: 男性月亮黄道经度(0-360)
female_moon_degree: 女性月亮黄道经度(0-360)
male_chart: 男性星盘数据(可选，用于附加Kuta)
female_chart: 女性星盘数据(可选)
Returns:
完整的兼容性分析
"""
m_nak = int(male_moon_degree / (360.0/27.0))
f_nak = int(female_moon_degree / (360.0/27.0))
m_sign = SIGNS[int(male_moon_degree/30)%12]
f_sign = SIGNS[int(female_moon_degree/30)%12]
scores = {
"Varna": _calc_varna(m_sign, f_sign),
"Vashya": 2.0, # Simplified
"Tara": _calc_tara(m_nak, f_nak),
"Yoni": _calc_yoni(m_nak, f_nak),
"GrahaMaitri": _calc_graha_maitri(m_sign, f_sign),
"Gana": _calc_gana(m_nak, f_nak),
"Bhakoot": _calc_bhakoot(m_sign, f_sign),
"Nadi": _calc_nadi(m_nak, f_nak),
}
total = sum(scores.values())
rajju = _calc_rajju(m_nak, f_nak)
additional = {
"Mahendra": _calc_mahendra(m_nak, f_nak),
"StreeDeergha": _calc_stree_deergha(m_nak, f_nak),
"Vedha": _calc_vedha(m_nak, f_nak),
"Rajju": rajju,
}
# Dosha缓解逻辑
exceptions = []
if scores["Nadi"]==0 and scores["Bhakoot"]>0 and rajju["result"]=="good":
exceptions.append("Nadi Dosha mitigated by good Bhakoot and Rajju")
if rajju["result"]=="bad" and scores["GrahaMaitri"]>=4 and scores["Bhakoot"]>0 and scores["Tara"]>=1.5 and additional["Mahendra"]=="good":
exceptions.append("Rajju Dosha mitigated")
assessment = "优秀" if total>=28 else "良好" if total>=21 else "一般" if total>=18 else "不推荐"
approved = total >= 18.0 and (scores["Nadi"]>0 or len(exceptions)>0)
return {
"version": "3.8-dashaflow-mit-adapted",
"method": "Ashtakoot 16因子兼容性分析 (dashaflow MIT)",
"male": {"moon_sign":m_sign,"nakshatra":NAKSHATRAS[m_nak],"gana":GANA[m_nak],"nadi":NADI[m_nak],"yoni":YONI_ANIMALS[m_nak]},
"female": {"moon_sign":f_sign,"nakshatra":NAKSHATRAS[f_nak],"gana":GANA[f_nak],"nadi":NADI[f_nak],"yoni":YONI_ANIMALS[f_nak]},
"scores": scores,
"total_score": total,
"max_score": 36.0,
"assessment": assessment,
"is_approved": approved,
"additional_kutas": additional,
"exceptions": exceptions,
}
def calc_synastry(male_chart: Dict, female_chart: Dict) -> Dict:
"""
Backward-compatible synastry wrapper used by integration tests and older CLI paths.
Expected chart keys:
- moon_lon: Moon longitude in degrees (required)
- mars_lon / asc_lon / gender: optional, reserved for Kuja and extended factors
"""
if "moon_lon" not in male_chart or "moon_lon" not in female_chart:
raise ValueError("calc_synastry requires moon_lon in both male_chart and female_chart")
result = calc_ashtakoot(
float(male_chart["moon_lon"]),
float(female_chart["moon_lon"]),
male_chart=male_chart,
female_chart=female_chart,
)
# Historical test/API compatibility: expose BadConstellations as an additional Kuta.
# A bad constellation condition is present if Vedha or Rajju is adverse.
vedha_bad = result["additional_kutas"].get("Vedha") == "bad"
rajju_bad = result["additional_kutas"].get("Rajju", {}).get("result") == "bad"
result["additional_kutas"]["BadConstellations"] = "bad" if (vedha_bad or rajju_bad) else "good"
return result
scripts/tajika.py
+31 -9
View File
@@ -96,14 +96,36 @@ def _calc_formula_saham(
second = _resolve_saham_operand(formula[1], planet_lons, asc_lon, computed)
third = _resolve_saham_operand(formula[2], planet_lons, asc_lon, computed)
result = (third + (first - second)) % 360
# references/saham_rules.json: add one sign when Ascendant is not on the
# forward zodiacal arc from the first formula point to the second.
ascendant_between_points = (
first <= asc_lon <= second
if first <= second
else asc_lon >= first or asc_lon <= second
)
return result if ascendant_between_points else (result + 30.0) % 360
# PVR §28.8.1, p394: in A-B+C, C must lie on the forward arc B -> A.
third_between_points = (third - second) % 360 <= (first - second) % 360
return result if third_between_points else (result + 30.0) % 360
def _formula_saham_trace(
rule_name: str,
planet_lons: Dict[str, float],
asc_lon: float,
is_day: bool,
computed: Dict[str, float],
) -> Dict:
"""Calculate a rule-backed Saham while preserving its sensitive convention."""
rule = _SAHAM_RULES['sahams'][rule_name]
formula = rule['formula_day'] if is_day else rule['formula_night']
first = _resolve_saham_operand(formula[0], planet_lons, asc_lon, computed)
second = _resolve_saham_operand(formula[1], planet_lons, asc_lon, computed)
third = _resolve_saham_operand(formula[2], planet_lons, asc_lon, computed)
base_longitude = (third + (first - second)) % 360
third_between_points = (third - second) % 360 <= (first - second) % 360
exception_applied = not third_between_points
return {
'longitude': (base_longitude + 30.0) % 360 if exception_applied else base_longitude,
'base_longitude': base_longitude,
'rule_name': rule_name,
'formula_operands': formula,
'formula_variant': 'day' if is_day else 'night',
'zodiacal_order_exception_applied': exception_applied,
'rule_source': 'references/saham_rules.json',
}
def calc_muntha(birth_asc_idx: int, age: int) -> Dict:
@@ -1592,7 +1614,7 @@ def calc_all_sahams(planet_lons: Dict[str, float],
'boundary': 'No solar-house day/night proxy is permitted in production.',
}
from saham_daynight import determine_daytime
daynight = determine_daytime(birth_dt, lat=float(lat), lon=float(lon), tz=float(tz))
daynight = determine_daytime(birth_dt, lat=float(lat), lon=float(lon), tz=float(tz), rising="hindu")
sun_lon = planet_lons.get('Sun', 0)
moon_lon = planet_lons.get('Moon', 0)
mars_lon = planet_lons.get('Mars', 0)
scripts/technique_registry.py
+1 -1
View File
@@ -365,7 +365,7 @@ def _register_existing_techniques():
module_path="transit.py", compute_func="computeTransit"),
TechniqueSpec("synastry", "合盘系统", TechniqueCategory.SYNASTRY,
TechniqueLevel.L2_STANDARD, TechniqueStatus.STABLE,
module_path="synastry.py", compute_func="calc_synastry"),
module_path="marriage_match.py", compute_func="build_marriage_match_dossier"),
TechniqueSpec("prashna", "问事系统", TechniqueCategory.PRASHNA,
TechniqueLevel.L2_STANDARD, TechniqueStatus.WIP,
module_path="prashna_context.py", compute_func="build_prashna_context"),
scripts/unified_western_reading.py
+268
View File
@@ -0,0 +1,268 @@
"""Auditable orchestration for the native tropical Western calculation layers."""
from __future__ import annotations
try: # reached as a script: this directory is on the path, the root is not
import _script_root # noqa: F401
except ModuleNotFoundError: # pragma: no cover - reached as scripts.x instead
try:
from scripts import _script_root # noqa: F401
except ModuleNotFoundError:
pass
import argparse
import json
from datetime import date
from typing import Any, Callable
try:
import swisseph as swe
except ImportError: # pragma: no cover - optional runtime dependency
swe = None
try:
from scripts.classical_astrology import build_classical_astrology_packet, build_classical_usable_loop_audit_packet
from scripts.western_chart_engine import build_tropical_natal_chart
from scripts.western_chart_engine import build_draconic_chart, build_relocation_chart
from scripts.western_timing_engine import (
calculate_annual_profection,
calculate_converse_secondary_progressions,
calculate_converse_solar_arc_directions,
calculate_lunar_return,
calculate_lunar_return_series,
calculate_midpoints,
calculate_parans_status,
calculate_secondary_progressions,
calculate_solar_arc_directions,
calculate_solar_return,
calculate_tertiary_progressed_moon,
calculate_transit_duration_scan,
calculate_transit_to_natal,
calculate_zodiacal_release,
)
except ModuleNotFoundError: # pragma: no cover - direct script execution
from classical_astrology import build_classical_astrology_packet, build_classical_usable_loop_audit_packet
from western_chart_engine import build_tropical_natal_chart
from western_chart_engine import build_draconic_chart, build_relocation_chart
from western_timing_engine import (
calculate_annual_profection,
calculate_converse_secondary_progressions,
calculate_converse_solar_arc_directions,
calculate_lunar_return,
calculate_lunar_return_series,
calculate_midpoints,
calculate_parans_status,
calculate_secondary_progressions,
calculate_solar_arc_directions,
calculate_solar_return,
calculate_tertiary_progressed_moon,
calculate_transit_duration_scan,
calculate_transit_to_natal,
calculate_zodiacal_release,
)
def _executed(inputs: dict[str, Any], calculation: Callable[[], dict[str, Any]]) -> dict[str, Any]:
try:
return {"status": "executed", "input": inputs, "raw_result": calculation()}
except Exception as exc: # pragma: no cover - defensive orchestration boundary
return {
"status": "blocked",
"input": inputs,
"reason": f"{exc.__class__.__name__}: {exc}",
}
def _blocked(reason: str, inputs: dict[str, Any] | None = None) -> dict[str, Any]:
result: dict[str, Any] = {"status": "blocked", "reason": reason}
if inputs:
result["input"] = inputs
return result
def _reference_layer(
reference_date: str | None,
birth: dict[str, Any],
calculation: Callable[[str], dict[str, Any]],
) -> dict[str, Any]:
if not reference_date:
return _blocked("reference_date is required")
return _executed({"reference_date": reference_date}, lambda: calculation(reference_date))
def _overall_status(layers: dict[str, dict[str, Any]]) -> str:
return "complete" if all(layer["status"] == "executed" for layer in layers.values()) else "partial"
def _eclipse_events(reference_date: str) -> dict[str, Any]:
if swe is None:
raise RuntimeError("swisseph is unavailable")
start = date.fromisoformat(reference_date)
jd = swe.julday(start.year, start.month, start.day, 0.0)
events = []
for kind, finder in (("solar", swe.sol_eclipse_when_glob), ("lunar", swe.lun_eclipse_when)):
flags, times = finder(jd)
year, month, day, hour = swe.revjul(times[0])
events.append({"kind": kind, "maximum_utc": f"{year:04d}-{month:02d}-{day:02d}T{int(hour):02d}:{int((hour % 1) * 60):02d}:00Z", "swiss_event_flags": int(flags)})
return {"technique": "eclipse_triggers", "reference_date": reference_date, "events": events, "boundary": "Geometric eclipse triggers only; no predictive interpretation is inferred."}
def build_unified_western_reading(
*,
reference_date: str | None = None,
duration_scan_end_date: str | None = None,
lunar_return_months: int = 1,
relocation_latitude: float | None = None,
relocation_longitude: float | None = None,
relocation_timezone: str | float | int | None = None,
zodiacal_release_lot: str = "spirit",
zodiacal_release_levels: int = 2,
include_classical_usable_loop: bool = False,
classical_event_window_hours: int = 4,
classical_event_step_hours: int = 2,
classical_blind_scoring_manifest: str | None = None,
classical_reviewer_manifest: str | None = None,
**birth: Any,
) -> dict[str, Any]:
"""Run every available native Western layer and preserve raw evidence.
``reference_date`` drives the one-day timing techniques. A duration scan
additionally needs an explicit end date, so a caller cannot unknowingly
request an unbounded scan. Eclipse geometry uses the local Swiss Ephemeris
calculation and does not infer an event interpretation.
"""
layers: dict[str, dict[str, Any]] = {
"natal": _executed(dict(birth), lambda: build_tropical_natal_chart(**birth)),
"classical": _executed(
{"reference_date": reference_date},
lambda: build_classical_astrology_packet(reference_date=reference_date, **birth),
),
"draconic": _executed(dict(birth), lambda: build_draconic_chart(**birth)),
"transits": _reference_layer(
reference_date, birth, lambda target: calculate_transit_to_natal(target_date=target, **birth)
),
"solar_return": _reference_layer(
reference_date,
birth,
lambda target: calculate_solar_return(target_year=date.fromisoformat(target).year, **birth),
),
"secondary_progressions": _reference_layer(
reference_date, birth, lambda target: calculate_secondary_progressions(target_date=target, **birth)
),
"tertiary_moon": _reference_layer(
reference_date, birth, lambda target: calculate_tertiary_progressed_moon(target_date=target, **birth)
),
"solar_arc": _reference_layer(
reference_date, birth, lambda target: calculate_solar_arc_directions(target_date=target, **birth)
),
"midpoints": _reference_layer(
reference_date, birth, lambda target: calculate_midpoints(target_date=target, **birth)
),
"lunar_return": _reference_layer(
reference_date, birth, lambda target: calculate_lunar_return_series(
target_date=target, months=lunar_return_months, **birth
)
),
"parans": _reference_layer(
reference_date, birth, lambda target: calculate_parans_status(target_date=target, **birth)
),
"profection": _reference_layer(
reference_date, birth, lambda target: calculate_annual_profection(target_date=target, **birth)
),
"zodiacal_release": _reference_layer(
reference_date,
birth,
lambda target: calculate_zodiacal_release(
target_date=target,
lot=zodiacal_release_lot,
levels=int(zodiacal_release_levels),
**birth,
),
),
}
if relocation_latitude is None or relocation_longitude is None or relocation_timezone is None:
layers["relocation"] = _blocked(
"relocation_latitude, relocation_longitude, and relocation_timezone are required"
)
else:
layers["relocation"] = _executed(
{
"relocation_latitude": relocation_latitude,
"relocation_longitude": relocation_longitude,
"relocation_timezone": relocation_timezone,
},
lambda: build_relocation_chart(
relocation_latitude=float(relocation_latitude),
relocation_longitude=float(relocation_longitude),
relocation_timezone=relocation_timezone,
**birth,
),
)
if reference_date:
layers["converse"] = _executed(
{"reference_date": reference_date},
lambda: {
"secondary_progressions": calculate_converse_secondary_progressions(
target_date=reference_date, **birth
),
"solar_arc_directions": calculate_converse_solar_arc_directions(
target_date=reference_date, **birth
),
},
)
else:
layers["converse"] = _blocked("reference_date is required")
if not reference_date:
layers["transit_duration"] = _blocked("reference_date is required")
elif not duration_scan_end_date:
layers["transit_duration"] = _blocked("duration_scan_end_date is required")
else:
layers["transit_duration"] = _executed(
{"start_date": reference_date, "end_date": duration_scan_end_date},
lambda: calculate_transit_duration_scan(
start_date=reference_date, end_date=duration_scan_end_date, **birth
),
)
layers["eclipses"] = _reference_layer(reference_date, birth, _eclipse_events)
if include_classical_usable_loop:
layers["classical_usable_loop"] = _executed(
{
"reference_date": reference_date,
"event_window_hours": classical_event_window_hours,
"event_step_hours": classical_event_step_hours,
"blind_scoring_manifest": classical_blind_scoring_manifest,
"reviewer_manifest": classical_reviewer_manifest,
},
lambda: build_classical_usable_loop_audit_packet(
reference_date=reference_date,
event_window_hours=int(classical_event_window_hours),
event_step_hours=int(classical_event_step_hours),
blind_scoring_manifest_path=classical_blind_scoring_manifest,
reviewer_manifest_path=classical_reviewer_manifest,
**birth,
),
)
return {
"system": "western_astrology",
"status": _overall_status(layers),
"reference_date": reference_date,
"layers": layers,
"boundary": "Raw native calculations only; this entrypoint makes no predictive interpretation.",
}
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Run the unified native Western calculation layers.")
parser.add_argument("--input-json", required=True, help="Structured birth and timing input as JSON.")
args = parser.parse_args(argv)
payload = json.loads(args.input_json)
if not isinstance(payload, dict):
raise ValueError("--input-json must decode to an object")
print(json.dumps(build_unified_western_reading(**payload), ensure_ascii=False, default=str))
return 0
if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())
scripts/western_chart_engine.py
+109
View File
@@ -153,6 +153,21 @@ def _birth_zone(value: str | float | int):
return fixed_timezone(timedelta(hours=offset)), f"UTC{offset:+g}"
def _birth_times(
*,
year: int,
month: int,
day: int,
hour: int,
minute: int,
second: int,
timezone: str | float | int,
) -> tuple[datetime, datetime, str]:
zone, timezone_label = _birth_zone(timezone)
local = datetime(year, month, day, hour, minute, second, tzinfo=zone)
utc = local.astimezone(ZoneInfo("UTC"))
return local, utc, timezone_label
def build_tropical_natal_chart(
*,
year: int,
@@ -217,6 +232,100 @@ def build_tropical_natal_chart(
}
def build_draconic_chart(**birth: Any) -> dict[str, Any]:
"""Derive a draconic chart by rotating tropical longitudes to the true node Aries point."""
natal_chart = build_tropical_natal_chart(**birth)
natal = natal_chart["natal"]
node_longitude = natal["planets"]["true_node"]["longitude"]
def _rotate_point(point: dict[str, Any]) -> dict[str, Any]:
rotated = _point(point["longitude"] - node_longitude)
if "house" in point:
rotated["house"] = point["house"]
if "speed_longitude" in point:
rotated["speed_longitude"] = point["speed_longitude"]
rotated["retrograde"] = point.get("retrograde", False)
return rotated
rotated_planets = {name: _rotate_point(point) for name, point in natal["planets"].items()}
rotated_angles = {name: _rotate_point(point) for name, point in natal["angles"].items()}
rotated_houses = [
{"house": row["house"], "cusp": _rotate_point(row["cusp"])}
for row in natal["houses"]
]
aspect_points = {**rotated_planets, "ascendant": rotated_angles["ascendant"], "mc": rotated_angles["mc"]}
return {
"technique": "draconic_chart",
"status": "used",
"source_engine": natal_chart["source_engine"],
"base_zodiac": natal_chart["zodiac"],
"derived_zodiac": "draconic",
"node_basis": "true_node",
"node_longitude": node_longitude,
"draconic": {
"ascendant": rotated_angles["ascendant"],
"mc": rotated_angles["mc"],
"angles": rotated_angles,
"planets": rotated_planets,
"houses": rotated_houses,
"aspects": _aspects(aspect_points),
"distribution": _distribution(rotated_planets),
"house_ruler_chains": _ruler_chains(
[row["cusp"]["longitude"] for row in rotated_houses],
rotated_planets,
),
},
"boundary": "Longitude rotation to the true-node Aries point only; interpretation and timing remain separate audited layers.",
}
def build_relocation_chart(
*,
relocation_latitude: float,
relocation_longitude: float,
relocation_timezone: str | float | int,
house_system: str = "P",
**birth: Any,
) -> dict[str, Any]:
"""Calculate a relocated chart for the same UTC birth instant at a new location."""
local, utc, _ = _birth_times(
year=int(birth["year"]),
month=int(birth["month"]),
day=int(birth["day"]),
hour=int(birth["hour"]),
minute=int(birth["minute"]),
second=int(birth.get("second", 0)),
timezone=birth["timezone"],
)
relocation_zone, relocation_label = _birth_zone(relocation_timezone)
relocation_local = utc.astimezone(relocation_zone)
relocated_chart = build_tropical_natal_chart(
year=relocation_local.year,
month=relocation_local.month,
day=relocation_local.day,
hour=relocation_local.hour,
minute=relocation_local.minute,
second=relocation_local.second,
latitude=float(relocation_latitude),
longitude=float(relocation_longitude),
timezone=relocation_timezone,
house_system=house_system,
)
return {
"technique": "relocation_chart",
"status": "used",
"source_engine": relocated_chart["source_engine"],
"same_birth_utc": utc.isoformat(),
"source_local_time": local.isoformat(),
"relocation_local_time": relocation_local.isoformat(),
"relocation_timezone": relocation_label,
"relocation_coordinates": {
"latitude": float(relocation_latitude),
"longitude": float(relocation_longitude),
},
"relocated_chart": relocated_chart,
"boundary": "Same UTC birth instant with relocated coordinates and timezone; interpretation and relocation outcome judgment remain separate audited layers.",
}
def _request_field(body: dict[str, Any], *keys: str) -> Any:
for key in keys:
if key in body and body[key] not in (None, ""):
scripts/western_real_case_calibration.py
+19
View File
@@ -16,6 +16,25 @@ except ImportError:
DEFAULT_MANIFEST = Path("references/real_case_calibration/replay_manifest.json")
def _build_calibration_summary(primary_theme: str = "general") -> dict:
"""Import shim for classical_astrology.
Upstream's scoring body is not in this tree. The caller only reads these
keys and treats a missing gate as blocked, so full-reading can import the
classical packet without inventing a specificity score.
"""
return {
"status": "blocked",
"reason": "specificity_gate_not_ported",
"primary_theme": primary_theme,
"false_positive_audit": {"frozen_negative_holdout": {"status": "blocked"}},
"positive_replay": {},
"positive_holdout": {},
"domain_replay_scoring": {},
"review_pack": {},
}
def build_case(case_id: str, manifest_path: Path = DEFAULT_MANIFEST) -> dict:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
case = next((row for row in manifest["cases"] if row["case_id"] == case_id), None)