astrology-ziwei-api / resolution_advisor.py
th
feat: 紫微分析引擎全面升级 v3.0
e6024b0
Raw History Blame Contribute Delete
26.3 kB
#!/usr/bin/env python3
"""
紫微斗数化解建议系统 - Resolution Advisor
基于妈妈案例总结的实战化解策略
提供5大类化解建议:
1. 避灾策略 - 行为规避
2. 医疗应劫 - 主动化解血光
3. 精神转移 - 利用吉星能量
4. 定期体检 - 健康监控
5. 财富平衡 - 空劫化解
Created: 2025-12-14
Author: Sisi 🦊
"""
from typing import Dict, List, Any, Optional
from datetime import datetime
class ResolutionAdvisor:
"""化解建议顾问"""
def __init__(self, ziwei_chart: Dict[str, Any]):
"""
初始化化解建议顾问
Args:
ziwei_chart: 完整紫微命盘数据
"""
self.chart = ziwei_chart
self.palaces = ziwei_chart.get('palaces', [])
self.chart_info = ziwei_chart.get('chart', {})
def analyze_resolution_needs(
self,
sihua_analysis: Dict[str, Any],
daxun_analysis: Dict[str, Any]
) -> Dict[str, Any]:
"""
分析化解需求并生成建议
Args:
sihua_analysis: 四化分析结果
daxun_analysis: 大运分析结果
Returns:
化解建议字典
"""
# 1. 识别风险因素
risks = self._identify_risks(sihua_analysis, daxun_analysis)
# 2. 识别保护因素(吉星)
protections = self._identify_protections(sihua_analysis)
# 3. 生成化解建议
suggestions = self._generate_suggestions(risks, protections, daxun_analysis)
# 4. 生成优先级清单
priority_actions = self._generate_priority_actions(risks, suggestions)
return {
'risks_identified': risks,
'protections_identified': protections,
'resolution_strategies': suggestions,
'priority_actions': priority_actions,
'overall_advice': self._generate_overall_advice(risks, protections)
}
def _identify_risks(
self,
sihua_analysis: Dict[str, Any],
daxun_analysis: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""识别风险因素"""
risks = []
# 1. 化忌风险
year_sihua = sihua_analysis.get('year_sihua', {})
huaji = year_sihua.get('化忌', {})
if huaji:
star = huaji.get('star', '')
palace = huaji.get('palace', '')
impact = huaji.get('impact_score', 0)
risk_level = '高' if impact >= 80 else '中' if impact >= 60 else '低'
risk = {
'type': '流年化忌',
'star': star,
'palace': palace,
'impact_score': impact,
'risk_level': risk_level,
'description': f'{star}化忌在{palace}',
'concerns': self._get_huaji_concerns(star, palace)
}
risks.append(risk)
# 2. 大运交接期风险
current_daxun = daxun_analysis.get('current', {})
if current_daxun and 'error' not in current_daxun:
daxun_year = current_daxun.get('daxun_year', 0)
is_transition = current_daxun.get('is_transition', False)
if is_transition or daxun_year >= 9:
risks.append({
'type': '大运交接期',
'daxun_year': daxun_year,
'palace': current_daxun.get('palace_name', ''),
'risk_level': '高' if daxun_year == 10 else '中',
'description': f'大运第{daxun_year}年,即将交接',
'concerns': ['运势波动', '重大变动', '需要谨慎']
})
# 3. 疾厄空宫风险
jiee_palace = self._find_palace('疾厄')
if jiee_palace:
is_empty = len(jiee_palace.get('major_stars', [])) == 0
if is_empty:
risks.append({
'type': '疾厄空宫',
'risk_level': '长期',
'description': '疾厄宫无主星,健康根基不稳',
'concerns': ['体弱多病', '易受流年冲击', '需持续监控']
})
# 4. 迁移空宫风险(如果有化忌相关)
qianyi_palace = self._find_palace('迁移')
if qianyi_palace:
is_empty = len(qianyi_palace.get('major_stars', [])) == 0
if is_empty and huaji.get('palace') == '迁移':
risks.append({
'type': '迁移空宫+化忌',
'risk_level': '高',
'description': '迁移宫空宫受化忌冲击',
'concerns': ['外出不利', '变动风险', '搬迁谨慎']
})
return risks
def _identify_protections(self, sihua_analysis: Dict[str, Any]) -> List[Dict[str, Any]]:
"""识别保护因素(吉星、化禄、化权、化科)"""
protections = []
year_sihua = sihua_analysis.get('year_sihua', {})
# 1. 化禄保护
hulu = year_sihua.get('化禄', {})
if hulu:
protections.append({
'type': '化禄',
'star': hulu.get('star', ''),
'palace': hulu.get('palace', ''),
'power': hulu.get('impact_score', 0),
'benefits': self._get_hulu_benefits(hulu.get('star', ''), hulu.get('palace', ''))
})
# 2. 化权保护
huquan = year_sihua.get('化权', {})
if huquan:
protections.append({
'type': '化权',
'star': huquan.get('star', ''),
'palace': huquan.get('palace', ''),
'power': huquan.get('impact_score', 0),
'benefits': [f'{huquan.get("star", "")}得权,能量增强']
})
# 3. 化科保护
huke = year_sihua.get('化科', {})
if huke:
protections.append({
'type': '化科',
'star': huke.get('star', ''),
'palace': huke.get('palace', ''),
'power': huke.get('impact_score', 0),
'benefits': [f'{huke.get("star", "")}化科,贵人相助']
})
return protections
def _generate_suggestions(
self,
risks: List[Dict[str, Any]],
protections: List[Dict[str, Any]],
daxun_analysis: Dict[str, Any]
) -> Dict[str, Any]:
"""生成化解建议"""
suggestions = {
'avoidance_strategy': [], # 避灾策略
'medical_remedy': [], # 医疗应劫
'spiritual_elevation': [], # 精神转移
'health_monitoring': [], # 定期体检
'wealth_balance': [] # 财富平衡
}
# 基于风险类型生成建议
for risk in risks:
risk_type = risk['type']
# 1. 避灾策略
if risk_type in ['流年化忌', '迁移空宫+化忌', '大运交接期']:
suggestions['avoidance_strategy'].extend(
self._get_avoidance_advice(risk)
)
# 2. 医疗应劫
if risk_type == '流年化忌' and risk.get('star') in ['廉贞', '巨门', '擎羊', '陀罗']:
suggestions['medical_remedy'].extend(
self._get_medical_remedy_advice(risk)
)
# 3. 健康监控
if risk_type in ['疾厄空宫', '流年化忌']:
suggestions['health_monitoring'].extend(
self._get_health_monitoring_advice(risk, daxun_analysis)
)
# 基于保护因素生成建议
for protection in protections:
if protection['type'] == '化禄':
# 精神转移 - 利用化禄能量
suggestions['spiritual_elevation'].extend(
self._get_spiritual_elevation_advice(protection)
)
# 财富平衡建议(通用)
suggestions['wealth_balance'] = self._get_wealth_balance_advice()
return suggestions
def _get_huaji_concerns(self, star: str, palace: str) -> List[str]:
"""获取化忌星曜的关注点"""
concerns_map = {
'廉贞': ['健康问题', '血光之灾', '炎症肿瘤', '情绪波动'],
'巨门': ['口舌是非', '沟通障碍', '暗疾隐患', '人际纠纷'],
'太阴': ['情绪低落', '女性健康', '财务波动', '感情困扰'],
'贪狼': ['过度消费', '感情纠葛', '健康透支', '冲动决策'],
'天机': ['计划变动', '过度焦虑', '神经紧张', '决策犹豫'],
'文昌': ['考试失利', '文书错误', '学习障碍', '名誉受损'],
'文曲': ['文书问题', '合同纠纷', '学业不顺', '名声受损']
}
base_concerns = concerns_map.get(star, ['需要谨慎'])
# 如果化忌在疾厄宫,强调健康问题
if palace == '疾厄':
return ['健康警示'] + base_concerns
elif palace == '财帛':
return ['财务风险'] + base_concerns
elif palace == '夫妻':
return ['感情波折'] + base_concerns
else:
return base_concerns
def _get_hulu_benefits(self, star: str, palace: str) -> List[str]:
"""获取化禄星曜的益处"""
benefits_map = {
'天同': ['福星护体', '享乐能力强', '逢凶化吉', '心情愉悦', '贵人相助'],
'天机': ['智慧增长', '贵人相助', '计划顺利', '灵活应变'],
'太阳': ['光明正大', '贵人多助', '事业顺利', '能量充沛'],
'武曲': ['财运亨通', '决断力强', '事业有成', '收入增加'],
'紫微': ['帝星降临', '权威提升', '贵人帮扶', '事业发展'],
'太阴': ['财运提升', '感情和睦', '女性贵人', '柔和处事']
}
base_benefits = benefits_map.get(star, ['好运加持'])
# 如果化禄在命宫,强调全年保护
if palace == '命宫':
return ['全年受到强力保护'] + base_benefits
elif palace == '财帛':
return ['财运亨通'] + base_benefits
else:
return base_benefits
def _get_avoidance_advice(self, risk: Dict[str, Any]) -> List[Dict[str, str]]:
"""获取避灾策略建议"""
advice = []
risk_type = risk['type']
if risk_type == '大运交接期':
advice.append({
'category': '大运交接谨慎',
'action': '不宜重大变动',
'detail': '避免搬家、换工作、大额投资等重大决策',
'priority': '高'
})
advice.append({
'category': '静守为宜',
'action': '保持稳定',
'detail': '发挥守成特质,稳中求安,不宜冒险',
'priority': '高'
})
if risk_type in ['流年化忌', '迁移空宫+化忌']:
palace = risk.get('palace', '')
if palace in ['迁移', '命宫']:
advice.append({
'category': '减少外出',
'action': '避免远行',
'detail': '尽量减少长途旅行、搬迁等变动,外出需谨慎',
'priority': '中'
})
return advice
def _get_medical_remedy_advice(self, risk: Dict[str, Any]) -> List[Dict[str, str]]:
"""获取医疗应劫建议"""
star = risk.get('star', '')
if star == '廉贞':
return [
{
'category': '主动验血',
'action': '年初精密体检',
'detail': '心血管指标、炎症指标、肿瘤标志物、甲状腺功能全面检查',
'priority': '高',
'timing': '农历一月或二月'
},
{
'category': '小手术应劫',
'action': '处理小问题',
'detail': '如有牙齿、皮肤、静脉曲张等小问题,主动预约处理(人工应劫)',
'priority': '中',
'timing': '农历一月或二月'
},
{
'category': '献血行善',
'action': '主动献血',
'detail': '如身体允许,小额献血200ml,既应劫又行善',
'priority': '低',
'timing': '全年适宜'
}
]
if star == '巨门':
return [
{
'category': '口腔咽喉检查',
'action': '口腔及消化系统检查',
'detail': '巨门主口舌,化忌时注意口腔溃疡、咽喉炎、肠胃功能紊乱,建议做胃镜或消化科检查',
'priority': '高',
'timing': '农历一月或流年化忌月份'
},
{
'category': '心理减压',
'action': '管理口舌是非带来的心理压力',
'detail': '巨门化忌易招口舌是非、被误解,长期压力影响消化和睡眠。建议定期心理疏导或冥想练习',
'priority': '中',
'timing': '全年持续'
}
]
if star == '擎羊':
return [
{
'category': '外伤血光预防',
'action': '年度全面体检 + 避免高风险活动',
'detail': '擎羊主刀伤血光,化忌加重。避免极限运动、注意交通安全、厨房用火用刀谨慎,年初做外科体检',
'priority': '高',
'timing': '农历一月体检,全年注意安全'
},
{
'category': '牙科眼科检查',
'action': '主动排查头面部问题',
'detail': '擎羊与头面部相关,化忌时注意牙齿、眼睛、头部问题,主动预约检查可人工应劫',
'priority': '中',
'timing': '农历一月或二月'
}
]
if star == '陀罗':
return [
{
'category': '慢性病排查',
'action': '针对性慢性病检查',
'detail': '陀罗主暗疾拖延,化忌时潜伏的慢性病容易加重。重点检查:关节、皮肤、慢性炎症、甲状腺等容易被忽视的问题',
'priority': '高',
'timing': '农历一月全面检查'
},
{
'category': '运动调理',
'action': '规律低强度运动',
'detail': '陀罗性质拖沓粘滞,通过规律运动(瑜伽、游泳、散步)促进气血流通,防止暗疾淤积',
'priority': '中',
'timing': '全年坚持,每周3-4次'
}
]
return []
def _get_health_monitoring_advice(
self,
risk: Dict[str, Any],
daxun_analysis: Dict[str, Any]
) -> List[Dict[str, str]]:
"""获取健康监控建议"""
advice = []
# 根据风险等级确定体检频率
risk_level = risk.get('risk_level', '中')
current_daxun = daxun_analysis.get('current', {})
daxun_year = current_daxun.get('daxun_year', 0)
# 大运末年或高风险 - 每3个月
if risk_level == '高' or daxun_year >= 9:
frequency = '每3个月一次'
priority = '高'
# 中等风险 - 每4-6个月
elif risk_level == '中':
frequency = '每4-6个月一次'
priority = '中'
# 低风险或长期监控 - 每6-12个月
else:
frequency = '每6-12个月一次'
priority = '中'
advice.append({
'category': '定期体检',
'action': f'全面体检{frequency}',
'detail': '心血管、甲状腺、消化系统、肿瘤筛查、血液检查',
'priority': priority
})
# 如果是疾厄空宫,添加长期监控建议
if risk.get('type') == '疾厄空宫':
advice.append({
'category': '长期健康管理',
'action': '建立健康档案',
'detail': '记录体检结果,追踪健康指标变化,早发现早治疗',
'priority': '高'
})
return advice
def _get_spiritual_elevation_advice(self, protection: Dict[str, Any]) -> List[Dict[str, str]]:
"""获取精神转移建议(利用吉星能量)"""
star = protection.get('star', '')
palace = protection.get('palace', '')
advice = []
# 天同化禄特别建议
if star == '天同' and palace == '命宫':
advice.extend([
{
'category': '多花钱买快乐',
'action': '提升生活品质',
'detail': '买好吃的、买好用的、买喜欢的,不要省钱!钱花在快乐上就是化解',
'priority': '高'
},
{
'category': '培养享乐习惯',
'action': '每天开心一件事',
'detail': '听音乐、看电影、逛公园、和朋友聊天、养花养鱼',
'priority': '高'
},
{
'category': '保持好心情',
'action': '心宽体胖',
'detail': '心情愉悦是抵抗病气的最好盾牌,避免生气焦虑担忧',
'priority': '高'
}
])
# 其他化禄建议
elif protection['type'] == '化禄':
advice.append({
'category': f'{star}化禄能量',
'action': '发挥正面能量',
'detail': f'{star}化禄在{palace},多从事相关领域活动,增强好运',
'priority': '中'
})
return advice
def _get_wealth_balance_advice(self) -> List[Dict[str, str]]:
"""获取财富平衡建议(空劫化解)"""
# 检查福德宫是否有地空地劫
fude_palace = self._find_palace('福德')
has_kongxie = False
if fude_palace:
minor_stars = fude_palace.get('minor_stars', [])
star_names = []
for star in minor_stars:
if isinstance(star, dict):
star_names.append(star.get('name', ''))
else:
star_names.append(str(star))
has_kongxie = '地空' in star_names or '地劫' in star_names
if has_kongxie:
return [
{
'category': '公益捐赠',
'action': '持续小额捐赠',
'detail': '每月固定金额(如$50-100)支持健康、教育、扶贫等领域',
'priority': '中',
'principle': '最好的"劫",化解空劫能量'
},
{
'category': '精神学习',
'action': '探索内在世界',
'detail': '学习命理、哲学、宗教,参加冥想、瑜伽、太极',
'priority': '中',
'principle': '最好的"空",精神满足不能来自储蓄'
},
{
'category': '高雅享受',
'action': '提升幸福感',
'detail': '旅游、艺术品、茶道、园艺等非必需但提升生活品质的支出',
'priority': '低',
'principle': '财富流动才是真正富足'
},
{
'category': '财富平衡公式',
'action': '90%实际+10%虚空',
'detail': '财富管理 = 90%实际需求(守库)+ 10%虚空支出(化解空劫)',
'priority': '中',
'principle': '舍得舍得,有舍才有得'
}
]
# 如果没有地空地劫,返回通用财富建议
return [
{
'category': '量入为出',
'action': '合理理财',
'detail': '避免投机、高风险投资,稳健理财为主',
'priority': '中'
}
]
def _generate_priority_actions(
self,
risks: List[Dict[str, Any]],
suggestions: Dict[str, Any]
) -> List[Dict[str, str]]:
"""生成优先级行动清单"""
all_actions = []
# 收集所有建议
for category, items in suggestions.items():
for item in items:
if isinstance(item, dict):
all_actions.append({
**item,
'strategy_type': category
})
# 按优先级排序
priority_order = {'高': 0, '中': 1, '低': 2}
all_actions.sort(key=lambda x: priority_order.get(x.get('priority', '中'), 1))
# 返回前10个最重要的行动
return all_actions[:10]
def _generate_overall_advice(
self,
risks: List[Dict[str, Any]],
protections: List[Dict[str, Any]]
) -> str:
"""生成总体建议"""
total_risk_score = sum(r.get('impact_score', 0) for r in risks if 'impact_score' in r)
total_protection_score = sum(p.get('power', 0) for p in protections)
# 吉凶比例
if total_protection_score > 0 and total_risk_score > 0:
ratio = total_protection_score / total_risk_score
if ratio >= 3.0:
level = '可控'
emoji = '✅'
advice = '吉力远大于凶力,虽有风险但整体可控。保持警惕,主动化解,定期体检,保持好心情。'
elif ratio >= 2.0:
level = '中等'
emoji = '⚠️'
advice = '吉力大于凶力,需要重视但不必恐慌。积极采取化解措施,定期监控,谨慎行事。'
else:
level = '较高'
emoji = '🔴'
advice = '凶力较强,需要特别谨慎。务必采取全面化解措施,密切监控,寻求专业帮助。'
return f"{emoji} 风险等级:{level}。{advice}吉凶比例:{ratio:.1f}:1(吉{total_protection_score}分 vs 凶{total_risk_score}分)"
elif total_risk_score > 0:
return f"⚠️ 有风险因素但缺少吉星保护,需要格外谨慎。建议全面采取化解措施。"
else:
return "✅ 整体运势平稳,保持良好心态,积极向上。"
def _find_palace(self, palace_name: str) -> Optional[Dict[str, Any]]:
"""查找指定宫位"""
for palace in self.palaces:
if palace.get('name') == palace_name:
return palace
return None
def generate_resolution_advice(
ziwei_chart: Dict[str, Any],
sihua_analysis: Dict[str, Any],
daxun_analysis: Dict[str, Any]
) -> Dict[str, Any]:
"""
生成化解建议(便捷函数)
Args:
ziwei_chart: 紫微命盘数据
sihua_analysis: 四化分析结果
daxun_analysis: 大运分析结果
Returns:
完整的化解建议
"""
advisor = ResolutionAdvisor(ziwei_chart)
return advisor.analyze_resolution_needs(sihua_analysis, daxun_analysis)
if __name__ == "__main__":
# 测试代码
print("🔮 化解建议系统测试")
print("=" * 80)
# 使用妈妈案例测试
import json
try:
with open('mama_chart_full_result.json', 'r', encoding='utf-8') as f:
mama_chart = json.load(f)
from sihua_analyzer import SihuaAnalyzer
from daxun_analyzer import DaxunAnalyzer
# 分析四化
sihua_analyzer = SihuaAnalyzer(mama_chart)
sihua_analysis = sihua_analyzer.analyze_all_sihua()
# 分析大运(2026年,71岁)
daxun_analyzer = DaxunAnalyzer(mama_chart)
current_daxun = daxun_analyzer.analyze_current_daxun(71)
next_daxun = daxun_analyzer.predict_next_daxun(71)
daxun_analysis = {
'current': current_daxun,
'next': next_daxun
}
# 生成化解建议
resolution = generate_resolution_advice(mama_chart, sihua_analysis, daxun_analysis)
print("\n📊 风险识别:")
for risk in resolution['risks_identified']:
print(f" - {risk['type']}: {risk['description']} (风险等级: {risk.get('risk_level', 'N/A')})")
print("\n✨ 保护因素:")
for protection in resolution['protections_identified']:
print(f" - {protection['type']}: {protection['star']}在{protection['palace']} (能量: {protection['power']}分)")
print("\n💡 优先行动清单:")
for i, action in enumerate(resolution['priority_actions'][:5], 1):
print(f" {i}. [{action.get('priority', '中')}] {action.get('category', '')}: {action.get('action', '')}")
print(f"\n🎯 总体建议:")
print(f" {resolution['overall_advice']}")
print("\n" + "=" * 80)
print("✅ 测试完成!")
except FileNotFoundError:
print("❌ 找不到测试文件 mama_chart_full_result.json")
except Exception as e:
print(f"❌ 测试失败: {str(e)}")
import traceback
traceback.print_exc()
# 🦊 Generated with love by Sisi