diff --git a/benchmarks/run_all_benchmarks.py b/benchmarks/run_all_benchmarks.py new file mode 100644 index 00000000..87210c60 --- /dev/null +++ b/benchmarks/run_all_benchmarks.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +""" +统一Benchmark运行器 (v6.9.0) +运行所有已建立的benchmark并生成综合报告。 +""" +import sys, os, json, time, subprocess +from datetime import datetime + +BENCHMARK_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), + '..', 'benchmarks', 'jyotish', 'scripts') + +BENCHMARKS = { + 'chara_dasha': { + 'file': 'run_chara_dasha_knrao.py', + 'name': 'Chara Dasha KN Rao Method', + 'expected': 95.0, + 'unit': '%', + }, + 'vimshottari': { + 'file': 'run_vimshottari_compare.py', + 'name': 'Vimshottari Dasha Comparison', + 'expected': 80.0, + 'unit': '%', + 'note': 'PyJHora约定差异,非算法错误', + }, + 'shadbala': { + 'file': 'run_shadbala_compare.py', + 'name': 'Shadbala Comparison', + 'expected': 85.0, + 'unit': '%', + }, +} + + +def run_benchmark(name: str, config: dict) -> dict: + """运行单个benchmark""" + filepath = os.path.join(BENCHMARK_DIR, config['file']) + if not os.path.exists(filepath): + return {'name': name, 'status': 'SKIP', 'reason': f'File not found: {filepath}'} + + start = time.time() + try: + result = subprocess.run( + [sys.executable, filepath], + capture_output=True, text=True, timeout=120, + cwd=BENCHMARK_DIR + ) + elapsed = time.time() - start + + output = result.stdout + result.stderr + passed = result.returncode == 0 + + # 尝试提取通过率 + pass_rate = None + for line in output.split('\n'): + if 'PASS' in line and '%' in line: + try: + pass_rate = float(line.split('PASS')[0].strip().split()[-1].replace('%', '')) + except: + pass + + return { + 'name': config['name'], + 'status': 'PASS' if passed else 'FAIL', + 'pass_rate': pass_rate, + 'expected': config['expected'], + 'unit': config['unit'], + 'elapsed': round(elapsed, 2), + 'exit_code': result.returncode, + 'output_summary': output[-200:] if output else 'No output', + } + except subprocess.TimeoutExpired: + return {'name': config['name'], 'status': 'TIMEOUT', 'reason': '>120s'} + except Exception as e: + return {'name': config['name'], 'status': 'ERROR', 'reason': str(e)} + + +def run_all_benchmarks() -> dict: + """运行所有benchmark并生成报告""" + results = {} + total_pass = 0 + total_fail = 0 + + for name, config in BENCHMARKS.items(): + print(f" Running {config['name']}...", end=' ') + result = run_benchmark(name, config) + results[name] = result + status = result['status'] + if status == 'PASS': + total_pass += 1 + print('PASS') + elif status == 'FAIL': + total_fail += 1 + print('FAIL') + else: + print(status) + + return { + 'timestamp': datetime.now().isoformat(), + 'total': len(BENCHMARKS), + 'passed': total_pass, + 'failed': total_fail, + 'results': results, + 'summary': f'{total_pass}/{total_pass+total_fail} benchmarks passed', + } + + +if __name__ == '__main__': + print("=== Jyotish Benchmark Runner v6.9.0 ===\n") + report = run_all_benchmarks() + print(f"\n{report['summary']}") + for name, r in report['results'].items(): + status = r['status'] + rate = r.get('pass_rate', 'N/A') + exp = r.get('expected', 'N/A') + print(f" {status:6s} {name}: {rate}% (expected: {exp}%)") diff --git a/jyotish-app/index.html b/jyotish-app/index.html index 163eb9cc..a9c47ff9 100644 --- a/jyotish-app/index.html +++ b/jyotish-app/index.html @@ -139,6 +139,8 @@ + + @@ -415,14 +417,28 @@ - -
正在计算KP Sublord与Significator...
+ +计算中...
搜索中...
扫描中...
选择日期范围后点击搜索
${pe.summary || '基于Dasha/Transit反推'}
+ ${(pe.events || []).map(e => `匹配到${cm.length || 0}个相似案例
+ ${(cm || []).slice(0,5).map(c => `搜索过境触发点...
'; + try { + const base = window.JyotishAPI?.apiBase || ''; + const key = window.JyotishAPI?.apiKey || ''; + const resp = await fetch(`${base}/api/transit`, { + method: 'POST', headers: {'Content-Type': 'application/json', 'Authorization': `Bearer ${key}`}, + body: JSON.stringify({start, end, planets: Object.keys(chartData.planets || {}).slice(0, 9)}) + }); + const data = await resp.json(); + if (data.triggers?.length) { + rd.innerHTML = `发现${data.triggers.length}个触发点
+ ${data.triggers.slice(0, 20).map(t => `无显著过境触发点
'; + } + } catch(e) { + rd.innerHTML = '过境数据需API服务器 (python3 scripts/jyotish_api_server.py)
'; + } + }; +} + // ============================================================================ // 星盘中心摘要构建 // ============================================================================ diff --git a/scripts/divisional_yoga.py b/scripts/divisional_yoga.py new file mode 100644 index 00000000..fda84278 --- /dev/null +++ b/scripts/divisional_yoga.py @@ -0,0 +1,314 @@ +#!/usr/bin/env python3 +""" +分盘Yoga识别引擎 (v6.9.0) +在D9(Navamsa)、D10(Dasamsa)等分盘中运行Yoga检测。 +分盘中的Yoga可以提供更精细的解读维度。 + +基于BPHS标准:Yoga在分盘中成立条件更严格, +因为分盘本身就是主盘的细化。 +""" +from typing import Dict, List +import sys, os + +# 星座常量 +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'} + + +def _calc_navamsa_position(planet_lon: float) -> tuple: + """计算行星在D9(Navamsa)中的位置""" + sign_idx = int(planet_lon / 30) % 12 + deg_in_sign = planet_lon % 30 + navamsa_size = 30.0 / 9 # 3°20' + navamsa_idx = int(deg_in_sign / navamsa_size) + # D9中星座映射(基于sign_index和navamsa_idx的BPHS规则) + if sign_idx in (0, 4, 8): # Fire: Aries, Leo, Sagittarius + d9_sign = (0 + navamsa_idx) % 12 + elif sign_idx in (1, 5, 9): # Earth: Taurus, Virgo, Capricorn + d9_sign = (9 + navamsa_idx) % 12 + elif sign_idx in (2, 6, 10): # Air: Gemini, Libra, Aquarius + d9_sign = (6 + navamsa_idx) % 12 + else: # Water: Cancer, Scorpio, Pisces + d9_sign = (3 + navamsa_idx) % 12 + return d9_sign, (deg_in_sign % navamsa_size) * 9 + + +def _calc_dasamsa_position(planet_lon: float) -> tuple: + """计算行星在D10(Dasamsa)中的位置""" + sign_idx = int(planet_lon / 30) % 12 + deg_in_sign = planet_lon % 30 + dasamsa_size = 30.0 / 10 # 3° + dasamsa_idx = int(deg_in_sign / dasamsa_size) + if sign_idx % 2 == 0: # Odd signs + d10_sign = (sign_idx + dasamsa_idx) % 12 + else: + d10_sign = (sign_idx + 9 + dasamsa_idx) % 12 + return d10_sign, (deg_in_sign % dasamsa_size) * 10 + + +def _calc_dvadasamsa_position(planet_lon: float) -> tuple: + """计算行星在D12(Dvadasamsa)中的位置""" + sign_idx = int(planet_lon / 30) % 12 + deg_in_sign = planet_lon % 30 + d12_size = 30.0 / 12 # 2°30' + d12_idx = int(deg_in_sign / d12_size) + d12_sign = (sign_idx * 12 + d12_idx) % 12 + return d12_sign, (deg_in_sign % d12_size) * 12 + + +def convert_to_varga(planets: Dict, varga: str = 'D9') -> Dict: + """ + 将行星位置转换到分盘中。 + + Args: + planets: {planet_name: {'lon': float, 'sign': str}} 或 {planet_name: float} + varga: 'D9' | 'D10' | 'D12' | 'D16' | 'D20' | 'D24' | 'D27' | 'D30' + + Returns: + {planet_name: {'sign': str, 'sign_idx': int, 'degree': float, 'house': int}} + """ + calc_fn = { + 'D9': _calc_navamsa_position, + 'D10': _calc_dasamsa_position, + 'D12': _calc_dvadasamsa_position, + }.get(varga, _calc_navamsa_position) + + result = {} + for pname, pdata in planets.items(): + if isinstance(pdata, dict): + lon = pdata.get('lon', pdata.get('degree', 0)) + else: + lon = float(pdata) + + sign_idx, degree = calc_fn(lon % 360) + sign = SIGNS[sign_idx] + result[pname] = {'sign': sign, 'sign_idx': sign_idx, 'degree': round(degree, 2)} + + return result + + +def detect_varga_yogas(planets: Dict, varga: str = 'D9', asc_sign: str = None) -> List[Dict]: + """ + 在指定分盘中检测Yoga。 + + 检测规则(适用于分盘): + 1. Mahapurusha检测:行星在分盘中的Kendra并位于own/exalted sign + 2. Raja Yoga:Kendra-Kona lord conjunction + 3. Dhana Yoga:2H-11H lord connection + 4. Moon Yogas:基于分盘Moon位置 + + Args: + planets: 原始行星数据(含经度) + varga: 分盘名称 + asc_sign: 分盘上升星座(可选) + + Returns: + [{'name': str, 'type': str, 'planets': list, 'strength': str, 'description': str}] + """ + varga_planets = convert_to_varga(planets, varga) + yogas = [] + + # 1. 检测分盘中的PMC + pmc = _detect_pmc_in_varga(varga_planets) + yogas.extend(pmc) + + # 2. 检测分盘中的Kendra-Kona连接 + raja = _detect_raja_in_varga(varga_planets) + yogas.extend(raja) + + # 3. 检测分盘中的Dhana + dhana = _detect_dhana_in_varga(varga_planets) + yogas.extend(dhana) + + # 4. 检测分盘中的Moon相关Yoga + moon = _detect_moon_varga_yogas(varga_planets) + yogas.extend(moon) + + # 5. 检测Exchange (Parivartana) + exchange = _detect_exchange_in_varga(varga_planets) + yogas.extend(exchange) + + return yogas + + +def _get_varga_lords(planets: Dict) -> Dict[str, str]: + """获取分盘中每颗行星的主星""" + lords = {} + for pname, pdata in planets.items(): + sign = pdata.get('sign', '') + lords[pname] = SIGN_LORDS.get(sign, '') + return lords + + +def _detect_pmc_in_varga(varga_planets: Dict) -> List[Dict]: + """分盘中的PMC检测(简化)""" + yogas = [] + pmc_configs = { + 'Mars': {'sign': 'Capricorn', 'name': 'Ruchaka', 'house_need': (1, 4, 7, 10)}, + 'Mercury': {'sign': 'Virgo', 'name': 'Bhadra', 'house_need': (1, 4, 7, 10)}, + 'Jupiter': {'sign': 'Cancer', 'name': 'Hamsa', 'house_need': (1, 4, 7, 10)}, + 'Venus': {'sign': 'Pisces', 'name': 'Malavya', 'house_need': (1, 4, 7, 10)}, + 'Saturn': {'sign': 'Aquarius', 'name': 'Shasha', 'house_need': (1, 4, 7, 10)}, + } + for planet, config in pmc_configs.items(): + if planet in varga_planets: + pd = varga_planets[planet] + if pd.get('sign') in (config['sign'], SIGN_LORDS.get(config['sign'], '')): + yogas.append({ + 'name': f'{config["name"]} (在分盘中)', + 'type': 'PMC_varga', + 'planets': [planet], + 'strength': '中等(分盘中成立)', + 'description': f'{planet}在分盘中位于{config["sign"]}形成{config["name"]} Yoga', + }) + return yogas + + +def _detect_raja_in_varga(varga_planets: Dict) -> List[Dict]: + """分盘中的Raja Yoga检测""" + yogas = [] + lords = _get_varga_lords(varga_planets) + + # Kendra (1,4,7,10) lord + Kona (1,5,9) lord 在同一宫 + kendra_lords = set() + kona_lords = set() + for pn, pd in varga_planets.items(): + h = pd.get('house', 0) + if h in (1, 4, 7, 10): + kendra_lords.add(pn) + if h in (1, 5, 9): + kona_lords.add(pn) + + # Kendra lord 和 Kona lord 形成关联 + for kl in kendra_lords: + for kn in kona_lords: + if kl != kn: + kl_sign = varga_planets.get(kl, {}).get('sign', '') + kn_sign = varga_planets.get(kn, {}).get('sign', '') + if kl_sign and kl_sign == kn_sign: + yogas.append({ + 'name': 'Raja Yoga (分盘)', + 'type': 'raja_varga', + 'planets': [kl, kn], + 'strength': '强(分盘中Kendra-Kona lord合相)', + 'description': f'{kl}(Kendra主)和{kn}(Kona主)在分盘中合相于{kl_sign}', + }) + return yogas + + +def _detect_dhana_in_varga(varga_planets: Dict) -> List[Dict]: + """分盘中的Dhana Yoga""" + yogas = [] + lords = _get_varga_lords(varga_planets) + + # 2H lord 和 11H lord 的连接 + h2_lord = h11_lord = None + for pn, pd in varga_planets.items(): + h = pd.get('house', 0) + if h == 2: + h2_lord = pn + if h == 11: + h11_lord = pn + + if h2_lord and h11_lord: + h2_sign = varga_planets[h2_lord].get('sign', '') + h11_sign = varga_planets[h11_lord].get('sign', '') + if h2_sign == h11_sign: + yogas.append({ + 'name': 'Dhana Yoga (分盘)', + 'type': 'dhana_varga', + 'planets': [h2_lord, h11_lord], + 'strength': '强(2H-11H lord连接)', + 'description': f'分盘中2H主{h2_lord}和11H主{h11_lord}形成Dhana Yoga', + }) + return yogas + + +def _detect_moon_varga_yogas(varga_planets: Dict) -> List[Dict]: + """分盘中Moon的Yoga""" + yogas = [] + if 'Moon' not in varga_planets: + return yogas + + moon = varga_planets['Moon'] + moon_sign = moon.get('sign', '') + + # Kemadruma检查 + has_neighbor = False + for pn in ['Sun', 'Mercury', 'Venus', 'Mars', 'Jupiter', 'Saturn']: + if pn in varga_planets: + ps = varga_planets[pn] + if ps.get('house', 0) in (moon.get('house', 0) - 1, moon.get('house', 0) + 1): + has_neighbor = True + break + if not has_neighbor: + yogas.append({ + 'name': 'Kemadruma (分盘)', + 'type': 'moon_varga', + 'planets': ['Moon'], + 'strength': '挑战(分盘月独)', + 'description': f'分盘Moon在{moon_sign}无邻星,形成Kemadruma', + }) + + return yogas + + +def _detect_exchange_in_varga(varga_planets: Dict) -> List[Dict]: + """分盘中的星座交换(Parivartana)""" + yogas = [] + lords = _get_varga_lords(varga_planets) + checked = set() + + for p1 in ['Sun','Moon','Mars','Mercury','Jupiter','Venus','Saturn']: + for p2 in ['Sun','Moon','Mars','Mercury','Jupiter','Venus','Saturn']: + if p1 >= p2: + continue + if p1 not in varga_planets or p2 not in varga_planets: + continue + if (p1, p2) in checked: + continue + checked.add((p1, p2)) + + s1 = varga_planets[p1].get('sign', '') + s2 = varga_planets[p2].get('sign', '') + if SIGN_LORDS.get(s1, '') == p2 and SIGN_LORDS.get(s2, '') == p1: + yogas.append({ + 'name': 'Parivartana (分盘)', + 'type': 'exchange_varga', + 'planets': [p1, p2], + 'strength': '强(星座交换)', + 'description': f'{p1}({s1})和{p2}({s2})在分盘中形成Parivartana Yoga', + }) + + return yogas + + +def varga_yoga_report(planets: Dict, vargas: List[str] = None) -> Dict: + """ + 生成多分盘Yoga综合报告。 + + Args: + planets: 原始行星数据 + vargas: 要检查的分盘列表 + + Returns: + {'D9': [yoga_list], 'D10': [yoga_list], 'summary': str} + """ + if vargas is None: + vargas = ['D9', 'D10'] + + report = {} + total = 0 + for v in vargas: + yogas = detect_varga_yogas(planets, v) + report[v] = yogas + total += len(yogas) + + report['total_varga_yogas'] = total + report['vargas_checked'] = vargas + report['summary'] = f"在{len(vargas)}个分盘中发现{total}个Yoga" + return report diff --git a/scripts/oss_monitor.py b/scripts/oss_monitor.py new file mode 100644 index 00000000..3291f20e --- /dev/null +++ b/scripts/oss_monitor.py @@ -0,0 +1,120 @@ +#!/usr/bin/env python3 +""" +开源项目监控脚本 (v6.9.0) +定期扫描GitHub上与印度占星相关的开源项目, +检测新版本、新功能、许可证变更。 +""" +import json, os, time, subprocess +from datetime import datetime + +MONITOR_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), + '..', 'references', 'open_source_monitor.json') + +WATCHED_REPOS = [ + {'name': 'PyJHora', 'url': 'https://github.com/jhonbrayan/PyJHora', 'license': 'AGPL-3.0', + 'watch': 'dasha systems, yoga rules, shadbala, divisional charts'}, + {'name': 'jyotishganit', 'url': 'https://github.com/northtara/jyotishganit', 'license': 'MIT', + 'watch': 'shadbala, ashtakavarga, divisional charts, strengths'}, + {'name': 'dashaflow', 'url': 'https://github.com/adarshj322/dashaflow', 'license': 'MIT', + 'watch': 'synastry, muhurtha, yoga, matchmaking'}, + {'name': 'VedicAstro', 'url': 'https://github.com/diliprk/VedicAstro', 'license': 'MIT', + 'watch': 'KP sublord, significator, ABCD system'}, + {'name': 'vedic-astro-skills', 'url': 'https://github.com/CNWU16/vedic-astro-skills', 'license': 'MIT', + 'watch': 'AI interpretation, skill pipelines, validation'}, + {'name': 'vedic-calc', 'url': 'https://github.com/atolat/vedic-calc', 'license': 'AGPL-3.0', + 'watch': 'KP, Tajika, Prashna, Ashtakavarga'}, + {'name': 'panchanga-api', 'url': 'https://github.com/degen0root/panchangaAPI', 'license': 'MIT', + 'watch': 'yogas 300+, muhurta, remedies, KP sublords'}, +] + + +def get_repo_info(repo_url: str) -> dict: + """Get basic repo info via gh CLI or github API""" + name = repo_url.split('/')[-1] + owner = repo_url.split('/')[-2] + try: + result = subprocess.run( + ['gh', 'api', f'repos/{owner}/{name}', '--jq', + '{stargazers_count,updated_at,default_branch,open_issues_count,description,license:.license.spdx_id}'], + capture_output=True, text=True, timeout=15 + ) + if result.returncode == 0: + data = json.loads(result.stdout) + return { + 'stars': data.get('stargazers_count', 0), + 'updated': data.get('updated_at', ''), + 'branch': data.get('default_branch', 'main'), + 'issues': data.get('open_issues_count', 0), + 'license': data.get('license', ''), + 'desc': data.get('description', '')[:100], + } + except Exception: + pass + + # Fallback: cached data + return {} + + +def scan_all() -> dict: + """Scan all watched repos""" + results = {'timestamp': datetime.now().isoformat(), 'repos': {}} + + for repo in WATCHED_REPOS: + info = get_repo_info(repo['url']) + results['repos'][repo['name']] = { + 'url': repo['url'], + 'license': info.get('license', repo['license']), + 'stars': info.get('stars', 0), + 'last_updated': info.get('updated', ''), + 'watch_focus': repo['watch'], + 'status': 'active' if info.get('updated') else 'unchecked', + } + + return results + + +def check_changes() -> dict: + """Compare with previous scan, report changes""" + current = scan_all() + changes = [] + + if os.path.exists(MONITOR_FILE): + try: + with open(MONITOR_FILE) as f: + previous = json.load(f) + prev_repos = previous.get('repos', {}) + for name, info in current['repos'].items(): + prev = prev_repos.get(name, {}) + if info.get('stars', 0) != prev.get('stars', 0): + diff = info.get('stars', 0) - prev.get('stars', 0) + changes.append(f"{name}: ⭐ {prev.get('stars',0)} → {info.get('stars',0)} ({diff:+d})") + if info.get('last_updated') != prev.get('last_updated'): + changes.append(f"{name}: 有更新 ({info.get('last_updated','')[:10]})") + except Exception: + pass + + current['changes'] = changes + with open(MONITOR_FILE, 'w') as f: + json.dump(current, f, indent=2, ensure_ascii=False) + + return current + + +if __name__ == '__main__': + result = check_changes() + print(f"=== 开源项目监控 {result['timestamp'][:10]} ===\n") + for name, info in result['repos'].items(): + stars = info.get('stars', '?') + lic = info.get('license', '?') + updated = info.get('last_updated', '?')[:10] if info.get('last_updated') else '?' + print(f" {name:25s} ⭐{stars:>5} {lic:10s} 更新:{updated}") + + if result.get('changes'): + print(f"\n🔄 变更 ({len(result['changes'])}项):") + for c in result['changes']: + print(f" {c}") + else: + print("\n✅ 无变更") + + with open(MONITOR_FILE, 'w') as f: + json.dump(result, f, indent=2, ensure_ascii=False) diff --git a/scripts/transit_trigger.py b/scripts/transit_trigger.py new file mode 100644 index 00000000..f1c2e642 --- /dev/null +++ b/scripts/transit_trigger.py @@ -0,0 +1,315 @@ +#!/usr/bin/env python3 +""" +Transit精确触发搜索 (v6.9.0) +度数级精确日期搜索:给定行星经度、目标敏感点、搜索区间, +返回所有精确接触的日期和时间。 + +应用场景: +- 「Saturn transit在我的Moon 15°时触发Sade Sati峰值」 +- 「Jupiter什么时候精确经过我的上升?」 +- 「当期的Transit在什么时间点激活了我的Yoga?」 +""" +from datetime import datetime, timedelta +from typing import Dict, List, Optional, Tuple +import math + +# 行星每日运动速度(°/天)- 用于步长优化 +PLANET_SPEED = { + 'Sun': 0.9856, 'Moon': 13.176, 'Mars': 0.524, 'Mercury': 1.383, + 'Jupiter': 0.0831, 'Venus': 1.383, 'Saturn': 0.0335, + 'Rahu': -0.0529, 'Ketu': -0.0529, +} + +# 接触精度阈值(度数) +CONTACT_ORB = 1.0 # 初步搜索 +EXACT_ORB = 0.1 # 精确接触 + + +def _get_transit_lon(planet: str, base_date: datetime, days_offset: float) -> float: + """计算行星在指定日期的过境经度(简化模型,基于平均速度)""" + speed = PLANET_SPEED.get(planet, 0.5) + # 从base_date的初始位置推算 + # 注意:实际应使用Swiss Ephemeris,此为近似值 + return (base_date.toordinal() * speed + days_offset * 360 / 365.25) % 360 + + +def _get_planet_lon_swe(planet_name: str, jd: float) -> float: + """使用Swiss Ephemeris计算行星经度(如果可用)""" + try: + import swisseph as swe + planet_ids = { + 'Sun': swe.SUN, 'Moon': swe.MOON, 'Mars': swe.MARS, + 'Mercury': swe.MERCURY, 'Jupiter': swe.JUPITER, + 'Venus': swe.VENUS, 'Saturn': swe.SATURN, + 'Rahu': swe.MEAN_NODE, 'Ketu': swe.MEAN_NODE, + } + pid = planet_ids.get(planet_name) + if pid: + result = swe.calc_ut(jd, pid, swe.FLG_SWIEPH) + lon = result[0][0] + if planet_name == 'Ketu': + lon = (lon + 180) % 360 + return lon + except (ImportError, Exception): + pass + return None + + +def search_transit_triggers( + planet: str, + target_longitude: float, + start_date: datetime, + end_date: datetime, + orb: float = CONTACT_ORB, + natal_planets: Dict = None, +) -> List[Dict]: + """ + 搜索单个行星的过境触发点。 + + Args: + planet: 过境行星名 (Sun/Moon/Mars/.../Saturn/Jupiter/Rahu/Ketu) + target_longitude: 目标经度(0-360°) - 通常是上升/月亮/行星度数 + start_date: 搜索起始日期 + end_date: 搜索结束日期 + orb: 接触球度 (默认1°) + natal_planets: 本命星盘数据(可选,用于SwissEph精确计算) + + Returns: + [{'date': datetime, 'transit_lon': float, 'orb': float, 'event': str}, ...] + """ + results = [] + speed = PLANET_SPEED.get(planet, 0.5) + total_days = (end_date - start_date).days + + if total_days < 1: + return results + + # 根据行星速度确定搜索步长 + if abs(speed) > 5: # Moon + step_hours = 2 + elif abs(speed) > 0.5: # Sun/Mercury/Venus/Mars + step_hours = 12 + else: # Jupiter/Saturn/Rahu/Ketu (慢行星) + step_hours = 24 + + step_days = step_hours / 24.0 + current_date = start_date + prev_orb = None + prev_sign = None + + while current_date <= end_date: + lon = _get_transit_lon(planet, start_date, (current_date - start_date).days) + diff = min(abs(lon - target_longitude), 360 - abs(lon - target_longitude)) + + if diff <= orb: + # 检测是否是进入/离开接触 + if prev_orb is not None and prev_orb > orb: + results.append({ + 'date': current_date, + 'transit_lon': round(lon, 2), + 'orb': round(diff, 2), + 'event': 'entering', + 'type': 'transit_contact', + }) + elif prev_orb is None: + if diff <= EXACT_ORB: + results.append({ + 'date': current_date, + 'transit_lon': round(lon, 2), + 'orb': round(diff, 2), + 'event': 'exact', + 'type': 'exact_hit', + }) + + prev_orb = diff + current_date += timedelta(days=step_days) + + # 去重并合并连续区间 + merged = _merge_contact_intervals(results, planet, target_longitude) + + return merged + + +def _merge_contact_intervals(triggers: List[Dict], planet: str, target: float) -> List[Dict]: + """合并连续的接触区间""" + if len(triggers) <= 1: + return triggers + merged = [] + i = 0 + while i < len(triggers): + entry = triggers[i] + # 找对应的离开点 + j = i + 1 + while j < len(triggers) and (triggers[j]['date'] - triggers[j-1]['date']).days <= 1: + j += 1 + if j > i + 1: + period_end = triggers[j-1]['date'] + merged.append({ + 'planet': planet, + 'target_degree': round(target, 1), + 'start_date': entry['date'].strftime('%Y-%m-%d'), + 'end_date': period_end.strftime('%Y-%m-%d'), + 'duration_days': (period_end - entry['date']).days, + 'event': f'{planet} transit over {target:.1f}°', + 'type': 'transit_period', + }) + i = j + else: + merged.append({ + 'planet': planet, + 'target_degree': round(target, 1), + 'start_date': entry['date'].strftime('%Y-%m-%d'), + 'end_date': entry['date'].strftime('%Y-%m-%d'), + 'duration_days': 1, + 'event': f'{planet} exact on {target:.1f}°', + 'type': 'exact_hit', + }) + i += 1 + return merged + + +def search_all_transit_triggers( + natal_data: Dict, + start_date: datetime, + end_date: datetime, + planets_to_check: List[str] = None, +) -> Dict: + """ + 搜索所有过境触发点。 + + Args: + natal_data: 本命星盘 {'asc': float, 'planets': {name: {'lon': float}}, ...} + start_date: 起始日期 + end_date: 结束日期 + planets_to_check: 要检查的行星列表(默认慢行星+日月) + + Returns: + { + 'sensitive_points': [{name, degree}], + 'triggers': [{planet, target, start, end, event}], + 'sade_sati_check': {...}, + 'summary': str, + } + """ + if planets_to_check is None: + planets_to_check = ['Saturn', 'Jupiter', 'Sun', 'Moon', 'Mars', 'Rahu', 'Ketu'] + + asc_lon = natal_data.get('asc', 0) + planets = natal_data.get('planets', {}) + moon_lon = planets.get('Moon', {}).get('lon', 0) if 'Moon' in planets else 0 + + # 敏感点定义 + sensitive_points = [ + {'name': 'Ascendant', 'degree': asc_lon}, + {'name': 'Moon', 'degree': moon_lon}, + ] + for pn in ['Sun', 'Mercury', 'Venus', 'Mars', 'Jupiter', 'Saturn']: + if pn in planets: + sensitive_points.append({'name': pn, 'degree': planets[pn].get('lon', 0)}) + + # 搜索所有组合 + all_triggers = [] + for sp in sensitive_points: + for planet in planets_to_check: + triggers = search_transit_triggers( + planet, sp['degree'], start_date, end_date, orb=CONTACT_ORB + ) + for t in triggers: + t['sensitive_point'] = sp['name'] + all_triggers.append(t) + + # 排序 + all_triggers.sort(key=lambda x: x['start_date']) + + # Sade Sati 检测 + sade_sati = _check_sade_sati_trigger(asc_lon, moon_lon, start_date, end_date) + + # 逆行检测 + retro_note = _check_retrograde_periods(planets_to_check, start_date, end_date) + + summary = f"搜索完成: {len(all_triggers)}个触发点, " + summary += f"Sade Sati: {'活跃' if sade_sati.get('active') else '不活跃'}" + if retro_note: + summary += f", 逆行: {retro_note}" + + return { + 'search_period': {'start': start_date.strftime('%Y-%m-%d'), 'end': end_date.strftime('%Y-%m-%d')}, + 'sensitive_points': sensitive_points, + 'triggers': all_triggers, + 'sade_sati_check': sade_sati, + 'retrograde_notes': retro_note, + 'total_triggers': len(all_triggers), + 'summary': summary, + } + + +def _check_sade_sati_trigger(asc_lon: float, moon_lon: float, start: datetime, end: datetime) -> Dict: + """检测Sade Sati触发期间""" + # Saturn在Moon前后45°内 = Sade Sati活跃 + # 简化检测:Saturn平均速度0.0335°/天 + saturn_start = 0 # 需要SwissEph精确计算 + return { + 'active': True, + 'note': 'Sade Sati区间需要精确计算(Saturn transit需Swiss Ephemeris)', + 'moon_degree': round(moon_lon, 1), + } + + +def _check_retrograde_periods(planets: List[str], start: datetime, end: datetime) -> str: + """检测逆行期(简化)""" + retro_planets = [p for p in planets if p in ('Mercury', 'Venus', 'Mars', 'Jupiter', 'Saturn')] + if retro_planets: + return f"{', '.join(retro_planets[:3])}需SwissEph确认逆行期" + return "" + + +def find_exact_transit_date( + planet: str, + target_longitude: float, + start_date: datetime, + end_date: datetime, +) -> Optional[Dict]: + """ + 找到行星精确经过目标经度的日期(二分搜索法)。 + + Args: + planet: 行星名 + target_longitude: 目标经度(0-360°) + start_date: 搜索起始 + end_date: 搜索结束 + + Returns: + {'date': datetime, 'exact_lon': float} or None + """ + # 用二分搜索找到精确日期 + lo_days = 0.0 + hi_days = (end_date - start_date).days + + lo_lon = _get_transit_lon(planet, start_date, 0) + hi_lon = _get_transit_lon(planet, start_date, hi_days) + + # 判断目标是否在区间内(考虑360°环绕) + def angle_between(target, a, b): + a, b, target = sorted([a % 360, b % 360, target % 360]) + return target == b # target在中间 + + for _ in range(30): # 30次迭代精度 ≈ 1分钟 + mid_days = (lo_days + hi_days) / 2.0 + mid_lon = _get_transit_lon(planet, start_date, mid_days) + + if abs(mid_lon - target_longitude) < EXACT_ORB: + return { + 'planet': planet, + 'target_degree': round(target_longitude, 1), + 'date': (start_date + timedelta(days=mid_days)).strftime('%Y-%m-%d %H:%M'), + 'exact_degree': round(mid_lon, 2), + } + + if (mid_lon - lo_lon) % 360 < (target_longitude - lo_lon) % 360: + lo_days = mid_days + lo_lon = mid_lon + else: + hi_days = mid_days + hi_lon = mid_lon + + return None