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.horary_observation import build_horary_observation from scripts.rectification.refinement_packet import build_refinement_packet from scripts.rectification.scoring_service import precision_weight from scripts.rectification.sealed_holdout import holdout_passed, load_sealed_minute_holdout from scripts.rectification_policy import ( MAX_CONFIRMATION_WIDTH_MINUTES, MIN_CONFIRMATION_DOMAINS, MIN_CONFIRMATION_EVENTS, MIN_CONFIRMATION_MARGIN_PERCENT, ) 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") POLICY_SKIPPED_LAYERS = frozenset({"KP_cusps"}) _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 成就分盘", "本轮已对照学业或被委以责任的变化。"), "d3-drekkana": ("D3 兄弟分盘", "本轮已对照兄弟姐妹主题。"), "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 _clock_minutes(value: Any) -> int | None: text = str(value or "")[:5] if len(text) != 5 or text[2] != ":": return None try: hour = int(text[:2]) minute = int(text[3:]) except ValueError: return None if hour > 23 or minute > 59: return None return hour * 60 + minute def indistinguishable_width_minutes(candidates: Sequence[dict[str, Any]]) -> int: if not candidates: return 0 ranked = sorted(candidates, key=lambda row: int(row.get("rank") or 0)) top = ranked[0] minutes = [value for value in (_clock_minutes(row.get("time")) for row in ranked) if value is not None] span = (max(minutes) - min(minutes) + 1) if minutes else 0 tied = int(top.get("tied_minute_count") or 1) return max(tied, span, 1) def vedastro_audit_row(status: str) -> dict[str, str]: if status == "passed": return { "technique": "VedAstro 分钟级校验", "status": "executed", "note": "官方分钟敏感校验已通过。仍不能单独确认唯一分钟。", } if status == "failed": return { "technique": "VedAstro 分钟级校验", "status": "blocked", "note": "官方分钟敏感校验未能区分相邻分钟，不能写确认。", } return { "technique": "VedAstro 分钟级校验", "status": "blocked", "note": "官方分钟敏感校验尚未跑通。未调用不等于失败，但缺这一层不能写确认。", } def unique_minute_audit_row(allowed: bool) -> dict[str, str]: if allowed: return { "technique": "唯一分钟确认", "status": "executed", "note": "确认门已允许。只有用户明确同意才能写已确认校正时间。", } return { "technique": "唯一分钟确认", "status": "blocked", "note": "采用不等于确认唯一分钟。", } def rewrite_confirmation_audit_rows(receipt: dict[str, Any]) -> None: rows = receipt.get("technique_audit_table") if not isinstance(rows, list): return exact = (receipt.get("gates") or {}).get("exact_confirmation") or {} vedastro_status = str(exact.get("external_validation_status") or "not_evaluated") allowed = receipt.get("confirmation_allowed") is True replacements = { "VedAstro 分钟级校验": vedastro_audit_row(vedastro_status), "唯一分钟确认": unique_minute_audit_row(allowed), } receipt["technique_audit_table"] = [ replacements.get(str(row.get("technique")), row) if isinstance(row, dict) else row for row in rows ] def apply_confirmation_decision(receipt: dict[str, Any]) -> dict[str, Any]: gates = receipt.setdefault("gates", {}) exact = gates.setdefault("exact_confirmation", {}) vedastro_status = str(exact.get("external_validation_status") or "not_evaluated") if vedastro_status not in {"passed", "failed", "not_evaluated"}: vedastro_status = "not_evaluated" exact["external_validation_status"] = vedastro_status holdout = load_sealed_minute_holdout() engine_granted = exact.get("engine_granted") is True adjacent_passed = exact.get("adjacent_passed") is True vedastro_ok = vedastro_status == "passed" holdout_ok = holdout_passed(holdout) confirmation_allowed = bool(engine_granted and adjacent_passed and vedastro_ok and holdout_ok) reasons = [str(item) for item in (receipt.get("confirmation_reasons") or []) if str(item)] for flag, reason in ( (engine_granted, "engine_exact_confirmation_not_granted"), (vedastro_ok, "external_validation_not_passed"), (adjacent_passed, "adjacent_minutes_indistinguishable"), (holdout_ok, "public_aa_holdout_not_ready"), ): if flag: reasons = [item for item in reasons if item != reason] elif reason not in reasons: reasons.append(reason) receipt["confirmation_allowed"] = confirmation_allowed receipt["confirm_allowed"] = confirmation_allowed receipt["unique_minute_claim"] = False exact["passed"] = confirmation_allowed exact["fail_closed"] = True exact["external_validation_status"] = vedastro_status exact["holdout"] = holdout exact["reason"] = ( "confirmation_allowed" if confirmation_allowed else "engine_and_external_validation_must_explicitly_pass" ) receipt["confirmation_reasons"] = reasons receipt["reasons"] = [ *list(receipt.get("acceptance_reasons") or []), *reasons, ] natal = receipt.get("natal_recast") if isinstance(natal, dict): natal["confirmation_allowed"] = False natal["unique_minute_claim"] = False rewrite_confirmation_audit_rows(receipt) return receipt def apply_vedastro_minute_sensitive_to_receipt( receipt: dict[str, Any], status: str, *, summary: dict[str, Any] | None = None, ) -> dict[str, Any]: exact = receipt.setdefault("gates", {}).setdefault("exact_confirmation", {}) normalized = status if status in {"passed", "failed", "not_evaluated"} else "not_evaluated" exact["external_validation_status"] = normalized if summary is not None: exact["vedastro_minute_sensitive"] = summary return apply_confirmation_decision(receipt) def build_technique_audit( built: dict[str, Any], *, house_table: dict[str, Any] | None, vedastro_status: str = "not_evaluated", confirmation_allowed: bool = False, ) -> 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(_kp_audit_row(built)) rows.extend(( vedastro_audit_row(vedastro_status), unique_minute_audit_row(confirmation_allowed), )) return rows def _kp_audit_row(built: dict[str, Any]) -> dict[str, str]: executed = False for context in built.get("static_contexts") or []: if not isinstance(context, dict): continue feature = context.get("feature") snapshot = feature.get("kp_cusps") if isinstance(feature, dict) else None if isinstance(snapshot, dict) and snapshot.get("status") == "executed": executed = True break if executed: return { "technique": "KP 宫头", "status": "executed", "note": "已按 Swiss Ephemeris Placidus + Krishnamurti 观察 12 宫头；不计分，不参与提出门或确认门。", } return { "technique": "KP 宫头", "status": "blocked", "note": "Swiss Ephemeris Placidus 宫头无法计算或尚未执行。KP 观察不计分，不挡提出门。", } 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 _actionable_missing_layers(layers: Any) -> list[str]: return sorted({ str(layer) for layer in (layers or []) if str(layer) not in POLICY_SKIPPED_LAYERS }) 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) skipped_layers = sorted( str(layer) for layer in (built.get("missing_layers") or []) if str(layer) in POLICY_SKIPPED_LAYERS ) actionable_missing_layers = _actionable_missing_layers(built.get("missing_layers")) required_layers = _gate( not actionable_missing_layers, missing_layers=actionable_missing_layers, skipped_by_policy=skipped_layers, ) # 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") 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") width = indistinguishable_width_minutes(candidate_decisions) adjacent_passed = unique_top["passed"] and width <= MAX_CONFIRMATION_WIDTH_MINUTES confirmation_event_quality = len(scoreable_events) >= MIN_CONFIRMATION_EVENTS confirmation_domain_quality = len(domains) >= MIN_CONFIRMATION_DOMAINS confirmation_margin = margin >= Decimal(MIN_CONFIRMATION_MARGIN_PERCENT) dasha_conflict = packet["dasha_agreement"]["status"] == "conflict" if not confirmation_event_quality: confirmation_reasons.append("insufficient_confirmation_events") if not confirmation_domain_quality: confirmation_reasons.append("insufficient_confirmation_domains") if not confirmation_margin: confirmation_reasons.append("insufficient_confirmation_margin") if not adjacent_passed: confirmation_reasons.append("adjacent_minutes_indistinguishable") propose_allowed = all(( acceptance_allowed, unique_top["passed"], diagnostic_quality["passed"], required_layers["passed"], confirmation_event_quality, confirmation_domain_quality, adjacent_passed, )) engine_granted = all(( propose_allowed, confirmation_margin, not dasha_conflict, )) if not engine_granted: confirmation_reasons.append("engine_exact_confirmation_not_granted") exact_confirmation = { "passed": False, "fail_closed": True, "engine_granted": engine_granted, "adjacent_passed": adjacent_passed, "indistinguishable_width_minutes": width, "max_confirmation_width_minutes": MAX_CONFIRMATION_WIDTH_MINUTES, "external_validation_status": "not_evaluated", "required_scoreable_events": MIN_CONFIRMATION_EVENTS, "required_scoreable_domains": MIN_CONFIRMATION_DOMAINS, "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, "propose_allowed": propose_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, vedastro_status="not_evaluated", confirmation_allowed=False, ) 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"], "horary_observation": build_horary_observation(request), "unique_minute_claim": False, }) return apply_confirmation_decision(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