#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ KP (Krishnamurti Paddhati) 占星系统模块 基于 diliprk/VedicAstro (MIT License) 核心算法适配 核心功能： 1. Sublord/Subsublord 计算（基于Vimshottari比例划分） 2. Planet Significator ABCD体系 3. House Significator ABCD体系 """ import json from datetime import datetime, timedelta from pathlib import Path from typing import Any, Dict, List, Tuple, Optional 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' ] # Vimshottari年限（KP系统使用相同比例划分sublord） VIMSHOTTARI_DURATION = [7, 20, 6, 10, 7, 18, 16, 19, 17] KP_LORDS = ["Ketu", "Venus", "Sun", "Moon", "Mars", "Rahu", "Jupiter", "Saturn", "Mercury"] STAR_LORDS = KP_LORDS * 3 # 27 Nakshatras = 3 cycles of 9 lords VIMSHOTTARI_YEARS = dict(zip(KP_LORDS, VIMSHOTTARI_DURATION)) NAKSHATRA_SPAN = 360.0 / 27.0 # 13.333... degrees ROOT = Path(__file__).resolve().parents[1] def _load_json_artifact(relative_path: str) -> Dict[str, Any]: path = ROOT / relative_path with open(path, encoding='utf-8') as f: data = json.load(f) data.setdefault('artifact_path', relative_path) return data def kp_maturity_profile() -> Dict[str, Any]: """Return the current KP maturity boundary for reports and API consumers. This intentionally reuses pinned oracle/status packets instead of inventing a second truth policy inside runtime code. It is a display/report guard: KP layers can be shown, but prediction truth stays blocked until the status packets say the numeric oracle and independent holdout gates are closed. """ event = _load_json_artifact('references/oracle/kp_exact_cusp_mainline_status_2026_08_22.json') gate = _load_json_artifact('references/oracle/kp_exact_cusp_closure_dashboard_2026_08_22.json') replay = _load_json_artifact('references/oracle/kp_real_event_replay_gate_2026_07_30.json') cusp = _load_json_artifact('references/oracle/kp_12_cusp_numeric_oracle_readiness_2026_07_23.json') table = _load_json_artifact('references/oracle/kp_external_table_hash_manifest_2026_07_20.json') workflow_gate = _load_json_artifact('references/oracle/kp_significator_workflow_gate_2026_07_23.json') runtime_evidence_layers = workflow_gate.get('runtime_evidence_layers') or [] remaining_hard_reasons = ((event.get('promotion_gate') or {}).get('remaining_hard_reasons')) or [] holdout_counts = replay.get('current_holdout_counts') or {} required_counts = replay.get('required_holdout_counts') or {} closed = ( event.get('truth_matrix_allowed') is True and event.get('timing_truth_promoted') is True and replay.get('timing_truth_promoted') is True and (holdout_counts.get('frozen_positive_count') or 0) >= (required_counts.get('minimum_frozen_positive') or 20) and (holdout_counts.get('frozen_negative_count') or 0) >= (required_counts.get('minimum_frozen_negative') or 80) ) if closed: claim_status = 'timing_truth_closed' display_policy = 'verified_prediction_allowed_with_evidence' else: claim_status = 'observation_only_truth_blocked' display_policy = 'show_kp_layers_as_research_evidence_only_do_not_claim_precise_timing' blockers = [] blockers.extend([row.get('blocker') for row in (event.get('remaining_blockers') or []) if isinstance(row, dict)]) blockers.extend(replay.get('maturity_gap') or []) blockers.extend(cusp.get('remaining_blockers') or []) blockers = list(dict.fromkeys(str(item) for item in blockers if item)) return { 'scope': 'kp_maturity_profile', 'claim_status': claim_status, 'display_policy': display_policy, 'timing_truth_promoted': closed, 'truth_matrix_allowed': closed, 'production_tuning_allowed': closed, 'source_artifacts': { 'event_closure_status': event['artifact_path'], 'event_closure_dashboard': gate['artifact_path'], 'real_event_replay_gate': replay['artifact_path'], 'cusp_numeric_oracle_readiness': cusp['artifact_path'], 'external_table_hash_manifest': table['artifact_path'], 'significator_workflow_gate': workflow_gate['artifact_path'], }, 'runtime_evidence_layers': runtime_evidence_layers, 'runtime_evidence_layer_count': len(runtime_evidence_layers), 'remaining_hard_reasons': remaining_hard_reasons, 'remaining_hard_reason_count': len(remaining_hard_reasons), 'selected_lane': ((event.get('selector') or {}).get('selected_lane')), 'fallback_lane': ((event.get('selector') or {}).get('fallback_lane')), 'promotion_gate_passed': bool((event.get('promotion_gate') or {}).get('gate_passed', False)), 'closed_numeric_assets': { 'kp_sub_lord_fixture_hash_fixed': table.get('status') == 'fixed_hash' and table.get('row_count') == 249, 'kp_sub_lord_midpoint_all_249_guarded': True, 'kp_sub_lord_249_segment_parity_closed': table.get('status') == 'fixed_hash' and table.get('row_count') == 249, 'public_12_cusp_packet_ready': bool(cusp.get('ready_evidence')), }, 'closed_oracle_parity_assets': { 'kp_sub_lord_249_segment_parity': { 'status': 'closed', 'scope': 'structure_only_not_event_timing', 'source_artifact': table['artifact_path'], 'test_guard': 'tests/test_kp_system.py::test_kp_sublord_matches_vedicastro_csv_all_249_segments', 'boundary': 'This closes the 249 SubLord segmentation/table parity only; it does not close cusp, ruling-planet, significator workflow, DBA, real-event, or timing-outcome truth.', }, }, 'holdout_counts': holdout_counts, 'required_holdout_counts': required_counts, 'remaining_blockers': blockers, 'claim_boundary': ( 'KP star/sub/sub-sub/significator/DBA layers may be displayed in the personal report, ' 'but exact event timing remains blocked until numeric oracle settings, ruling planets, ' 'and independent 20 positive / 80 negative holdout replay are closed.' ), } def build_kp_western_support_surface( western_support: Optional[Dict[str, Any]] = None, *, maturity_profile: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Normalize Western support for KP report-facing surfaces only. This packages support/convergence/negative-evidence summaries without changing the underlying KP maturity boundary. """ maturity = maturity_profile or kp_maturity_profile() source = western_support if isinstance(western_support, dict) else {} convergence_source = source.get('convergence') if isinstance(source.get('convergence'), dict) else {} negative_source = source.get('negative_evidence') if isinstance(source.get('negative_evidence'), dict) else {} shared_signal_count = convergence_source.get('shared_signal_count') shared_signal_count = shared_signal_count if isinstance(shared_signal_count, int) else 0 conflict_count = convergence_source.get('conflict_count') conflict_count = conflict_count if isinstance(conflict_count, int) else 0 missing_layers = negative_source.get('missing_layers') missing_layers = [str(item) for item in missing_layers] if isinstance(missing_layers, list) else [] rejected_windows = negative_source.get('rejected_windows') rejected_windows = [str(item) for item in rejected_windows] if isinstance(rejected_windows, list) else [] convergence_status = str(convergence_source.get('status') or ('not_provided' if not source else 'partial')) negative_status = str(negative_source.get('status') or ('not_provided' if not source else 'partial')) if not source: status = 'not_provided' elif 'blocked' in {convergence_status, negative_status}: status = 'blocked' elif 'partial' in {convergence_status, negative_status}: status = 'partial' elif convergence_status == negative_status == 'used': status = 'used' else: status = 'partial' convergence_summary = convergence_source.get('summary') if not convergence_summary: convergence_summary = f'{shared_signal_count} shared signals; {conflict_count} conflicts' negative_summary = negative_source.get('summary') if not negative_summary: negative_summary = f'{len(missing_layers)} missing layers; {len(rejected_windows)} rejected windows' return { 'status': status, 'convergence': { 'status': convergence_status, 'shared_signal_count': shared_signal_count, 'conflict_count': conflict_count, 'summary': str(convergence_summary), }, 'negative_evidence': { 'status': negative_status, 'missing_layers': missing_layers, 'rejected_windows': rejected_windows, 'summary': str(negative_summary), }, 'claim_boundary': ( 'Western support remains a KP support layer only; it cannot upgrade blocked KP timing truth ' 'or replace Jyotish-first adjudication.' ), 'maturity_claim_status': maturity.get('claim_status'), 'truth_matrix_allowed': maturity.get('truth_matrix_allowed') is True, } def get_kp_lords(degree: float) -> Dict: """ KP Sublord/Subsublord 计算核心（基于 diliprk/VedicAstro MIT 算法）。 输入任意黄道经度，返回： - Rasi Lord: 星座主星 - Nakshatra: 星宿名称 - Nakshatra Lord: 星宿主星 - Nakshatra Pada: 星宿四分之一 - Sub Lord: 子主星（KP特有，按Vimshottari比例划分） - Sub Sub Lord: 次子主星（KP特有，进一步细分） Args: degree: 黄道经度（0-360） Returns: KP lords字典 """ 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 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', } def get_planet_significators(planet_positions: Dict, houses: List[Dict]) -> Dict: """ Planet Significator ABCD（基于 diliprk/VedicAstro MIT 算法）。 对每颗行星计算KP体系的A/B/C/D四个significator： - A: 星宿主星(Nakshatra Lord)所在的宫位 - B: 行星自身所在的宫位 - C: 星宿主星也是宫主星的那些宫位 - D: 行星自身也是宫主星的那些宫位 Args: planet_positions: {planet_name: {...包含kp_lords/house...}} houses: 12宫位列表 [{'house': 1, 'sign': 'Aries', 'rasi_lord': 'Mars'}, ...] Returns: Planet significators """ # 构建辅助索引 planet_kp_data = {} for pname, pdata in planet_positions.items(): kp_lords = pdata.get('kp_lords', {}) planet_kp_data[pname] = { 'nakshatra_lord': kp_lords.get('nakshatra_lord', ''), 'house': pdata.get('house', 1), } results = {} for pname, kp_data in planet_kp_data.items(): nl = kp_data['nakshatra_lord'] # A: 星宿主星所在的宫位 A = None if nl in planet_kp_data: A = planet_kp_data[nl]['house'] # B: 行星自身所在宫位 B = kp_data['house'] # C: 星宿主星是宫主星的那些宫位 C = [h['house'] for h in houses if h.get('rasi_lord', '') == nl] # D: 行星自身是宫主星的那些宫位 D = [h['house'] for h in houses if h.get('rasi_lord', '') == pname] results[pname] = {'A': A, 'B': B, 'C': C, 'D': D} return results def get_house_significators(planet_positions: Dict, houses: List[Dict]) -> Dict: """ House Significator ABCD（基于 diliprk/VedicAstro MIT 算法）。 对每个宫位计算KP体系的A/B/C/D四个significator： - A: 在该宫位居住者的星宿中的行星 - B: 该宫位中的行星 - C: 在该宫位主星的星宿中的行星 - D: 该宫位的主星 Args: planet_positions: {planet_name: {...包含kp_lords/house...}} houses: 12宫位列表 Returns: House significators """ # 构建行星ID到星宿主星的映射 planet_nl = {} for pname, pdata in planet_positions.items(): kp_lords = pdata.get('kp_lords', {}) planet_nl[pname] = kp_lords.get('nakshatra_lord', '') results = {} for h in houses: house_num = h['house'] # A: 在该宫位居住者的星宿中的行星 occupants = [pname for pname, pdata in planet_positions.items() if pdata.get('house') == house_num] A = [pname for pname, nl in planet_nl.items() if nl in occupants] # B: 该宫位中的行星 B = occupants # C: 在该宫位主星的星宿中的行星 rasi_lord = h.get('rasi_lord', '') C = [pname for pname, nl in planet_nl.items() if nl == rasi_lord] # D: 该宫位的主星 D = rasi_lord results[house_num] = {'A': A, 'B': B, 'C': C, 'D': D} return results def calc_kp_analysis( planet_positions: Dict, asc_sign: str = 'Aries', house_cusps: Optional[List[float]] = None, ) -> Dict: """ 完整KP分析（基于 diliprk/VedicAstro MIT 算法）。 Args: planet_positions: 行星位置 {planet: {'sign': str, 'degree': float, 'house': int}} asc_sign: 上升星座名称（无显式宫头时用于 whole-sign 代理） house_cusps: 可选的 12 个实际宫头黄经。提供时优先用于 KP 宫头与显著星。 Returns: 完整KP分析结果 """ asc_sign_idx = SIGNS.index(asc_sign) if asc_sign in SIGNS else 0 # 1. 为每颗行星计算KP lords kp_planets = {} for pname, pdata in planet_positions.items(): sign = pdata.get('sign', 'Aries') deg_in_sign = pdata.get('degree', 0) % 30 if sign in SIGNS: sign_idx = SIGNS.index(sign) degree = sign_idx * 30 + deg_in_sign else: degree = deg_in_sign kp_lords = get_kp_lords(degree) kp_planets[pname] = { 'sign': sign, 'degree': degree, 'house': pdata.get('house', 1), 'kp_lords': kp_lords, } explicit_cusps = house_cusps is not None if explicit_cusps and len(house_cusps) != 12: raise ValueError('house_cusps must contain exactly 12 longitudes') # 2. 构建宫位信息（含 KP lords）。无实际宫头时保留历史 whole-sign 中点代理。 houses = [] for house_num in range(1, 13): if explicit_cusps: house_center_degree = float(house_cusps[house_num - 1]) % 360.0 sign_idx = int(house_center_degree // 30) % 12 else: sign_idx = (asc_sign_idx + house_num - 1) % 12 house_center_degree = sign_idx * 30 + 15.0 sign_name = SIGNS[sign_idx] kp_lords = get_kp_lords(house_center_degree) houses.append({ 'house': house_num, 'sign': sign_name, 'rasi_lord': SIGN_LORDS.get(sign_name, ''), 'kp_lords': kp_lords, 'cusp_longitude': round(house_center_degree, 6), }) # 3. 计算significators planet_sig = get_planet_significators(kp_planets, houses) house_sig = get_house_significators(kp_planets, houses) return { 'method': 'KP (Krishnamurti Paddhati) 系统', 'version': '1.0', 'source': 'diliprk/VedicAstro MIT License', 'maturity_profile': kp_maturity_profile(), 'house_basis': 'explicit_cusps' if explicit_cusps else 'whole_sign_proxy', 'planets': {pname: {'kp_lords': data['kp_lords'], 'significators': planet_sig.get(pname, {})} for pname, data in kp_planets.items()}, 'houses': {h['house']: {'sign': h['sign'], 'cusp_longitude': h['cusp_longitude'], 'kp_lords': h['kp_lords'], 'significators': house_sig.get(h['house'], {})} for h in houses}, } def build_kp_report_pack( kp_analysis: Dict[str, Any], timeline: Optional[Dict[str, Any]] = None, *, profile: Optional[Dict[str, Any]] = None, western_support: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Build a report-ready KP evidence pack without upgrading prediction truth. This is a thin packaging layer for personal reports. It does not recompute astrology, does not adjudicate event truth, and keeps the KP maturity profile visible so downstream PL9+ renderers can show KP layers without presenting blocked timing evidence as verified prediction. """ maturity = profile or kp_analysis.get('maturity_profile') or kp_maturity_profile() planets = kp_analysis.get('planets') or {} houses = kp_analysis.get('houses') or {} periods = (timeline or {}).get('periods') or [] runtime_layers = maturity.get('runtime_evidence_layers') or [] runtime_layer_names = [str(layer.get('layer')) for layer in runtime_layers if isinstance(layer, dict) and layer.get('layer')] closure_gap_matrix = maturity.get('closure_gap_matrix') or [ {'gap_id': reason, 'truth_upgrade_allowed': False} for reason in (maturity.get('remaining_hard_reasons') or []) ] closure_gap_ids = [str(row.get('gap_id')) for row in closure_gap_matrix if isinstance(row, dict) and row.get('gap_id')] claim_boundary = maturity.get('claim_boundary') or 'KP timing truth remains blocked.' western_support_surface = build_kp_western_support_surface( western_support, maturity_profile=maturity, ) status = 'verified_prediction_allowed' if maturity.get('truth_matrix_allowed') is True else 'observation_only' must_not_claim = [ 'kp_precise_event_timing_truth_closed', 'kp_real_event_replay_completed', 'kp_holdout_20_positive_80_negative_closed', 'kp_dba_periods_are_verified_predictions', ] executive = ( 'KP layers are report-ready as research evidence, but precise event timing remains blocked.' if status == 'observation_only' else 'KP timing evidence is marked verified by the maturity profile.' ) closed_parity_assets = maturity.get('closed_oracle_parity_assets') or {} markdown = '\n'.join([ '## KP Evidence Pack', f"- status: `{status}`", f"- claim_status: `{maturity.get('claim_status')}`", f"- truth_matrix_allowed: `{maturity.get('truth_matrix_allowed')}`", f"- closed_oracle_parity_assets: `{', '.join(closed_parity_assets.keys())}`", f"- planet_count: `{len(planets)}`", f"- house_count: `{len(houses)}`", f"- dba_period_count: `{len(periods)}`", f"- runtime_evidence_layer_count: `{len(runtime_layers)}`", f"- runtime_evidence_layers: `{', '.join(runtime_layer_names)}`", f"- closure_gap_count: `{len(closure_gap_matrix)}`", f"- closure_gap_ids: `{', '.join(closure_gap_ids)}`", f"- western_support_status: `{western_support_surface.get('status')}`", f"- western_support_convergence_status: `{western_support_surface.get('convergence', {}).get('status')}`", f"- western_support_negative_evidence_status: `{western_support_surface.get('negative_evidence', {}).get('status')}`", f"- claim_boundary: {claim_boundary}", ]) return { 'schema': 'jyotish.kp_report_pack.v1', 'status': status, 'summary': { 'planet_count': len(planets), 'house_count': len(houses), 'dba_period_count': len(periods), 'runtime_evidence_layer_count': len(runtime_layers), 'closure_gap_count': len(closure_gap_matrix), 'claim_status': maturity.get('claim_status'), 'truth_matrix_allowed': maturity.get('truth_matrix_allowed') is True, 'closed_oracle_parity_asset_count': len(closed_parity_assets), 'western_support': { 'status': western_support_surface.get('status'), 'convergence_status': western_support_surface.get('convergence', {}).get('status'), 'negative_evidence_status': western_support_surface.get('negative_evidence', {}).get('status'), }, }, 'kp_analysis': kp_analysis, 'kp_dba_timeline': timeline or {}, 'maturity_profile': maturity, 'report_sections': { 'executive_summary': [executive], 'thematic_narrative': [ { 'domain': 'kp_research_evidence', 'paragraph': ( 'KP star, sub-lord, significator, and DBA layers can enrich the personal report as ' 'auditable research evidence. The report must keep the current blocked timing boundary visible.' ), 'evidence_label': status, 'must_not_claim': list(must_not_claim), } ], 'evidence_appendix': [ { 'segment_id': 'kp_maturity_profile', 'status': maturity.get('claim_status'), 'source_artifacts': maturity.get('source_artifacts') or {}, 'runtime_evidence_layers': runtime_layers, 'closed_oracle_parity_assets': closed_parity_assets, 'closure_gap_matrix': closure_gap_matrix, 'remaining_blockers': maturity.get('remaining_blockers') or [], 'western_support': western_support_surface, 'claim_boundary': claim_boundary, } ], 'pdf_sections': [executive, claim_boundary], }, 'exports': { 'markdown': markdown, 'ai_evidence_bundle': { 'schema': 'jyotish.kp_report_pack.ai_evidence.v1', 'contains_private_pl9_text': False, 'maturity_profile': maturity, 'runtime_evidence_layers': runtime_layers, 'closed_oracle_parity_assets': closed_parity_assets, 'closure_gap_matrix': closure_gap_matrix, 'western_support': western_support_surface, 'summary': { 'planet_count': len(planets), 'house_count': len(houses), 'dba_period_count': len(periods), 'runtime_evidence_layer_count': len(runtime_layers), 'closure_gap_count': len(closure_gap_matrix), }, }, }, 'audit': { 'status': status, 'must_not_claim': must_not_claim, 'claim_boundary': claim_boundary, 'normalization_boundary': 'packaging_only_no_astrological_recalculation', }, } def _kp_next_lords(start_lord: str) -> List[str]: idx = KP_LORDS.index(start_lord) return KP_LORDS[idx:] + KP_LORDS[:idx] def _kp_years_to_days(years: float) -> float: return years * 365.2425 def _kp_birth_star_balance(moon_longitude: float) -> Tuple[str, float]: moon_longitude = moon_longitude % 360.0 nak_idx = int(moon_longitude // NAKSHATRA_SPAN) % 27 star_lord = STAR_LORDS[nak_idx] elapsed = (moon_longitude % NAKSHATRA_SPAN) / NAKSHATRA_SPAN return star_lord, max(0.0, min(1.0, 1.0 - elapsed)) def _kp_period_score(lords: List[str], planet_house_significators: Optional[Dict[str, Dict]] = None) -> Dict: supportive_houses = {2, 5, 7, 11} blocking_houses = {1, 6, 8, 10, 12} supportive = 0 blocking = 0 details = {} for lord in lords: sig = (planet_house_significators or {}).get(lord, {}) houses = set() for value in sig.values(): if isinstance(value, int): houses.add(value) elif isinstance(value, list): houses.update(v for v in value if isinstance(v, int)) support_hits = sorted(houses & supportive_houses) block_hits = sorted(houses & blocking_houses) supportive += len(support_hits) blocking += len(block_hits) details[lord] = {'supportive_houses': support_hits, 'blocking_houses': block_hits} score = supportive - blocking if score >= 2: judgement = 'supportive' elif score <= -2: judgement = 'blocking' else: judgement = 'mixed' return { 'marriage_score': score, 'supportive_hits': supportive, 'blocking_hits': blocking, 'judgement': judgement, 'lord_details': details, } def calc_kp_dba_timeline( birth_datetime: datetime, moon_longitude: float, target_start: datetime, target_end: datetime, planet_house_significators: Optional[Dict[str, Dict]] = None, ) -> Dict: """Build Vimshottari MD/AD/PD windows for KP-style marriage timing review.""" birth_star_lord, balance = _kp_birth_star_balance(moon_longitude) periods = [] md_start = birth_datetime for md_i, md_lord in enumerate(_kp_next_lords(birth_star_lord) * 3): md_years = VIMSHOTTARI_YEARS[md_lord] * (balance if md_i == 0 else 1.0) md_end = md_start + timedelta(days=_kp_years_to_days(md_years)) ad_start = md_start for ad_lord in _kp_next_lords(md_lord): ad_years = md_years * VIMSHOTTARI_YEARS[ad_lord] / 120.0 ad_end = ad_start + timedelta(days=_kp_years_to_days(ad_years)) pd_start = ad_start for pd_lord in _kp_next_lords(ad_lord): pd_years = ad_years * VIMSHOTTARI_YEARS[pd_lord] / 120.0 pd_end = pd_start + timedelta(days=_kp_years_to_days(pd_years)) if pd_end >= target_start and pd_start <= target_end: scored = _kp_period_score([md_lord, ad_lord, pd_lord], planet_house_significators) periods.append({ 'md_lord': md_lord, 'ad_lord': ad_lord, 'pd_lord': pd_lord, 'start': pd_start.isoformat(), 'end': pd_end.isoformat(), **scored, }) pd_start = pd_end ad_start = ad_end md_start = md_end if md_start > target_end: break return { 'method': 'KP DBA timeline (Vimshottari MD/AD/PD)', 'birth_star_lord': birth_star_lord, 'birth_star_balance_fraction': balance, 'target_start': target_start.isoformat(), 'target_end': target_end.isoformat(), 'periods': periods, }