Files
Jyotisha/scripts/reference_transparency_contract.py
T
ClaudeandClaude Opus 5.5 295d23bf57 fix(dasha): float sub-period split on every non-frozen AD/PD path (BUG-1719)
reference_transparency_contract: birth-moment MDs + birth_moment.antardashas
instead of the date-anchored frozen service + integer-day build_antardasha.
KP monthly: elapsed years from the full-precision MD start (total_seconds).
birth_moment.split_period: generic float split, last end pinned to parent.
Frozen dasha_analyzer / domain_calculation_service untouched; rectification
scoring keeps the integer split pending a product decision.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DbBXoyn7Hj3LCW3BUs5bZ8
2026-11-07 23:35:38 +08:00

529 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""User-visible disclosure for timing, engine observations, and public case references."""
from __future__ import annotations
import json
import logging
import sys
from datetime import datetime
from functools import lru_cache
from pathlib import Path
from typing import Any
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
logger = logging.getLogger(__name__)
try:
from scripts.domain_calculation_service import compute_chart, compute_transit_longitude
from scripts.timing_precision_contract import build_timing_precision_contract
from scripts.varga import SIGNS, SIGN_LORDS, calc_varga
from scripts.chart_dasha import birth_moment
from scripts.narayana_dasha import narayana_dasha_full_report
except ModuleNotFoundError: # pragma: no cover - direct script execution
from domain_calculation_service import compute_chart, compute_transit_longitude
from timing_precision_contract import build_timing_precision_contract
from varga import SIGNS, SIGN_LORDS, calc_varga
from chart_dasha import birth_moment
from narayana_dasha import narayana_dasha_full_report
ROOT = Path(__file__).resolve().parents[1]
DEFAULT_MANIFEST = ROOT / "references" / "real_case_calibration" / "replay_manifest.json"
DEFAULT_CONTEXT_MANIFEST = ROOT / "references" / "real_case_calibration" / "public_context_manifest.json"
DOMAIN_HOUSES = {
"career": "house_10",
"marriage": "house_7",
"wealth": "house_2",
"health": "house_6",
}
FEATURE_WEIGHTS = {
"ascendant": 0.35,
"moon_sign": 0.30,
"domain_lord_sign": 0.25,
"node_axis": 0.10,
"d9_ascendant": 0.10,
"d9_venus": 0.10,
"d10_ascendant": 0.10,
"d10_sun": 0.10,
"vimshottari_mahadasha": 0.15,
"vimshottari_antardasha": 0.10,
"narayana_mahadasha_sign": 0.10,
"narayana_antardasha_sign": 0.10,
"jupiter_transit_house": 0.10,
"saturn_transit_house": 0.10,
}
HIGH_SIMILARITY_THRESHOLD = 0.75
def _sign(chart: dict[str, Any], planet: str) -> str | None:
planets = chart.get("planets") if isinstance(chart, dict) else None
value = planets.get(planet) if isinstance(planets, dict) else None
return value.get("sign") if isinstance(value, dict) else None
def _domain_lord_sign(chart: dict[str, Any], domain: str) -> str | None:
house = DOMAIN_HOUSES.get(domain)
houses = chart.get("houses") if isinstance(chart, dict) else None
house_index = int(house.split("_", 1)[1]) if house else None
house_value = None
if isinstance(houses, dict) and house:
house_value = houses.get(house)
if house_value is None and house_index is not None:
house_value = houses.get(house_index) or houses.get(str(house_index))
lord = house_value.get("lord") if isinstance(house_value, dict) else None
if not isinstance(lord, str) and isinstance(house_value, dict):
lord = SIGN_LORDS.get(house_value.get("sign") or house_value.get("cusp_sign"))
return _sign(chart, lord) if isinstance(lord, str) else None
def _longitude(chart: dict[str, Any], key: str) -> float | None:
if key == "Ascendant":
value = chart.get("ascendant") if isinstance(chart, dict) else None
else:
planets = chart.get("planets") if isinstance(chart, dict) else None
value = planets.get(key) if isinstance(planets, dict) else None
if not isinstance(value, dict):
return None
try:
return float(value["lon"])
except (KeyError, TypeError, ValueError):
return None
def _varga_sign(chart: dict[str, Any], key: str, division: int) -> str | None:
longitude = _longitude(chart, key)
return calc_varga(longitude, division)["sign"] if longitude is not None else None
def _birth_datetime(chart: dict[str, Any]) -> datetime | None:
birth = chart.get("birth_info") if isinstance(chart, dict) else None
if not isinstance(birth, dict):
birth = chart.get("birth") if isinstance(chart, dict) else None
if not isinstance(birth, dict) or not isinstance(birth.get("date"), str):
return None
raw_time = birth.get("time")
if not isinstance(raw_time, str):
raw_time = f"{int(birth.get('hour', 0)):02d}:{int(birth.get('minute', 0)):02d}:{int(birth.get('second', 0)):02d}"
try:
return datetime.fromisoformat(f"{birth['date']}T{raw_time}")
except ValueError:
return None
def _vimshottari_state(chart: dict[str, Any], reference_date: str | None) -> dict[str, str] | None:
if not isinstance(reference_date, str):
return None
try:
target = datetime.fromisoformat(reference_date[:10])
except ValueError:
return None
birth_dt, moon_lon = _birth_datetime(chart), _longitude(chart, "Moon")
if birth_dt is None or moon_lon is None:
return None
try:
# BUG-1719: 原值 domain_calculation_service.compute_vimshottari_timeline(出生日 0 点起算、
# 日期粒度)+ dasha_analyzer.build_antardasha(父段取整数天,末段早结束);
# 新值 birth_moment(cmd_dasha 口径:出生时刻起算、浮点切分、末段钉到父段终点)。
periods, *_ = birth_moment.mahadashas(birth_dt, moon_lon, through=target)
for index, period in enumerate(periods):
if period["start"] <= target < period["end"]:
antardashas = birth_moment.antardashas(period, visible_from=birth_dt if index == 0 else None)
antardasha = next(
(item["lord"] for item in antardashas if item["start"] <= target < item["end"]),
None,
)
return {"mahadasha": period["lord"], "antardasha": antardasha}
except (KeyError, TypeError, ValueError):
return None
return None
def _narayana_state(chart: dict[str, Any], reference_date: str | None) -> dict[str, str] | None:
if not isinstance(reference_date, str):
return None
try:
target = datetime.fromisoformat(reference_date[:10])
except ValueError:
return None
birth_dt = _birth_datetime(chart)
ascendant = chart.get("ascendant") if isinstance(chart, dict) else None
asc_sign = ascendant.get("sign") if isinstance(ascendant, dict) else None
planets = chart.get("planets") if isinstance(chart, dict) else None
if birth_dt is None or asc_sign not in SIGNS or not isinstance(planets, dict):
return None
try:
planet_lons = {name: float(value["lon"]) for name, value in planets.items() if isinstance(value, dict) and value.get("lon") is not None}
if not planet_lons:
return None
age = (target - birth_dt).total_seconds() / (365.25 * 86400)
if age <= 0:
return None
current = narayana_dasha_full_report(
lagna_sign_idx=SIGNS.index(asc_sign),
planet_lons=planet_lons,
current_age=age,
birth_year=birth_dt.year,
).get("current_dasha", {})
md, ad = current.get("md"), current.get("ad")
if not isinstance(md, dict) or not isinstance(ad, dict):
return None
return {"mahadasha_sign": md.get("sign"), "antardasha_sign": ad.get("sign")}
except (KeyError, TypeError, ValueError):
return None
def _timezone_offset(chart: dict[str, Any]) -> float | None:
birth = chart.get("birth_info") if isinstance(chart, dict) else None
if not isinstance(birth, dict):
birth = chart.get("birth") if isinstance(chart, dict) else None
raw = birth.get("tz") if isinstance(birth, dict) else None
try:
return float(str(raw).replace("UTC", ""))
except (TypeError, ValueError):
return None
@lru_cache(maxsize=256)
def _transit_sign_index(planet: str, reference_date: str, tz: float, ayanamsa: str) -> int | None:
try:
longitude = compute_transit_longitude(
planet=planet,
reference_date=reference_date,
tz=tz,
ayanamsa=ayanamsa,
)["longitude"]
return int(float(longitude) / 30) % 12
except (KeyError, TypeError, ValueError):
return None
def _transit_state(chart: dict[str, Any], reference_date: str | None) -> dict[str, Any] | None:
if not isinstance(reference_date, str):
return None
try:
datetime.strptime(reference_date[:10], "%Y-%m-%d")
except ValueError:
return None
ascendant = chart.get("ascendant") if isinstance(chart, dict) else None
asc_sign = ascendant.get("sign") if isinstance(ascendant, dict) else None
tz = _timezone_offset(chart)
if asc_sign not in SIGNS or tz is None:
return None
birth = chart.get("birth_info") if isinstance(chart, dict) else None
if not isinstance(birth, dict):
birth = chart.get("birth") if isinstance(chart, dict) else None
ayanamsa = birth.get("ayanamsa_name", "lahiri") if isinstance(birth, dict) else "lahiri"
asc_idx = SIGNS.index(asc_sign)
jupiter = _transit_sign_index("Jupiter", reference_date[:10], tz, str(ayanamsa))
saturn = _transit_sign_index("Saturn", reference_date[:10], tz, str(ayanamsa))
if jupiter is None or saturn is None:
return None
return {
"jupiter_transit_house": (jupiter - asc_idx) % 12 + 1,
"saturn_transit_house": (saturn - asc_idx) % 12 + 1,
"reference_date": reference_date[:10],
}
def _features(chart: dict[str, Any], domain: str) -> dict[str, Any]:
ascendant = chart.get("ascendant") if isinstance(chart, dict) else None
rahu, ketu = _sign(chart, "Rahu"), _sign(chart, "Ketu")
features = {
"ascendant": ascendant.get("sign") if isinstance(ascendant, dict) else None,
"moon_sign": _sign(chart, "Moon"),
"domain_lord_sign": _domain_lord_sign(chart, domain),
"node_axis": (rahu, ketu) if rahu is not None and ketu is not None else None,
}
if domain == "marriage":
features.update({
"d9_ascendant": _varga_sign(chart, "Ascendant", 9),
"d9_venus": _varga_sign(chart, "Venus", 9),
})
elif domain == "career":
features.update({
"d10_ascendant": _varga_sign(chart, "Ascendant", 10),
"d10_sun": _varga_sign(chart, "Sun", 10),
})
return features
@lru_cache(maxsize=64)
def _case_chart(case_id: str, year: int, month: int, day: int, hour: int, minute: int,
lat: float, lon: float, tz: float, node_mode: str) -> dict[str, Any] | None:
try:
return compute_chart({
"year": year, "month": month, "day": day, "hour": hour, "minute": minute,
"lat": lat, "lon": lon, "tz": tz, "ayanamsa": "lahiri", "node_mode": node_mode,
})
except Exception as exc:
logger.warning("reference_transparency_contract._case_chart degraded: %s", type(exc).__name__)
logger.debug("reference_transparency_contract._case_chart degraded", exc_info=True)
return None
def _chart_for_case(case: dict[str, Any]) -> dict[str, Any] | None:
provided = case.get("chart")
if isinstance(provided, dict):
return provided
subject = case.get("subject")
if not isinstance(subject, dict):
return None
required = ("year", "month", "day", "hour", "minute", "lat", "lon", "tz")
if any(subject.get(field) is None for field in required):
return None
return _case_chart(
str(case.get("case_id", "")), int(subject["year"]), int(subject["month"]), int(subject["day"]),
int(subject["hour"]), int(subject["minute"]), float(subject["lat"]), float(subject["lon"]),
float(subject["tz"]), str(subject.get("node_mode", "mean")),
)
def _similarity(
user_chart: dict[str, Any], case_chart: dict[str, Any], domain: str,
*, reference_date: str | None = None, case_event_date: str | None = None,
) -> dict[str, Any]:
user = _features(user_chart, domain)
candidate = _features(case_chart, domain)
user_dasha = _vimshottari_state(user_chart, reference_date)
case_dasha = _vimshottari_state(case_chart, case_event_date)
user_narayana = _narayana_state(user_chart, reference_date)
case_narayana = _narayana_state(case_chart, case_event_date)
user_transit = _transit_state(user_chart, reference_date)
case_transit = _transit_state(case_chart, case_event_date)
if user_dasha is not None and case_dasha is not None:
user["vimshottari_mahadasha"] = user_dasha["mahadasha"]
candidate["vimshottari_mahadasha"] = case_dasha["mahadasha"]
if user_dasha.get("antardasha") is not None and case_dasha.get("antardasha") is not None:
user["vimshottari_antardasha"] = user_dasha["antardasha"]
candidate["vimshottari_antardasha"] = case_dasha["antardasha"]
md_matches = user_dasha["mahadasha"] == case_dasha["mahadasha"]
ad_matches = user_dasha.get("antardasha") == case_dasha.get("antardasha")
timing_state = {
"status": "matched" if md_matches and ad_matches else "partial_match" if md_matches else "different",
"user_vimshottari_mahadasha": user_dasha["mahadasha"],
"case_event_vimshottari_mahadasha": case_dasha["mahadasha"],
"user_vimshottari_antardasha": user_dasha.get("antardasha"),
"case_event_vimshottari_antardasha": case_dasha.get("antardasha"),
"reference_date": reference_date,
"case_event_date": case_event_date,
}
else:
timing_state = {"status": "not_compared"}
if user_narayana is not None and case_narayana is not None:
user["narayana_mahadasha_sign"] = user_narayana["mahadasha_sign"]
candidate["narayana_mahadasha_sign"] = case_narayana["mahadasha_sign"]
user["narayana_antardasha_sign"] = user_narayana["antardasha_sign"]
candidate["narayana_antardasha_sign"] = case_narayana["antardasha_sign"]
timing_state["narayana_status"] = (
"matched"
if user_narayana == case_narayana
else "partial_match"
if user_narayana["mahadasha_sign"] == case_narayana["mahadasha_sign"]
else "different"
)
timing_state["user_narayana"] = user_narayana
timing_state["case_event_narayana"] = case_narayana
else:
timing_state["narayana_status"] = "not_compared"
if user_transit is not None and case_transit is not None:
user["jupiter_transit_house"] = user_transit["jupiter_transit_house"]
user["saturn_transit_house"] = user_transit["saturn_transit_house"]
candidate["jupiter_transit_house"] = case_transit["jupiter_transit_house"]
candidate["saturn_transit_house"] = case_transit["saturn_transit_house"]
timing_state["transit_status"] = (
"matched"
if user_transit["jupiter_transit_house"] == case_transit["jupiter_transit_house"]
and user_transit["saturn_transit_house"] == case_transit["saturn_transit_house"]
else "partial_match"
if user_transit["jupiter_transit_house"] == case_transit["jupiter_transit_house"]
or user_transit["saturn_transit_house"] == case_transit["saturn_transit_house"]
else "different"
)
timing_state["user_transit"] = user_transit
timing_state["case_event_transit"] = case_transit
else:
timing_state["transit_status"] = "not_compared"
matching, dissimilar, total = [], [], 0.0
for name, weight in FEATURE_WEIGHTS.items():
if name not in user or name not in candidate:
continue
if user[name] is None or candidate[name] is None:
continue
total += weight
if user[name] == candidate[name]:
matching.append(name)
else:
dissimilar.append(name)
score = round(sum(FEATURE_WEIGHTS[name] for name in matching) / total, 3) if total else 0.0
compared_vargas = []
if domain == "marriage" and all(user.get(name) is not None and candidate.get(name) is not None for name in ("d9_ascendant", "d9_venus")):
compared_vargas.append("D9")
if domain == "career" and all(user.get(name) is not None and candidate.get(name) is not None for name in ("d10_ascendant", "d10_sun")):
compared_vargas.append("D10")
uncompared = []
if timing_state["status"] == "not_compared":
uncompared.insert(0, "vimshottari_mahadasha")
uncompared.insert(1, "vimshottari_antardasha")
if timing_state["narayana_status"] == "not_compared":
uncompared.append("narayana_dasha")
if timing_state["transit_status"] == "not_compared":
uncompared.append("transit_event_state")
if domain == "marriage" and "D9" not in compared_vargas:
uncompared.insert(0, "D9")
if domain == "career" and "D10" not in compared_vargas:
uncompared.insert(0, "D10")
return {
"score": score,
"matching_factors": matching,
"dissimilar_factors": dissimilar,
"feature_scope": "D1 ascendant, Moon, theme-house lord, Rahu/Ketu axis" + (f", {'/'.join(compared_vargas)}" if compared_vargas else "") + (", Vimshottari MD/AD" if timing_state["status"] != "not_compared" else "") + (", Narayana MD/AD" if timing_state["narayana_status"] != "not_compared" else "") + (", Jupiter/Saturn transit houses" if timing_state["transit_status"] != "not_compared" else ""),
"uncompared_layers": uncompared,
"timing_state": timing_state,
}
def _load_cases(manifest_path: Path) -> list[dict[str, Any]]:
try:
payload = json.loads(manifest_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return []
cases = payload.get("cases") if isinstance(payload, dict) else None
return [case for case in cases if isinstance(case, dict)] if isinstance(cases, list) else []
def _coverage(cases: list[dict[str, Any]], themes: list[str]) -> dict[str, list[str]]:
domains: set[str] = set()
for case in cases:
replay = case.get("replay")
if not isinstance(replay, dict):
continue
replay_status = replay.get("outcome_replay_status")
if replay_status == "replayed" and replay.get("do_not_use_for_prediction") is not True:
pass
elif replay_status == "pending" and replay.get("do_not_use_for_prediction") is True:
pass
else:
continue
for event in case.get("event_outcomes", []):
if isinstance(event, dict) and isinstance(event.get("domain"), str):
domains.add(event["domain"])
available = sorted(domains)
return {
"available_event_domains": available,
"requested_uncovered_domains": sorted(set(themes) - domains),
}
def select_similar_public_cases(
user_chart: dict[str, Any],
themes: list[str],
*,
cases: list[dict[str, Any]] | None = None,
reference_date: str | None = None,
threshold: float = HIGH_SIMILARITY_THRESHOLD,
max_cases: int = 3,
) -> dict[str, Any]:
candidates = cases if cases is not None else (
_load_cases(DEFAULT_MANIFEST) + _load_cases(DEFAULT_CONTEXT_MANIFEST)
)
selected: list[dict[str, Any]] = []
for case in candidates:
replay = case.get("replay")
if not isinstance(replay, dict):
continue
replay_status = replay.get("outcome_replay_status")
if replay_status == "replayed" and replay.get("do_not_use_for_prediction") is not True:
reference_status = "calibration_replayed"
elif replay_status == "pending" and replay.get("do_not_use_for_prediction") is True:
reference_status = "public_context_only"
else:
continue
case_chart = _chart_for_case(case)
if not isinstance(case_chart, dict):
continue
for event in case.get("event_outcomes", []):
if not isinstance(event, dict) or event.get("domain") not in themes:
continue
similarity = _similarity(
user_chart,
case_chart,
event["domain"],
reference_date=reference_date,
case_event_date=event.get("event_date"),
)
if similarity["score"] < threshold:
continue
source = case.get("source") if isinstance(case.get("source"), dict) else {}
event_source = event.get("source") if isinstance(event.get("source"), dict) else {}
subject = case.get("subject") if isinstance(case.get("subject"), dict) else {}
selected.append({
"case_id": case.get("case_id"),
"subject": subject.get("name"),
"domain": event.get("domain"),
"event_type": event.get("event_type"),
"event_date": event.get("event_date"),
"outcome": event.get("outcome"),
"case_source": {"url": source.get("url"), "source_grade": source.get("source_grade")},
"event_source": {"url": event_source.get("url"), "source_grade": event_source.get("source_grade")},
"similarity": similarity,
"reference_only": True,
"reference_status": reference_status,
"difference_notice": "相似仅限列出的 D1 特征;未比较层不得推断为相同。",
})
selected.sort(key=lambda item: (-item["similarity"]["score"], item["case_id"] or ""))
selected = selected[:max_cases]
return {
"status": "high_similarity_public_references_available" if selected else "no_high_similarity_public_reference",
"cases": selected,
"threshold": threshold,
"manifest": [
"references/real_case_calibration/replay_manifest.json",
"references/real_case_calibration/public_context_manifest.json",
],
"public_figures_only": True,
"does_not_predict_user_outcome": True,
"coverage": _coverage(candidates, themes),
"boundary": "公开案例用于比较与理解,不表示用户会复现该事件。",
}
def build_reference_transparency_contract(
chart: dict[str, Any], themes: list[str], *, timing: dict[str, Any] | None = None,
cases: list[dict[str, Any]] | None = None, reference_date: str | None = None,
) -> dict[str, Any]:
timing_contract = build_timing_precision_contract(timing)
return {
"version": "transparent_reference_v1",
"timing_display": {
"claim_status": timing_contract["claim_status"],
"verified_window": "display_with_evidence_scope",
"candidate_windows": "display_with_signals_and_confidence_cap",
"exact_triggers": "display_as_technical_trigger_not_guarantee",
"boundary": timing_contract["boundary"],
},
"external_engine_observations": {
"Local native": {"role": "primary_calculation", "source": "current request calculation contract"},
"VedAstro hosted": {
"role": "external_observation", "deployment_identity": "not_publicly_proven",
"source": "references/oracle/vedastro_contract_arbitration_2026_07_17.json",
},
"Xalen": {"role": "formula_isolation_observation", "source": "references/oracle/xalen_fourth_oracle_comparison_2026_07_17.json"},
"jyotishyamitra": {"role": "independent_observation", "source": "references/oracle/jyotishyamitra_steve_jobs_probe_2026_07_18.json"},
},
"method_variants": {
"display": "show_parallel_methods_with_sources",
"source": "references/oracle/xalen_formula_unit_attribution_2026_07_17.json",
"boundary": "流派/公式差异并列展示;不以单一引擎多数投票决定真值。",
},
"similar_public_cases": select_similar_public_cases(
chart,
themes,
cases=cases,
reference_date=reference_date,
),
}