from __future__ import annotations from collections.abc import Sequence from decimal import ROUND_FLOOR, ROUND_HALF_UP, Decimal from typing import Any from uuid import NAMESPACE_URL, uuid5 from scripts.active_rectification_events import CandidateScoreRow from scripts.rectification.contracts import ( EVENT_CONTRACT_VERSION, RectificationRequest, is_primary_scoreable_event, is_scoreable_event, ) from scripts.rectification.house_table import compact_house_table_from_contexts from scripts.rectification.refinement_packet import build_refinement_packet from scripts.rectification.scoring_service import precision_weight POLICY_VERSION = "rectification-candidate-policy-v2" RECEIPT_VERSION = "candidate-decision-receipt-v2" EXECUTION_LEDGER_VERSION = "rectification-execution-ledger-v2" SCORE_QUANTUM = Decimal("0.0001") TIE_ABSOLUTE_TOLERANCE = Decimal("0.0001") MIN_ACCEPTANCE_EVENTS = 3 MIN_ACCEPTANCE_DOMAINS = 2 MIN_DATE_QUALITY_MEAN = Decimal("0.65") MIN_DIAGNOSTIC_RETENTION = Decimal("0.75") MIN_ACCEPTANCE_MARGIN_PERCENT = Decimal("10") _BAD_DATE_RELIABILITY = frozenset({"low", "uncertain", "unreliable"}) _CLEAR_DATE_CONFLICT = frozenset({"", "none", "resolved", "no_conflict"}) def _decimal(value: Any, default: str = "0") -> Decimal: if isinstance(value, bool): return Decimal(default) try: return Decimal(str(value)) except Exception: return Decimal(default) _AUDIT_LABELS = { "d1-rashi": ("D1 本命盘", "本轮已按该分钟重算本命宫位。"), "d2-hora": ("D2 财帛分盘", "本轮已对照财帛主题。"), "d4-chaturthamsha": ("D4 迁移分盘", "本轮已对照居所或迁移。"), "d5-panchamsha": ("D5 成就分盘", "本轮已对照学业或被委以责任的变化。"), "d7-saptamsha": ("D7 子女分盘", "本轮已对照子女或伴侣细节。"), "d9-navamsa": ("D9 婚姻分盘", "本轮已对照关系主题，未给类型标签。"), "d10-dashamsa": ("D10 事业分盘", "本轮已对照事业主题，未给类型标签。"), "d11-labhamsha": ("D11 收益分盘", "本轮已对照收益主题。"), "d12-dwadashamsha": ("D12 父母分盘", "本轮已对照家人主题。"), "d24-chaturvimshamsha": ("D24 教育分盘", "本轮已对照学业主题。"), "d30-trimshamsha": ("D30 健康压力分盘", "本轮已对照健康压力主题。"), "vimshottari-dasha": ("Vimshottari", "本轮已对照主限。"), "narayana-dasha": ("Narayana", "本轮已对照分盘大运。"), "gochara": ("Gochara", "本轮已做受控行运辅助对照。"), "ashtakavarga": ("Ashtakavarga", "本轮已做 Ashtakavarga 辅助对照。"), "shadbala": ("Shadbala", "本轮已做已核验的 Shadbala 分量辅助对照。"), "arudha-pada": ("Arudha Pada", "本轮已做 Arudha 辅助对照。"), "functional-benefic-malefic": ("功能吉凶星", "本轮已叠加本命功能吉凶星。"), } def natal_recast_copy(time: str, lagna: str) -> dict[str, Any]: return { "time": time[:5], "lagna": lagna, "user_meaning": ( f"本命宫位已按 {time[:5]} 重算（上升 {lagna}）。" "下面是本轮实际执行的技法，不能当作唯一分钟确认。" ), "unique_minute_claim": False, "confirmation_allowed": False, } def _executed_public_methods(built: dict[str, Any]) -> list[str]: methods: set[str] = set() for contributions in (built.get("matrix") or {}).values(): if not isinstance(contributions, dict): continue for cell in contributions.values(): if not isinstance(cell, dict): continue for layer in cell.get("technique_layers") or []: if layer in _AUDIT_LABELS: methods.add(str(layer)) for rule in cell.get("rule_ids") or []: text = str(rule) if text.startswith("vim_"): methods.add("vimshottari-dasha") elif text.startswith("narayana_"): methods.add("narayana-dasha") elif "functional_benefic" in text or "functional_malefic" in text: methods.add("functional-benefic-malefic") elif text.startswith("gochara") or "controlled_transit" in text: methods.add("gochara") elif "ashtakavarga" in text: methods.add("ashtakavarga") elif "shadbala" in text: methods.add("shadbala") elif "arudha" in text: methods.add("arudha-pada") return [key for key in _AUDIT_LABELS if key in methods] def build_technique_audit( built: dict[str, Any], *, house_table: dict[str, Any] | None, ) -> list[dict[str, str]]: executed = set(_executed_public_methods(built)) if house_table: executed.add("d1-rashi") rows: list[dict[str, str]] = [] for method in _AUDIT_LABELS: if method not in executed: continue label, note = _AUDIT_LABELS[method] rows.append({"technique": label, "status": "executed", "note": note}) rows.append({ "technique": "KP 宫头", "status": "blocked", "note": "KP 宫头本轮未计算。", }) rows.extend(( { "technique": "VedAstro 分钟级校验", "status": "blocked", "note": "官方分钟级校验尚未评估。", }, { "technique": "唯一分钟确认", "status": "blocked", "note": "采用不等于确认唯一分钟。", }, )) return rows def _quantized_score(row: CandidateScoreRow) -> Decimal: return _decimal(row.get("score")).quantize(SCORE_QUANTUM, rounding=ROUND_HALF_UP) def _relative_support(scores: Sequence[Decimal]) -> list[int]: if not scores: return [] weights = [max(score, Decimal(0)) for score in scores] total = sum(weights, Decimal(0)) if total == 0: base, remainder = divmod(100, len(scores)) return [base + (1 if index < remainder else 0) for index in range(len(scores))] exact = [weight * Decimal(100) / total for weight in weights] floors = [int(value.to_integral_value(rounding=ROUND_FLOOR)) for value in exact] remaining = 100 - sum(floors) order = sorted( range(len(scores)), key=lambda index: (-(exact[index] - Decimal(floors[index])), index), ) for index in order[:remaining]: floors[index] += 1 return floors def build_candidate_decisions( rows: Sequence[CandidateScoreRow], *, result_id: str, ) -> list[dict[str, Any]]: ranked = sorted(rows, key=lambda row: (-_quantized_score(row), row["time"])) public_rows = ranked[:3] supports = _relative_support([_quantized_score(row) for row in public_rows]) all_scores = [_quantized_score(row) for row in ranked] decisions = [] for index, row in enumerate(public_rows): score = _quantized_score(row) tied_minute_count = sum( abs(score - other) <= TIE_ABSOLUTE_TOLERANCE for other in all_scores ) decisions.append({ "candidate_id": str(uuid5(NAMESPACE_URL, f"{POLICY_VERSION}:{result_id}:{row['time']}")), "rank": index + 1, "time": row["time"], "relative_support": supports[index], "tied_minute_count": tied_minute_count, }) return decisions def _gate(passed: bool, **details: Any) -> dict[str, Any]: return {"passed": passed, **details} def _date_quality(events: Sequence[dict[str, Any]]) -> dict[str, Any]: weights = [_decimal(precision_weight(str(event["precision"]))) for event in events] total = sum(weights, Decimal(0)) mean = total / Decimal(len(weights)) if weights else Decimal(0) low_reliability = sorted( event["id"] for event in events if str(event.get("date_reliability") or "").strip().lower() in _BAD_DATE_RELIABILITY ) unresolved_conflicts = sorted( event["id"] for event in events if str(event.get("date_conflict_status") or "").strip().lower() not in _CLEAR_DATE_CONFLICT ) passed = bool(events) and mean >= MIN_DATE_QUALITY_MEAN and not low_reliability and not unresolved_conflicts return _gate( passed, precision_weight_total=float(total), precision_weight_mean=float(mean.quantize(SCORE_QUANTUM, rounding=ROUND_HALF_UP)), minimum_precision_weight_mean=float(MIN_DATE_QUALITY_MEAN), low_reliability_event_ids=low_reliability, unresolved_conflict_event_ids=unresolved_conflicts, ) def _diagnostic_quality(diagnostics: dict[str, Any]) -> dict[str, Any]: retention_names = ( "leave_one_event_out_retention_rate", "leave_one_domain_out_retention_rate", "date_sensitivity_retention_rate", ) retentions = {name: _decimal(diagnostics.get(name)) for name in retention_names} margin = _decimal(diagnostics.get("primary_secondary_margin_percent")) passed = ( all(value >= MIN_DIAGNOSTIC_RETENTION for value in retentions.values()) and margin >= MIN_ACCEPTANCE_MARGIN_PERCENT ) return _gate( passed, minimum_retention=float(MIN_DIAGNOSTIC_RETENTION), minimum_margin_percent=float(MIN_ACCEPTANCE_MARGIN_PERCENT), margin_percent=float(margin), **{name: float(value) for name, value in retentions.items()}, ) def build_decision_receipt( request: RectificationRequest, candidate_decisions: Sequence[dict[str, Any]], built: dict[str, Any], diagnostics: dict[str, Any], ) -> dict[str, Any]: scoreable_events = [event for event in request["events"] if is_primary_scoreable_event(event)] domains = sorted({event["domain"] for event in scoreable_events}) candidate_presence = _gate(bool(candidate_decisions), candidate_count=len(candidate_decisions)) event_quality = _gate( len(scoreable_events) >= MIN_ACCEPTANCE_EVENTS, scoreable_event_count=len(scoreable_events), minimum=MIN_ACCEPTANCE_EVENTS, ) domain_diversity = _gate( len(domains) >= MIN_ACCEPTANCE_DOMAINS, scoreable_domain_count=len(domains), minimum=MIN_ACCEPTANCE_DOMAINS, domains=domains, ) date_quality = _date_quality(scoreable_events) top_tied_count = candidate_decisions[0]["tied_minute_count"] if candidate_decisions else 0 unique_top = _gate(top_tied_count == 1, tied_minute_count=top_tied_count) diagnostic_quality = _diagnostic_quality(diagnostics) required_layers = _gate( not built.get("missing_layers"), missing_layers=sorted(built.get("missing_layers") or []), ) # Adoption is the session result when a representative time exists. # Unique-top and diagnostic stability still block confirmation, not accept. acceptance_allowed = all(( candidate_presence["passed"], event_quality["passed"], domain_diversity["passed"], date_quality["passed"], )) margin = _decimal(diagnostics.get("primary_secondary_margin_percent")) if acceptance_allowed and margin >= Decimal("20"): overall_confidence = "high" elif acceptance_allowed: overall_confidence = "medium" else: overall_confidence = "low" acceptance_reasons: list[str] = [] for passed, reason in ( (candidate_presence["passed"], "no_candidates"), (event_quality["passed"], "insufficient_events"), (domain_diversity["passed"], "insufficient_domain_diversity"), (date_quality["passed"], "low_date_quality"), ): if not passed: acceptance_reasons.append(reason) confirmation_reasons = [] if not unique_top["passed"]: confirmation_reasons.append("tied_top_score") if not diagnostic_quality["passed"]: confirmation_reasons.append("insufficient_diagnostic_stability") if not required_layers["passed"]: confirmation_reasons.append("missing_mandatory_layers") confirmation_reasons.extend(("engine_exact_confirmation_not_granted", "external_validation_not_passed")) reasons = [*acceptance_reasons, *confirmation_reasons] representative = candidate_decisions[0] if candidate_decisions else None packet = build_refinement_packet( request, built, representative_time=representative["time"] if representative else None, candidate_times=[item["time"] for item in candidate_decisions], ) if packet["dasha_agreement"]["status"] == "conflict": if overall_confidence == "high": overall_confidence = "medium" elif overall_confidence == "medium": overall_confidence = "low" reasons.append("vimshottari_narayana_conflict") confirmation_reasons.append("vimshottari_narayana_conflict") exact_confirmation = { "passed": False, "fail_closed": True, "engine_granted": False, "external_validation_status": "not_evaluated", "required_scoreable_events": 4, "required_scoreable_domains": 3, "reason": "engine_and_external_validation_must_explicitly_pass", } receipt = { "receipt_version": RECEIPT_VERSION, "contract_version": "v2", "event_contract_version": EVENT_CONTRACT_VERSION, "policy_version": POLICY_VERSION, "decision_policy_version": POLICY_VERSION, "display_allowed": bool(candidate_decisions), "selection_allowed": acceptance_allowed, "acceptance_allowed": acceptance_allowed, "confirmation_allowed": False, "accept_allowed": acceptance_allowed, "confirm_allowed": False, "representative_candidate_id": representative["candidate_id"] if representative else None, "representative_time": representative["time"] if representative else None, "overall_confidence": overall_confidence, "margin_percent": float(margin), "reasons": reasons, "acceptance_reasons": acceptance_reasons, "confirmation_reasons": confirmation_reasons, "tie_policy": { "score_quantum": float(SCORE_QUANTUM), "absolute_tolerance": float(TIE_ABSOLUTE_TOLERANCE), "rounding": "ROUND_HALF_UP", }, "gates": { "candidate_presence": candidate_presence, "event_quality": event_quality, "domain_diversity": domain_diversity, "date_quality": date_quality, "unique_top": unique_top, "diagnostic_quality": diagnostic_quality, "required_layers": required_layers, "exact_confirmation": exact_confirmation, }, } house_tables_by_time: dict[str, dict[str, Any]] = {} for decision in candidate_decisions: table = compact_house_table_from_contexts(built.get("static_contexts"), decision.get("time")) if table: house_tables_by_time[table["time"]] = table house_table = house_tables_by_time.get(representative["time"] if representative else "") or compact_house_table_from_contexts( built.get("static_contexts"), representative["time"] if representative else None, ) if house_table: receipt["house_table"] = house_table recast = natal_recast_copy(house_table["time"], house_table["lagna"]) receipt["natal_recast"] = recast if house_tables_by_time: receipt["house_tables_by_time"] = house_tables_by_time receipt["technique_audit_table"] = build_technique_audit(built, house_table=house_table) receipt.update({ "window_scan": packet["window_scan"], "event_dasha_ledger": packet["event_dasha_ledger"], "dasha_agreement": packet["dasha_agreement"], "lagna_contrast": packet["lagna_contrast"], "nakshatra_boundary": packet["nakshatra_boundary"], "precision_stage": packet["precision_stage"], "oos_blind_prompts": packet["oos_blind_prompts"], "unique_minute_claim": False, }) return receipt def build_execution_ledger( request: RectificationRequest, built: dict[str, Any], diagnostics: dict[str, Any], candidate_decisions: Sequence[dict[str, Any]], ) -> list[dict[str, Any]]: matrix = built.get("matrix") or {} date_sensitivity = { item.get("event_id"): item for item in built.get("date_sensitivity") or [] if isinstance(item, dict) } entries: list[dict[str, Any]] = [] all_layers: set[str] = set() for event in request["events"]: candidates = matrix.get(event["id"], {}) layers = sorted({ layer for contribution in candidates.values() for layer in contribution.get("technique_layers", []) }) all_layers.update(layers) sensitivity = date_sensitivity.get(event["id"], {}) scoreable = is_scoreable_event(event) entries.append({ "ledger_version": EXECUTION_LEDGER_VERSION, "stage": "event_scoring", "status": "executed" if scoreable and candidates else "not_executed" if scoreable else "retained_not_scored", "source": "python-engine", "event_id": event["id"], "domain": event["domain"], "event_kind": event["event_kind"], "date_precision": event["precision"], "precision_weight": precision_weight(event["precision"]), "sampled_date_count": len(sensitivity.get("sample_dates") or []), "candidate_count": len(candidates), "technique_layers": layers, }) for layer in sorted(all_layers): entries.append({ "ledger_version": EXECUTION_LEDGER_VERSION, "stage": "technique_layer", "method": layer, "status": "executed", "source": "python-engine", }) entries.extend(( { "ledger_version": EXECUTION_LEDGER_VERSION, "stage": "candidate_ranking", "status": "executed" if candidate_decisions else "not_executed", "source": "python-decision-policy", "candidate_count": len(candidate_decisions), }, { "ledger_version": EXECUTION_LEDGER_VERSION, "stage": "diagnostics", "status": "executed" if diagnostics else "not_executed", "source": "python-engine", "metrics": sorted(diagnostics), }, )) return entries