diff --git a/scripts/event_prediction_model.py b/scripts/event_prediction_model.py index 68104f54..d4495230 100644 --- a/scripts/event_prediction_model.py +++ b/scripts/event_prediction_model.py @@ -1,26 +1,34 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ -事件预测模型 -基于三层验证法（静态星盘 + Dasha激活 + 过境触发）预测事件 +事件预测模型 v5.0 — 三层验证法（全功能版） +第一层：静态星盘分析（Yoga、宫位、行星聚集、Vivah Saham、Ashtakavarga） +第二层：Dasha激活（Vimshottari + Chara Dasha with Antardasha） +第三层：过境触发（Double Transit PAC+D9、Transit LL/7L、行星聚集、Vivah Saham激活） + +v5.0 重大升级：所有分析函数从空壳升级为真实逻辑，接入 jyotish_engine 的全部计算能力 """ from typing import Dict, List, Optional, Tuple -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime from enum import Enum +# ============================================================================ +# 数据结构 +# ============================================================================ + class EventType(Enum): """事件类型""" - MARRIAGE = "婚姻" - CAREER = "职业" - WEALTH = "财富" - HEALTH = "健康" - EDUCATION = "教育" - CHILDREN = "子女" - TRAVEL = "旅行" - SPIRITUAL = "灵性" + MARRIAGE = "marriage" + CAREER = "career" + WEALTH = "wealth" + HEALTH = "health" + EDUCATION = "education" + CHILDREN = "children" + TRAVEL = "travel" + SPIRITUAL = "spiritual" class RiskLevel(Enum): @@ -31,6 +39,13 @@ class RiskLevel(Enum): CRITICAL = "极高" +class Confidence(Enum): + """置信度（三级）""" + HIGH = "高" # 多层验证确认 + MEDIUM = "中" # 两层验证 + LOW = "低" # 单层信号或未经验证 + + @dataclass class Prediction: """预测结果""" @@ -38,529 +53,589 @@ class Prediction: description: str probability: float # 0-100 risk_level: RiskLevel - timing: Optional[str] - key_factors: List[str] - recommendations: List[str] + confidence: Confidence = Confidence.LOW + timing: Optional[str] = None + timing_windows: List[Dict] = field(default_factory=list) + key_factors: List[str] = field(default_factory=list) + recommendations: List[str] = field(default_factory=list) + dasha_signals: List[str] = field(default_factory=list) + transit_signals: List[str] = field(default_factory=list) + method: str = "三层验证法 v5.0" +# ============================================================================ +# 常量 +# ============================================================================ + +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', +} + +# 事件→宫位映射 +EVENT_HOUSES = { + EventType.MARRIAGE: [7, 2, 11, 5], + EventType.CAREER: [10, 6, 9, 11], + EventType.WEALTH: [2, 11, 5, 9], + EventType.HEALTH: [6, 8, 12, 1], + EventType.EDUCATION: [4, 5, 9, 2], + EventType.CHILDREN: [5, 9, 2, 11], + EventType.TRAVEL: [3, 9, 12, 7], + EventType.SPIRITUAL: [9, 12, 8, 4], +} + +# 事件→ Karaka 映射 +EVENT_KARAKAS = { + EventType.MARRIAGE: ['Venus', 'Jupiter', '7L'], + EventType.CAREER: ['Sun', 'Mercury', 'Saturn', '10L'], + EventType.WEALTH: ['Jupiter', 'Venus', 'Mercury', '2L'], + EventType.HEALTH: ['Saturn', 'Moon', '6L'], + EventType.EDUCATION: ['Mercury', 'Jupiter', '4L'], + EventType.CHILDREN: ['Jupiter', '5L'], + EventType.TRAVEL: ['Mercury', 'Rahu', '9L'], + EventType.SPIRITUAL: ['Jupiter', 'Saturn', 'Ketu', '9L'], +} + +# Vimshottari Dasha 行星 +DASHA_PLANETS = ['Ketu', 'Venus', 'Sun', 'Moon', 'Mars', 'Rahu', 'Jupiter', 'Saturn', 'Mercury'] +DASHA_YEARS = { + 'Ketu': 7, 'Venus': 20, 'Sun': 6, 'Moon': 10, 'Mars': 7, + 'Rahu': 18, 'Jupiter': 16, 'Saturn': 19, 'Mercury': 17, +} + +# Graha Drishti 相位 +PLANET_ASPECTS = { + 'Sun': [7], 'Moon': [7], 'Mars': [4, 7, 8], 'Mercury': [7], + 'Jupiter': [5, 7, 9], 'Venus': [7], 'Saturn': [3, 7, 10], + 'Rahu': [5, 7, 9], 'Ketu': [5, 7, 9], +} + + +# ============================================================================ +# 预测引擎 +# ============================================================================ + class EventPredictionModel: - """事件预测模型""" - - def __init__(self, chart_data: Dict, dasha_data: Optional[Dict] = None, - transit_data: Optional[Dict] = None): - """ - 初始化事件预测模型 - - Args: - chart_data: 星盘数据 - dasha_data: 大运周期数据 - transit_data: 过境数据 - """ + """事件预测模型 v5.0 — 全功能三层验证法""" + + def __init__(self, chart_data: Dict, dasha_data: Optional[Dict] = None, + transit_data: Optional[Dict] = None, + congregation_data: Optional[Dict] = None, + vivah_saham_data: Optional[Dict] = None, + chara_dasha_data: Optional[Dict] = None, + double_transit_data: Optional[Dict] = None, + ll7l_data: Optional[Dict] = None): self.chart = chart_data - self.dasha = dasha_data - self.transit = transit_data + self.dasha = dasha_data or {} + self.transit = transit_data or {} + self.congregation = congregation_data or {} + self.vivah_saham = vivah_saham_data or {} + self.chara_dasha = chara_dasha_data or {} + self.double_transit = double_transit_data or {} + self.ll7l = ll7l_data or {} self.predictions = [] - + + # 预计算 ascendant 索引 + asc_sign = self.chart.get('ascendant', {}).get('sign', 'Aries') + self.asc_idx = SIGNS.index(asc_sign) if asc_sign in SIGNS else 0 + + # 预计算宫主星 + self.house_lords = {} + for i in range(12): + h_sign = SIGNS[(self.asc_idx + i) % 12] + self.house_lords[i + 1] = SIGN_LORDS.get(h_sign, '') + + # 预计算行星宫位 + self.planet_houses = {} + for pn, pd in self.chart.get('planets', {}).items(): + if isinstance(pd, dict) and 'house' in pd: + self.planet_houses[pn] = pd['house'] + + # 预计算行星经度（度数精确） + self.planet_lons = {} + for pn, pd in self.chart.get('planets', {}).items(): + if isinstance(pd, dict) and 'degree' in pd: + self.planet_lons[pn] = pd['degree'] + def predict_all_events(self) -> List[Prediction]: """预测所有类型的事件""" predictions = [] - - # 1. 婚姻预测 - marriage_pred = self.predict_marriage() - if marriage_pred: - predictions.append(marriage_pred) - - # 2. 职业预测 - career_pred = self.predict_career() - if career_pred: - predictions.append(career_pred) - - # 3. 财富预测 - wealth_pred = self.predict_wealth() - if wealth_pred: - predictions.append(wealth_pred) - - # 4. 健康预测 - health_pred = self.predict_health() - if health_pred: - predictions.append(health_pred) - + for evt_type in EventType: + pred = self.predict_event(evt_type) + if pred: + predictions.append(pred) self.predictions = predictions return predictions - - def predict_marriage(self) -> Optional[Prediction]: - """ - 婚姻预测（三层验证法） - - 第一层：静态星盘分析 - - 7宫状态 - - 金星状态 - - 婚姻相关Yoga - - 第二层：Dasha激活 - - 当前Dasha主星是否掌管7宫或金星 - - 当前Antardasha主星是否掌管7宫或金星 - - 第三层：过境触发 - - 木星过境7宫 - - 土星过境7宫 - """ - # 第一层：静态分析 - static_factors = self._analyze_marriage_static() - - # 第二层：Dasha激活 - dasha_activation = self._analyze_marriage_dasha() - - # 第三层：过境触发 - transit_trigger = self._analyze_marriage_transit() - - # 综合判断 - probability = self._calculate_marriage_probability( - static_factors, dasha_activation, transit_trigger - ) - - risk_level = self._assess_marriage_risk(static_factors) - - timing = self._predict_marriage_timing(dasha_activation, transit_trigger) - - key_factors = [] - if static_factors.get('7th_house_afflicted'): - key_factors.append("7宫受克") - if static_factors.get('venus_weak'): - key_factors.append("金星弱势") - if static_factors.get('marriage_yoga'): - key_factors.append("婚姻Yoga") - - recommendations = [] - if risk_level in [RiskLevel.HIGH, RiskLevel.CRITICAL]: - recommendations.append("建议进行详细合婚分析") - recommendations.append("考虑补救措施") + + def predict_event(self, evt_type: EventType) -> Optional[Prediction]: + """预测单个事件类型（核心方法）""" + # ── 第一层：静态星盘分析 ── + static = self._layer1_static(evt_type) + + # ── 第二层：Dasha激活 ── + dasha_signals = self._layer2_dasha(evt_type) + + # ── 第三层：过境触发 ── + transit_signals = self._layer3_transit(evt_type) + + # ── 综合概率 ── + probability = self._calc_probability(static, dasha_signals, transit_signals) + risk = self._assess_risk(static, evt_type) + + # ── 置信度判定 ── + layer_count = sum([ + 1 if static['total_signals'] > 0 else 0, + 1 if len(dasha_signals) > 0 else 0, + 1 if len(transit_signals) > 0 else 0, + ]) + if layer_count >= 3: + confidence = Confidence.HIGH + elif layer_count >= 2: + confidence = Confidence.MEDIUM else: - recommendations.append("婚姻前景良好") - - return Prediction( - event_type=EventType.MARRIAGE, - description="婚姻预测", - probability=probability, - risk_level=risk_level, - timing=timing, - key_factors=key_factors, - recommendations=recommendations - ) - - def predict_career(self) -> Optional[Prediction]: - """职业预测""" - # 第一层：静态分析 - static_factors = self._analyze_career_static() - - # 第二层：Dasha激活 - dasha_activation = self._analyze_career_dasha() - - # 第三层：过境触发 - transit_trigger = self._analyze_career_transit() - - probability = self._calculate_career_probability( - static_factors, dasha_activation, transit_trigger - ) - - risk_level = self._assess_career_risk(static_factors) - timing = self._predict_career_timing(dasha_activation, transit_trigger) - + confidence = Confidence.LOW + + # ── 时间窗预测 ── + timing_windows = self._calc_timing_windows(evt_type, dasha_signals, transit_signals) + + # ── 关键因素汇总 ── key_factors = [] - if static_factors.get('10th_house_strong'): - key_factors.append("10宫强") - if static_factors.get('career_yoga'): - key_factors.append("职业Yoga") - - recommendations = [] - if risk_level == RiskLevel.LOW: - recommendations.append("职业前景良好") - else: - recommendations.append("需要努力和规划") - + for s in static['signals']: + key_factors.append(f"[静态] {s}") + for s in dasha_signals: + key_factors.append(f"[Dasha] {s}") + for s in transit_signals: + key_factors.append(f"[Transit] {s}") + + # ── 建议 ── + recommendations = self._gen_recommendations(evt_type, static, dasha_signals, transit_signals) + + # ── 时间描述 ── + timing = self._describe_timing(timing_windows, dasha_signals, transit_signals) + return Prediction( - event_type=EventType.CAREER, - description="职业预测", + event_type=evt_type, + description=evt_type.value, probability=probability, - risk_level=risk_level, + risk_level=risk, + confidence=confidence, timing=timing, + timing_windows=timing_windows, key_factors=key_factors, - recommendations=recommendations + recommendations=recommendations, + dasha_signals=dasha_signals, + transit_signals=transit_signals, ) - - def predict_wealth(self) -> Optional[Prediction]: - """财富预测""" - static_factors = self._analyze_wealth_static() - dasha_activation = self._analyze_wealth_dasha() - transit_trigger = self._analyze_wealth_transit() - - probability = self._calculate_wealth_probability( - static_factors, dasha_activation, transit_trigger - ) - - risk_level = self._assess_wealth_risk(static_factors) - timing = self._predict_wealth_timing(dasha_activation, transit_trigger) - - key_factors = [] - if static_factors.get('2nd_house_strong'): - key_factors.append("2宫强") - if static_factors.get('dhana_yoga'): - key_factors.append("财富Yoga") - - recommendations = ["注意财务管理"] - - return Prediction( - event_type=EventType.WEALTH, - description="财富预测", - probability=probability, - risk_level=risk_level, - timing=timing, - key_factors=key_factors, - recommendations=recommendations - ) - - def predict_health(self) -> Optional[Prediction]: - """健康预测""" - static_factors = self._analyze_health_static() - dasha_activation = self._analyze_health_dasha() - transit_trigger = self._analyze_health_transit() - - probability = self._calculate_health_probability( - static_factors, dasha_activation, transit_trigger - ) - - risk_level = self._assess_health_risk(static_factors) - timing = self._predict_health_timing(dasha_activation, transit_trigger) - - key_factors = [] - if static_factors.get('6th_house_afflicted'): - key_factors.append("6宫受克") - if static_factors.get('moon_weak'): - key_factors.append("月亮弱势") - - recommendations = [] - if risk_level in [RiskLevel.HIGH, RiskLevel.CRITICAL]: - recommendations.append("建议定期体检") - recommendations.append("注意健康生活方式") - - return Prediction( - event_type=EventType.HEALTH, - description="健康预测", - probability=probability, - risk_level=risk_level, - timing=timing, - key_factors=key_factors, - recommendations=recommendations - ) - - # ===== 静态分析方法 ===== - - def _analyze_marriage_static(self) -> Dict: - """婚姻静态分析""" - factors = {} - - if 'planets' in self.chart: - # 检查7宫 - factors['7th_house_afflicted'] = False # 简化版 - factors['venus_weak'] = False - - # 检查金星 - venus = self.chart['planets'].get('Venus') - if venus and venus.get('status') == 'debilitated': - factors['venus_weak'] = True - - # 检查婚姻Yoga - factors['marriage_yoga'] = False - - return factors - - def _analyze_career_static(self) -> Dict: - """职业静态分析""" - factors = {} - - if 'planets' in self.chart: - factors['10th_house_strong'] = False - factors['career_yoga'] = False - - # 检查10宫 - for planet, data in self.chart['planets'].items(): - if data.get('house') == 10 and data.get('status') in ['exalted', 'own_sign']: - factors['10th_house_strong'] = True - - return factors - - def _analyze_wealth_static(self) -> Dict: - """财富静态分析""" - factors = {} - - if 'planets' in self.chart: - factors['2nd_house_strong'] = False - factors['dhana_yoga'] = False - - # 检查2宫 - for planet, data in self.chart['planets'].items(): - if data.get('house') == 2 and planet in ['Jupiter', 'Venus']: - factors['2nd_house_strong'] = True - - return factors - - def _analyze_health_static(self) -> Dict: - """健康静态分析""" - factors = {} - - if 'planets' in self.chart: - factors['6th_house_afflicted'] = False - factors['moon_weak'] = False - - # 检查月亮 - moon = self.chart['planets'].get('Moon') - if moon and moon.get('status') == 'debilitated': - factors['moon_weak'] = True - - return factors - - # ===== Dasha激活分析 ===== - - def _analyze_marriage_dasha(self) -> Dict: - """婚姻Dasha激活分析""" - if not self.dasha: - return {'status': '缺失Dasha信息'} - - # 检查当前Dasha主星是否掌管7宫或金星 - return {'activated': False} - - def _analyze_career_dasha(self) -> Dict: - """职业Dasha激活分析""" - if not self.dasha: - return {'status': '缺失Dasha信息'} - - return {'activated': False} - - def _analyze_wealth_dasha(self) -> Dict: - """财富Dasha激活分析""" - if not self.dasha: - return {'status': '缺失Dasha信息'} - - return {'activated': False} - - def _analyze_health_dasha(self) -> Dict: - """健康Dasha激活分析""" - if not self.dasha: - return {'status': '缺失Dasha信息'} - - return {'activated': False} - - # ===== 过境触发分析 ===== - - def _analyze_marriage_transit(self) -> Dict: - """婚姻过境触发分析""" - if not self.transit: - return {'status': '缺失过境信息'} - - # 检查木星是否过境7宫 - return {'triggered': False} - - def _analyze_career_transit(self) -> Dict: - """职业过境触发分析""" - if not self.transit: - return {'status': '缺失过境信息'} - - return {'triggered': False} - - def _analyze_wealth_transit(self) -> Dict: - """财富过境触发分析""" - if not self.transit: - return {'status': '缺失过境信息'} - - return {'triggered': False} - - def _analyze_health_transit(self) -> Dict: - """健康过境触发分析""" - if not self.transit: - return {'status': '缺失过境信息'} - - return {'triggered': False} - - # ===== 概率计算 ===== - - def _calculate_marriage_probability(self, static: Dict, dasha: Dict, transit: Dict) -> float: - """计算婚姻概率""" - base_prob = 50.0 - - if static.get('marriage_yoga'): - base_prob += 20 - if static.get('7th_house_afflicted'): - base_prob -= 15 - if static.get('venus_weak'): - base_prob -= 10 - - if dasha.get('activated'): - base_prob += 15 - - if transit.get('triggered'): - base_prob += 10 - - return min(100, max(0, base_prob)) - - def _calculate_career_probability(self, static: Dict, dasha: Dict, transit: Dict) -> float: - """计算职业概率""" - base_prob = 50.0 - - if static.get('10th_house_strong'): - base_prob += 20 - if static.get('career_yoga'): - base_prob += 15 - - if dasha.get('activated'): - base_prob += 15 - - if transit.get('triggered'): - base_prob += 10 - - return min(100, max(0, base_prob)) - - def _calculate_wealth_probability(self, static: Dict, dasha: Dict, transit: Dict) -> float: - """计算财富概率""" - base_prob = 50.0 - - if static.get('2nd_house_strong'): - base_prob += 20 - if static.get('dhana_yoga'): - base_prob += 15 - - if dasha.get('activated'): - base_prob += 15 - - if transit.get('triggered'): - base_prob += 10 - - return min(100, max(0, base_prob)) - - def _calculate_health_probability(self, static: Dict, dasha: Dict, transit: Dict) -> float: - """计算健康概率""" - base_prob = 70.0 # 基础健康概率 - - if static.get('6th_house_afflicted'): - base_prob -= 20 - if static.get('moon_weak'): - base_prob -= 15 - - if dasha.get('activated'): - base_prob -= 10 - - if transit.get('triggered'): - base_prob -= 5 - - return min(100, max(0, base_prob)) - - # ===== 风险评估 ===== - - def _assess_marriage_risk(self, static: Dict) -> RiskLevel: - """评估婚姻风险""" - if static.get('7th_house_afflicted') and static.get('venus_weak'): - return RiskLevel.HIGH - elif static.get('7th_house_afflicted') or static.get('venus_weak'): - return RiskLevel.MEDIUM - else: - return RiskLevel.LOW - - def _assess_career_risk(self, static: Dict) -> RiskLevel: - """评估职业风险""" - if static.get('10th_house_strong'): + + # ======================================================================== + # 第一层：静态星盘分析 + # ======================================================================== + + def _layer1_static(self, evt_type: EventType) -> Dict: + """第一层：全面静态分析""" + result = {'signals': [], 'total_signals': 0, 'details': {}} + target_houses = EVENT_HOUSES.get(evt_type, []) + target_karakas = EVENT_KARAKAS.get(evt_type, []) + + # 1. 宫位强旺检查（主宫+辅助宫） + for h in target_houses: + h_lord = self.house_lords.get(h, '') + if h_lord and self.planet_houses.get(h_lord) in target_houses: + result['signals'].append(f'{h}宫主{h_lord}落入相关宫位({self.planet_houses[h_lord]}宫)') + + # 检查主宫有无吉星落陷/受克 + for pn, ph in self.planet_houses.items(): + if ph == h and pn in target_karakas: + pd = self.chart.get('planets', {}).get(pn, {}) + status = pd.get('status', '') + if status == 'exalted': + result['signals'].append(f'Karaka {pn}在{h}宫擢升') + elif status == 'own_sign': + result['signals'].append(f'Karaka {pn}在{h}宫入庙') + elif status == 'debilitated': + result['signals'].append(f'Karaka {pn}在{h}宫落陷(负面)') + + # 2. 行星聚集分析（使用 congregation_data） + if self.congregation.get('congregations'): + for c in self.congregation['congregations']: + ch = c.get('house', 0) + if ch in target_houses: + strength_label = {'strong': '吉星主导', 'mixed': '吉凶混合', 'malefic_heavy': '凶星主导'} + s = strength_label.get(c['strength'], c['strength']) + result['signals'].append( + f"行星聚集: {c['planets']}聚于{ch}宫({s})，" + f"影响{','.join(c.get('impact', []))}") + + # 3. Vivah Saham（婚姻事件特有） + if evt_type == EventType.MARRIAGE and self.vivah_saham: + vs = self.vivah_saham + if vs.get('saham_lon'): + sahams_house = vs.get('saham_house', 0) + if sahams_house in [7, 1, 5, 9]: + result['signals'].append( + f"Vivah Saham在{sahams_house}宫({vs['saham_sign']}) " + f"{vs['saham_deg_in_sign']:.1f}°，婚姻敏感点高度相关") + # 本命行星与 Saham 的合相 + for cj in vs.get('natal_conjuncts', []): + result['signals'].append( + f"本命{cj['planet']}与Vivah Saham合相({cj['diff_deg']}°)") + + # 4. Ashtakavarga 检查（如果数据可用） + av_data = self.transit.get('ashtakavarga') or self.dasha.get('ashtakavarga') + if av_data and isinstance(av_data, dict): + for h in target_houses[:2]: # 只检查前两个主宫 + sav = av_data.get('bhinnashtakavarga', {}).get(str(h), {}).get('total', 0) + if sav and sav >= 28: + result['signals'].append(f'Ashtakavarga: {h}宫SAV={sav}(≥28，强)') + elif sav and sav <= 18: + result['signals'].append(f'Ashtakavarga: {h}宫SAV={sav}(≤18，弱)') + + result['total_signals'] = len(result['signals']) + return result + + # ======================================================================== + # 第二层：Dasha 激活 + # ======================================================================== + + def _layer2_dasha(self, evt_type: EventType) -> List[str]: + """第二层：Dasha 激活分析（Vimshottari + Chara）""" + signals = [] + target_houses = EVENT_HOUSES.get(evt_type, []) + target_karakas = EVENT_KARAKAS.get(evt_type, []) + + # ── 2a. Vimshottari Dasha ── + if self.dasha: + # 当前 Mahadasha + md_lord = self.dasha.get('current_mahadasha', {}).get('lord', '') + if md_lord: + md_house = self.planet_houses.get(md_lord, 0) + # MD 主星是否关联目标宫位 + if md_house in target_houses: + signals.append(f'当前Vimshottari MD {md_lord}在{md_house}宫(目标宫位)') + + # MD 主星是否就是目标 Karaka + if md_lord in [k for k in target_karakas if not k.endswith('L')]: + signals.append(f'当前MD {md_lord}是{evt_type.value}的Karaka') + + # MD 主星是否是目标宫的宫主星 + for h in target_houses: + if self.house_lords.get(h) == md_lord: + signals.append(f'当前MD {md_lord}是{h}宫主(目标宫)') + + # 当前 Antardasha + ad_lord = self.dasha.get('current_antardasha', {}).get('lord', '') + if ad_lord: + ad_house = self.planet_houses.get(ad_lord, 0) + if ad_house in target_houses: + signals.append(f'当前AD {ad_lord}在{ad_house}宫(目标宫位)') + + for h in target_houses: + if self.house_lords.get(h) == ad_lord: + signals.append(f'当前AD {ad_lord}是{h}宫主(目标宫)') + + # MD+AD 组合信号（高权重） + if md_lord and ad_lord: + if ad_house in target_houses and md_house in target_houses: + signals.append(f'★ MD+AD双激活目标宫位({md_lord}+{ad_lord})') + + # ── 2b. Chara Dasha (Jaimini) ── + if self.chara_dasha: + cd_list = self.chara_dasha.get('dasha_list', []) + if cd_list: + # 当前 Chara Mahadasha + current_cd = cd_list[0] if cd_list else {} + cd_sign = current_cd.get('sign', '') + cd_lord = SIGN_LORDS.get(cd_sign, '') + cd_house = self.planet_houses.get(cd_lord, 0) + + if cd_house in target_houses: + signals.append(f'当前Chara Dasha {cd_sign}({cd_lord})在{cd_house}宫(目标宫)') + + # Chara Antardasha + antardashas = current_cd.get('antardashas', []) + if antardashas: + current_ad = antardashas[0] if antardashas else {} + ad_sign = current_ad.get('sign', '') + ad_lord_name = SIGN_LORDS.get(ad_sign, '') + ad_h = self.planet_houses.get(ad_lord_name, 0) + if ad_h in target_houses: + signals.append(f'Chara AD {ad_sign}({ad_lord_name})在{ad_h}宫(目标宫)') + + return signals + + # ======================================================================== + # 第三层：过境触发 + # ======================================================================== + + def _layer3_transit(self, evt_type: EventType) -> List[str]: + """第三层：过境触发分析""" + signals = [] + target_houses = EVENT_HOUSES.get(evt_type, []) + event_house = target_houses[0] if target_houses else 7 # 主宫 + + # ── 3a. Double Transit PAC + D9 (KN Rao) ── + if self.double_transit: + dt = self.double_transit + dt_list = dt.get('double_transit', []) + for item in dt_list: + layer = item.get('layer', '') + strength = item.get('strength', '') + # 检查是否命中目标宫 + target = item.get('target', '') + # 解析 target 中的宫号 + import re + nums = re.findall(r'(\d+)宫', target) + for n in nums: + if int(n) == event_house: + label = '★' if strength == 'strong' else '◇' + signals.append( + f'{label} Double Transit PAC [{layer}层]: ' + f'Jupiter+Saturn同时PAC到{event_house}宫({strength})') + + # 汇总 + d1_j = dt.get('d1', {}).get('jupiter', {}) + d1_s = dt.get('d1', {}).get('saturn', {}) + if d1_j or d1_s: + j_targets = list(d1_j.keys()) if isinstance(d1_j, dict) else [] + s_targets = list(d1_s.keys()) if isinstance(d1_s, dict) else [] + for t in j_targets: + nums = re.findall(r'(\d+)宫', t) + for n in nums: + if int(n) == event_house: + signals.append(f'Transit Jupiter PAC到{event_house}宫D1层') + for t in s_targets: + nums = re.findall(r'(\d+)宫', t) + for n in nums: + if int(n) == event_house: + signals.append(f'Transit Saturn PAC到{event_house}宫D1层') + + # ── 3b. Transit LL/7L (婚姻特有) ── + if evt_type == EventType.MARRIAGE and self.ll7l: + ll7l = self.ll7l + # P5: Transit LL PAC natal 7L / Transit 7L PAC natal LL + p5_signals = ll7l.get('p5', []) + for s in p5_signals: + signals.append(f'Transit LL/7L P5: {s}') + + # P8: Transit LL in 7H / Transit 7L in Lagna + p8_signals = ll7l.get('p8', []) + for s in p8_signals: + signals.append(f'Transit LL/7L P8: {s}') + + # Parivartana + pariv = ll7l.get('parivartana', []) + if pariv: + for p in pariv: + signals.append(f'Transit LL/7L Parivartana: {p}') + + # ── 3c. Vivah Saham 过境激活（婚姻特有）── + if evt_type == EventType.MARRIAGE and self.vivah_saham: + vs_activations = self.vivah_saham.get('transit_activations', []) + for act in vs_activations: + signals.append(f'Vivah Saham过境激活: {act}') + + # ── 3d. 基础过境分析（从 transit_data 中提取）── + if self.transit: + # Jupiter 过境目标宫 + jup_house = self.transit.get('jupiter', {}).get('house', 0) + if jup_house in target_houses: + signals.append(f'Jupiter过境{jup_house}宫(目标宫)') + + # Saturn 过境目标宫 + sat_house = self.transit.get('saturn', {}).get('house', 0) + if sat_house in target_houses: + signals.append(f'Saturn过境{sat_house}宫(目标宫)') + + # Rahu/Ketu 过境 + rahu_house = self.transit.get('rahu', {}).get('house', 0) + ketu_house = self.transit.get('ketu', {}).get('house', 0) + for rn, rh in [('Rahu', rahu_house), ('Ketu', ketu_house)]: + if rh in target_houses: + signals.append(f'{rn}过境{rh}宫(目标宫)') + + # Sade Sati（Moon 过境相关） + sade_sati = self.transit.get('sade_sati', {}) + if sade_sati and evt_type in [EventType.CAREER, EventType.HEALTH, EventType.MARRIAGE]: + phase = sade_sati.get('phase', '') + if phase: + signals.append(f'Sade Sati {phase}期（影响情绪和决策）') + + return signals + + # ======================================================================== + # 概率、风险、时间窗计算 + # ======================================================================== + + def _calc_probability(self, static: Dict, dasha: List, transit: List) -> float: + """综合概率计算""" + base = 30.0 # 基线概率 + + # 静态信号 + static_score = min(static['total_signals'] * 5, 25) + base += static_score + + # Dasha 信号 + dasha_score = min(len(dasha) * 8, 30) + base += dasha_score + # MD+AD 双激活加分 + if any('★' in s for s in dasha): + base += 10 + + # Transit 信号 + transit_score = min(len(transit) * 6, 25) + base += transit_score + # Double Transit strong 加分 + if any('★' in s for s in transit): + base += 15 + + return min(100, max(0, round(base, 1))) + + def _assess_risk(self, static: Dict, evt_type: EventType) -> RiskLevel: + """风险评估""" + if evt_type == EventType.HEALTH: + # 健康事件特殊处理（高信号 = 高风险） + if static['total_signals'] >= 3: + return RiskLevel.CRITICAL + elif static['total_signals'] >= 2: + return RiskLevel.HIGH return RiskLevel.LOW else: - return RiskLevel.MEDIUM - - def _assess_wealth_risk(self, static: Dict) -> RiskLevel: - """评估财富风险""" - if static.get('2nd_house_strong'): + if static['total_signals'] >= 4: + return RiskLevel.HIGH + elif static['total_signals'] >= 2: + return RiskLevel.MEDIUM return RiskLevel.LOW - else: - return RiskLevel.MEDIUM - - def _assess_health_risk(self, static: Dict) -> RiskLevel: - """评估健康风险""" - if static.get('6th_house_afflicted') and static.get('moon_weak'): - return RiskLevel.HIGH - elif static.get('6th_house_afflicted') or static.get('moon_weak'): - return RiskLevel.MEDIUM - else: - return RiskLevel.LOW - - # ===== 时间预测 ===== - - def _predict_marriage_timing(self, dasha: Dict, transit: Dict) -> Optional[str]: - """预测婚姻时间""" - if dasha.get('activated') and transit.get('triggered'): - return "近期可能" - elif dasha.get('activated'): - return "等待过境触发" - else: - return "需要等待Dasha激活" - - def _predict_career_timing(self, dasha: Dict, transit: Dict) -> Optional[str]: - """预测职业时间""" - if dasha.get('activated'): - return "当前阶段" - else: - return "需要等待Dasha激活" - - def _predict_wealth_timing(self, dasha: Dict, transit: Dict) -> Optional[str]: - """预测财富时间""" - if dasha.get('activated'): - return "当前阶段" - else: - return "需要等待Dasha激活" - - def _predict_health_timing(self, dasha: Dict, transit: Dict) -> Optional[str]: - """预测健康时间""" - if dasha.get('activated'): - return "当前需要注意" - else: - return "总体稳定" - - def generate_report(self) -> str: - """生成预测报告""" - if not self.predictions: - self.predict_all_events() - - lines = [] - lines.append("=" * 60) - lines.append("事件预测报告") - lines.append("=" * 60) + + def _calc_timing_windows(self, evt_type: EventType, dasha: List, transit: List) -> List[Dict]: + """计算时间窗""" + windows = [] + + # 从 Dasha 提取时间信息 + if self.dasha: + md_info = self.dasha.get('current_mahadasha', {}) + if md_info.get('start') and md_info.get('end'): + windows.append({ + 'type': 'Vimshottari MD', + 'period': f"{md_info.get('start', '?')} ~ {md_info.get('end', '?')}", + 'source': md_info.get('lord', '?'), + 'weight': 0.6, + }) + + # 从 Chara Dasha 提取 + if self.chara_dasha: + cd_list = self.chara_dasha.get('dasha_list', []) + if cd_list: + cd = cd_list[0] + start = cd.get('start_year', 0) + end = cd.get('end_year', 0) + if start: + windows.append({ + 'type': 'Chara Dasha', + 'period': f"{start} ~ {end}" if end else f"{start}起", + 'source': cd.get('sign', '?'), + 'weight': 0.4, + }) + + return windows + + def _describe_timing(self, windows, dasha, transit) -> str: + """生成时间描述""" + parts = [] + if windows: + for w in windows: + parts.append(f"{w['type']}({w['period']})") + if dasha: + parts.append(f"{len(dasha)}个Dasha信号") + if transit: + parts.append(f"{len(transit)}个Transit信号") + if not parts: + return "无明确时间窗" + return " | ".join(parts) + + def _gen_recommendations(self, evt_type, static, dasha, transit) -> List[str]: + """生成建议""" + recs = [] + if evt_type == EventType.MARRIAGE: + if len(dasha) >= 2 and len(transit) >= 1: + recs.append("Dasha+Transit双重激活，婚姻事件窗已开，积极把握") + elif len(dasha) >= 1: + recs.append("Dasha激活中，等待Transit触发确认") + if any('落陷' in s for s in static['signals']): + recs.append("相关Karaka落陷，建议补救措施") + elif evt_type == EventType.CAREER: + if len(dasha) >= 2 and len(transit) >= 1: + recs.append("事业突破事件窗已开，抓住机遇") + recs.append("持续关注10宫主Dasha和Jupiter/Saturn过境10宫") + elif evt_type == EventType.HEALTH: + if any('受克' in s or '落陷' in s for s in static['signals']): + recs.append("有健康警示信号，建议定期体检") + recs.append("注意Saturn过境6/8/12宫期间的身体健康") + elif evt_type == EventType.WEALTH: + if any('★' in s for s in transit): + recs.append("Double Transit激活财富宫，投资/商业机会窗口") + if not recs: + recs.append("保持观察，等待多层信号确认") + return recs + + +# ============================================================================ +# 报告生成 +# ============================================================================ + +def generate_prediction_report(model: EventPredictionModel) -> str: + """生成预测报告""" + if not model.predictions: + model.predict_all_events() + + lines = [] + lines.append("=" * 60) + lines.append("事件预测报告 — 三层验证法 v5.0") + lines.append("=" * 60) + lines.append("") + + # 置信度分布 + high_count = sum(1 for p in model.predictions if p.confidence == Confidence.HIGH) + med_count = sum(1 for p in model.predictions if p.confidence == Confidence.MEDIUM) + low_count = sum(1 for p in model.predictions if p.confidence == Confidence.LOW) + + lines.append(f"分析范围: {len(model.predictions)} 类事件") + lines.append(f"置信度: 高={high_count}, 中={med_count}, 低={low_count}") + lines.append("") + + for pred in model.predictions: + conf_icon = {'高': '⭐', '中': '◆', '低': '○'}.get(pred.confidence.value, '○') + lines.append(f"{conf_icon}【{pred.event_type.value}】置信度:{pred.confidence.value} 概率:{pred.probability:.0f}% 风险:{pred.risk_level.value}") + if pred.timing: + lines.append(f" 时间窗: {pred.timing}") + for f in pred.key_factors[:5]: # 最多显示5个 + lines.append(f" - {f}") + for r in pred.recommendations[:3]: + lines.append(f" → {r}") lines.append("") - - for pred in self.predictions: - lines.append(f"【{pred.event_type.value}】") - lines.append(f" 概率：{pred.probability:.1f}%") - lines.append(f" 风险等级：{pred.risk_level.value}") - if pred.timing: - lines.append(f" 时间预测：{pred.timing}") - lines.append(" 关键因素：") - for factor in pred.key_factors: - lines.append(f" - {factor}") - lines.append(" 建议：") - for rec in pred.recommendations: - lines.append(f" - {rec}") - lines.append("") - - return "\n".join(lines) + + return "\n".join(lines) if __name__ == "__main__": # 测试示例 test_chart = { + 'ascendant': {'sign': 'Leo', 'degree': 130.5}, 'planets': { - 'Sun': {'house': 9, 'sign': 'Aries', 'status': 'exalted'}, - 'Moon': {'house': 1, 'sign': 'Leo', 'status': 'own_sign'}, - 'Mars': {'house': 1, 'sign': 'Aries', 'status': 'exalted'}, - 'Mercury': {'house': 9, 'sign': 'Aries', 'status': ''}, - 'Jupiter': {'house': 1, 'sign': 'Sagittarius', 'status': 'own_sign'}, - 'Venus': {'house': 8, 'sign': 'Pisces', 'status': 'exalted'}, - 'Saturn': {'house': 7, 'sign': 'Aquarius', 'status': 'own_sign'}, + 'Sun': {'house': 1, 'sign': 'Leo', 'degree': 132.0, 'status': 'own_sign'}, + 'Moon': {'house': 7, 'sign': 'Aquarius', 'degree': 312.0, 'status': ''}, + 'Mars': {'house': 4, 'sign': 'Scorpio', 'degree': 222.0, 'status': 'own_sign'}, + 'Mercury': {'house': 1, 'sign': 'Leo', 'degree': 145.0, 'status': ''}, + 'Jupiter': {'house': 3, 'sign': 'Libra', 'degree': 192.0, 'status': 'debilitated'}, + 'Venus': {'house': 12, 'sign': 'Cancer', 'degree': 97.0, 'status': ''}, + 'Saturn': {'house': 7, 'sign': 'Aquarius', 'degree': 325.0, 'status': 'own_sign'}, + 'Rahu': {'house': 10, 'sign': 'Taurus', 'degree': 60.0, 'status': ''}, + 'Ketu': {'house': 4, 'sign': 'Scorpio', 'degree': 240.0, 'status': ''}, } } - + model = EventPredictionModel(test_chart) - predictions = model.predict_all_events() - print(model.generate_report()) + print(generate_prediction_report(model)) diff --git a/scripts/jyotish_engine.py b/scripts/jyotish_engine.py index 1248f1f2..967fd85d 100644 --- a/scripts/jyotish_engine.py +++ b/scripts/jyotish_engine.py @@ -596,6 +596,116 @@ def _navamsa_idx(lon): return (el_starts[si % 4] + ni) % 12 +def _calc_planetary_congregation(planets: Dict, asc_idx: int) -> Dict: + """ + 本命盘行星聚集检测（供 cmd_full_reading 调用） + 检测 3+ 行星同宫的聚集效应，返回聚集宫位、行星列表、影响领域 + """ + houses = {} + for pn, pd in planets.items(): + if not isinstance(pd, dict) or 'sign' not in pd: + continue + if pn in ['Rahu', 'Ketu']: + continue # Rahu/Ketu 不计入聚集 + si = SIGNS.index(pd['sign']) if pd['sign'] in SIGNS else 0 + h = ((si - asc_idx) % 12) + 1 + houses.setdefault(h, []).append(pn) + + congregations = [] + for h, plist in houses.items(): + if len(plist) >= 3: + h_sign = SIGNS[(asc_idx + h - 1) % 12] + # 判断影响领域 + impact = _house_theme(h) + # 判断聚集力量（有无吉星/凶星） + benefics = [p for p in plist if p in ['Jupiter', 'Venus', 'Mercury', 'Moon']] + malefics = [p for p in plist if p in ['Saturn', 'Mars', 'Sun', 'Rahu']] + strength = 'strong' if len(benefics) > len(malefics) else 'mixed' + if len(malefics) >= 3: + strength = 'malefic_heavy' + congregations.append({ + 'house': h, + 'sign': h_sign, + 'planets': plist, + 'count': len(plist), + 'benefics': benefics, + 'malefics': malefics, + 'strength': strength, + 'impact': impact, + 'description': f'{",".join(plist)} 聚集于{h}宫({h_sign})', + }) + + return { + 'congregations': congregations, + 'total': len(congregations), + 'note': '3+ 行星同宫为显著聚集，影响该宫主题领域', + } + + +def _house_theme(house: int) -> List[str]: + """返回宫位影响领域""" + themes = { + 1: ['自我', '健康', '性格'], + 2: ['财富', '家庭', '言语'], + 3: ['沟通', '旅行', '兄弟'], + 4: ['母亲', '房产', '情感'], + 5: ['子女', '投资', '创意'], + 6: ['疾病', '敌人', '债务'], + 7: ['婚姻', '合作', '伴侣'], + 8: ['转型', '意外', '遗产'], + 9: ['命运', '父亲', '灵性'], + 10: ['事业', '声望', '成就'], + 11: ['收益', '社交', '愿望'], + 12: ['损失', '外迁', '解脱'], + } + return themes.get(house, ['未知']) + + +def _calc_vivah_saham(planets: Dict, asc_deg: float) -> Dict: + """ + 计算本命 Vivah Saham 婚姻敏感点（供 cmd_full_reading 调用） + 公式: Saham = (Venus_lon - Saturn_lon + Asc_deg) % 360 + """ + venus_lon = planets.get('Venus', {}).get('degree', 0) + saturn_lon = planets.get('Saturn', {}).get('degree', 0) + if venus_lon == 0 or saturn_lon == 0: + return {'error': '缺少金星或土星数据', 'saham': None} + + sahams_lon = (venus_lon - saturn_lon + asc_deg) % 360 + sahams_sign = SIGNS[int(sahams_lon / 30) % 12] + sahams_deg_in_sign = sahams_lon % 30 + sahams_si = int(sahams_lon / 30) % 12 + + # 检查哪些本命行星与 Saham 同宫/合相 + conjuncts = [] + for pn, pd in planets.items(): + if pn in ['Rahu', 'Ketu']: + continue + if not isinstance(pd, dict) or 'degree' not in pd: + continue + p_lon = pd['degree'] + diff = abs(p_lon - sahams_lon) % 360 + if diff > 180: + diff = 360 - diff + if diff <= 5: + conjuncts.append({'planet': pn, 'diff_deg': round(diff, 2)}) + + # 从 Saham 位置反推婚姻相关宫位 + asc_si = int(asc_deg / 30) % 12 + sahams_house = ((sahams_si - asc_si) % 12) + 1 + + return { + 'saham_lon': round(sahams_lon, 4), + 'saham_sign': sahams_sign, + 'saham_deg_in_sign': round(sahams_deg_in_sign, 2), + 'saham_house': sahams_house, + 'formula': f'Venus({venus_lon:.2f}°) - Saturn({saturn_lon:.2f}°) + Asc({asc_deg:.2f}°)', + 'natal_conjuncts': conjuncts, + 'marriage_relevance': 'high' if sahams_house in [7, 1, 5, 9] else 'moderate', + 'note': 'Vivah Saham 是度数级婚姻敏感点，Transit 木星/土星过境此点时触发婚姻事件窗', + } + + def _check_pac(planet_name, planet_lon, target_lon, asc_idx): """PAC检查: Position(同宫)/Aspect(相位)/Conjunction(合相<=10度)""" results = [] @@ -1971,13 +2081,15 @@ def cmd_full_reading(args): # ── Step 6: Jaimini系统 ── try: - from jaimini import calc_chara_karaka_7, calc_chara_karaka_8, calc_chara_dasha, calc_karakamsha + from jaimini import calc_chara_karaka_7, calc_chara_karaka_8, calc_chara_dasha, calc_karakamsha, calc_chara_dasha_with_antardasha from varga import calc_varga jaimini_result = {} jaimini_result['chara_karaka_7'] = calc_chara_karaka_7(planet_degs) jaimini_result['chara_karaka_8'] = calc_chara_karaka_8(planet_degs) - jaimini_result['chara_dasha'] = calc_chara_dasha(asc_idx, planet_lons, args.year, args.month) + # 使用带 Antardasha 子周期的 Chara Dasha（v4.3.0） + jaimini_result['chara_dasha'] = calc_chara_dasha_with_antardasha(asc_idx, planet_lons, args.year, args.month) + jaimini_result['has_antardasha'] = True # Karakamsha（用AK灵魂星，非DK配偶星） # ⚠️ 2026-05-03修正：此前错误使用DK，现已修正为AK @@ -2079,6 +2191,20 @@ def cmd_full_reading(args): except Exception as e: report['errors'].append(f"actionable-context: {e}") + # ── Step 15: Planetary Congregation 行星聚集 (v4.3.0) ── + try: + congregation = _calc_planetary_congregation(planets, asc_idx) + report['modules']['congregation'] = congregation + except Exception as e: + report['errors'].append(f"congregation: {e}") + + # ── Step 16: Vivah Saham 婚姻敏感点 (v4.3.0) ── + try: + sahams = _calc_vivah_saham(planets, asc_deg) + report['modules']['vivah_saham'] = sahams + except Exception as e: + report['errors'].append(f"vivah-saham: {e}") + # ── 汇总 ── elapsed = round(time.time() - t0, 2) module_count = len(report['modules']) @@ -2088,7 +2214,7 @@ def cmd_full_reading(args): 'modules_computed': module_count, 'errors': error_count, 'status': 'complete' if error_count == 0 else f'{error_count} errors', - 'next_step': 'AI可直接基于此数据执行阶段二→三→四→五。⭐ v4.1.0: modules.actionable_context 已就绪，Transit分析时必须输出Actionable Output（时间段+行动类型+置信度），动态预测必须先检索案例。', + 'next_step': 'AI可直接基于此数据执行阶段二→三→四→五。⭐ v4.3.0: modules.congregation + modules.vivah_saham + antardasha已就绪，全链路分析包含行星聚集、婚姻敏感点、Chara子周期。Transit分析时必须输出Actionable Output（时间段+行动类型+置信度），动态预测必须先检索案例。', } return report