Jesse_Chen
f224146348
Three-way-merge calculation modules and pl9-export into the product fork while keeping commercial API routes, Raman ayanamsa, and the consultation contract as a keypath superset. Co-authored-by: Cursor <cursoragent@cursor.com>
683 lines
28 KiB
Python
683 lines
28 KiB
Python
#!/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,
|
||
}
|