#!/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 import math from bisect import bisect_right from datetime import datetime, timedelta from fractions import Fraction from pathlib import Path from typing import Any, Dict, List, Tuple, Optional 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） KP_LORDS = ["Ketu", "Venus", "Sun", "Moon", "Mars", "Rahu", "Jupiter", "Saturn", "Mercury"] try: from dasha_common import VIMSHOTTARI_YEARS as _SHARED_VIMSHOTTARI_YEARS except ModuleNotFoundError: # pragma: no cover - package import from scripts.dasha_common import VIMSHOTTARI_YEARS as _SHARED_VIMSHOTTARI_YEARS VIMSHOTTARI_DURATION = [_SHARED_VIMSHOTTARI_YEARS[lord] for lord in KP_LORDS] 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 def _angular_partition_starts(): """Exact proportional starts, rounded once at the float API boundary.""" starts, owners = [], [] for star in range(27): sub_start = Fraction(40 * star, 3) for offset in range(9): sub = (star + offset) % 9 sub_span = Fraction(40, 3) * VIMSHOTTARI_DURATION[sub] / 120 child_start = sub_start for child_offset in range(9): child = (sub + child_offset) % 9 starts.append(float(child_start)) owners.append((sub, child)) child_start += sub_span * VIMSHOTTARI_DURATION[child] / 120 sub_start += sub_span return starts, owners _KP_STARTS, _KP_OWNERS = _angular_partition_starts() _STAR_STARTS = [float(Fraction(40 * i, 3)) for i in range(27)] _PADA_STARTS = [float(Fraction(10 * i, 3)) for i in range(108)] ROOT = Path(__file__).resolve().parents[1] 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字典 """ if isinstance(degree, bool) or not isinstance(degree, (int, float)) or not math.isfinite(degree): raise ValueError('KP longitude must be a finite number') deg = degree % 360 # 1. Sign lord sign_index = int(deg // 30) # 2. Nakshatra. A shared boundary belongs to the subdivision that opens there. nakshatra_index = bisect_right(_STAR_STARTS, deg) - 1 pada = (bisect_right(_PADA_STARTS, deg) - 1) % 4 + 1 sub, child = _KP_OWNERS[bisect_right(_KP_STARTS, deg) - 1] return { 'rasi_lord': SIGN_LORDS.get(SIGNS[sign_index], ''), 'sign': SIGNS[sign_index], 'nakshatra': NAKSHATRAS[nakshatra_index], 'nakshatra_lord': STAR_LORDS[nakshatra_index], 'pada': pada, 'sub_lord': KP_LORDS[sub], 'sub_sub_lord': KP_LORDS[child], } 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, house_occupancy_basis: Optional[str] = None, ) -> Dict: """ 完整KP分析（基于 diliprk/VedicAstro MIT 算法）。 Args: planet_positions: 行星位置 {planet: {'sign': str, 'degree': float, 'house': int}} asc_sign: 上升星座名称（无显式宫头时用于 whole-sign 代理） house_cusps: 可选的 12 个实际宫头黄经。提供时优先用于 KP 宫头与显著星。 house_occupancy_basis: 传了宫头就必须声明判宫口径；没传宫头时记 whole_sign_proxy。 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 not house_occupancy_basis: raise ValueError('house_occupancy_basis is required when house_cusps are provided') if explicit_cusps and len(house_cusps) != 12: raise ValueError('house_cusps must contain exactly 12 longitudes') occupancy_basis = house_occupancy_basis if explicit_cusps else 'whole_sign_proxy' # 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', 'house_occupancy_basis': occupancy_basis, '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"- House occupancy basis: `{kp_analysis.get('house_occupancy_basis') or 'whole_sign_proxy'}`", 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, }