from __future__ import annotations import unittest from scripts.rectification.decision_policy import build_candidate_decisions, build_decision_receipt from scripts.rectification.refinement_packet import ( build_refinement_packet, cluster_scan, dasha_agreement, match_level, precision_stage, window_scan, ) def feature( time: str, *, d1: int | None = None, d9: int | None = None, d10: int | None = None, d4: int | None = None, d5: int | None = None, d7: int | None = None, d12: int | None = None, d24: int | None = None, d2: int | None = None, d11: int | None = None, d30: int | None = None, degree: float | None = None, pada: int | None = None, hora: int | None = None, ghati: int | None = None, bhava: int | None = None, pranapada: int | None = None, ) -> dict: vargas = {} if d9 is not None: vargas["D9"] = d9 if d10 is not None: vargas["D10"] = d10 if d4 is not None: vargas["D4"] = d4 if d5 is not None: vargas["D5"] = d5 if d7 is not None: vargas["D7"] = d7 if d12 is not None: vargas["D12"] = d12 if d24 is not None: vargas["D24"] = d24 if d2 is not None: vargas["D2"] = d2 if d11 is not None: vargas["D11"] = d11 if d30 is not None: vargas["D30"] = d30 row: dict = {"time": time, "varga_ascendants": vargas} if d1 is not None: row["ascendant_sign_index"] = d1 if degree is not None: row["ascendant_degree"] = degree if pada is not None: row["pada_index"] = pada if hora is not None: row["hora_sign_index"] = hora if ghati is not None: row["ghati_sign_index"] = ghati if bhava is not None: row["bhava_sign_index"] = bhava if pranapada is not None: row["pranapada_sign_index"] = pranapada return {"feature": row} def request_events() -> dict: return { "events": [ {"id": "00000000-0000-4000-8000-000000000001", "domain": "career", "summary": "入职", "event_kind": "career_entry", "precision": "day", "date_start": "2016-09-15", "date_end": "2016-09-15"}, {"id": "00000000-0000-4000-8000-000000000002", "domain": "relationship", "summary": "开始一段关系", "event_kind": "relationship_start", "precision": "day", "date_start": "2018-03-01", "date_end": "2018-03-01"}, {"id": "00000000-0000-4000-8000-000000000003", "domain": "education", "summary": "毕业", "event_kind": "education_completion", "precision": "day", "date_start": "2015-06-01", "date_end": "2015-06-01"}, ] } class RefinementPacketTest(unittest.TestCase): def test_window_scan_emits_d9_sign_names_for_type_table_report(self): built = { "static_contexts": [ feature("05:13", d1=1, d9=1, d10=4), feature("05:14", d1=1, d9=7, d10=4), ] } scan = window_scan(built) self.assertEqual(scan["d9_lagna_count"], 2) self.assertEqual(scan["d10_lagna_count"], 1) self.assertEqual(scan["d1_lagna_count"], 1) self.assertEqual(scan["d9_sign_names"], ["金牛座", "天蝎座"]) self.assertEqual(scan["d10_sign_names"], ["狮子座"]) self.assertEqual(scan["transitions"], [{ "layer": "d9", "at": "05:14", "user_meaning": "D9 在 05:14 发生变化", }]) encoded = str(scan) self.assertNotIn("Aries", encoded) self.assertFalse(scan["unique_minute_claim"]) self.assertFalse(scan["confirmation_allowed"]) def test_match_level_and_dual_dasha_conflict(self): self.assertEqual(match_level(["vim_md_domain_house"]), "strong") self.assertEqual(match_level(["vim_ad_domain_lord"]), "medium") self.assertEqual(match_level(["no_domain_activation"]), "none") built = { "matrix": { "00000000-0000-4000-8000-000000000001": { "05:13": {"points": 8, "rule_ids": ["vim_md_domain_house"]}, "05:14": {"points": 1, "rule_ids": ["vim_ad_domain_house"]}, }, "00000000-0000-4000-8000-000000000002": { "05:13": {"points": 1, "rule_ids": ["narayana_ad_domain_house"]}, "05:14": {"points": 9, "rule_ids": ["narayana_md_domain_house"]}, }, } } agreement = dasha_agreement(built, ["05:13", "05:14"]) self.assertEqual(agreement["status"], "conflict") self.assertEqual(agreement["vimshottari_top"], "05:13") self.assertEqual(agreement["narayana_top"], "05:14") self.assertIn("冲突", agreement["user_meaning"]) def test_precision_stage_walks_d1_then_d9_then_ready(self): self.assertEqual(precision_stage({"d1_candidates_differ": True}, 2)["current"], "lagna_frame") self.assertEqual(precision_stage({"d9_candidates_differ": True}, 2)["current"], "d9_refine") self.assertEqual(precision_stage({"d10_candidates_differ": True}, 2)["current"], "d10_refine") self.assertEqual(precision_stage({"d4_candidates_differ": True}, 2)["current"], "d4_refine") self.assertEqual(precision_stage({"d5_candidates_differ": True}, 2)["current"], "d5_refine") self.assertEqual( precision_stage({"d24_candidates_differ": True, "d5_candidates_differ": False}, 2)["current"], "d5_refine", ) self.assertNotEqual(precision_stage({"d4_candidates_differ": True}, 2)["current"], "theme_refine") self.assertIn("搬家", precision_stage({"d4_candidates_differ": True}, 2)["user_meaning"]) self.assertNotIn("家人", precision_stage({"d4_candidates_differ": True}, 2)["user_meaning"]) ready = precision_stage({}, 3) self.assertEqual(ready["current"], "ready_to_adopt") self.assertFalse(ready["unique_minute_claim"]) def test_precision_stage_uses_candidate_cluster_not_full_declared_range(self): built = { "static_contexts": [ feature("12:00", d1=3, d9=1, d10=4), feature("13:37", d1=4, d9=1, d10=4), feature("15:50", d1=4, d9=1, d10=4), feature("15:51", d1=4, d9=1, d10=4), feature("16:03", d1=5, d9=1, d10=4), ] } scan = window_scan(built) self.assertEqual(scan["d1_lagna_count"], 3) self.assertTrue(scan["d1_candidates_differ"]) cluster = cluster_scan(built, ["15:50", "15:51"], "15:50", 14) self.assertEqual(cluster["d1_lagna_count"], 1) self.assertFalse(cluster["d1_candidates_differ"]) packet = build_refinement_packet( request_events(), built, representative_time="15:50", candidate_times=["15:50", "15:51"], cluster_width_minutes=14, ) self.assertEqual(packet["window_scan"]["d1_lagna_count"], 3) self.assertNotEqual(packet["precision_stage"]["current"], "lagna_frame") def test_window_scan_reports_d24_and_fine_minute_changes_without_unique_claim(self): scan = window_scan({ "static_contexts": [ feature("05:13", d1=1, d5=3, d24=4, pada=10, hora=2, ghati=6), feature("05:14", d1=1, d5=3, d24=8, pada=11, hora=3, ghati=6), ] }) self.assertEqual(scan["d24_lagna_count"], 2) self.assertTrue(scan["d24_candidates_differ"]) self.assertFalse(scan["d5_candidates_differ"]) self.assertEqual(scan["pada_count"], 2) self.assertTrue(scan["pada_candidates_differ"]) self.assertTrue(scan["hora_candidates_differ"]) self.assertFalse(scan["ghati_candidates_differ"]) meanings = [item["user_meaning"] for item in scan["transitions"]] self.assertIn("D24 在 05:14 发生变化", meanings) self.assertIn("Nakshatra pada 在 05:14 发生变化", meanings) self.assertIn("Hora Lagna 在 05:14 发生变化", meanings) self.assertNotIn("Ghati Lagna 在 05:14 发生变化", meanings) self.assertFalse(scan["unique_minute_claim"]) self.assertFalse(scan["confirmation_allowed"]) self.assertEqual(precision_stage(scan, 2)["current"], "d5_refine") def test_window_scan_reports_finance_health_and_bhava_without_delaying_adopt(self): scan = window_scan({ "static_contexts": [ feature("05:13", d1=1, d2=2, d11=4, d30=6, bhava=1, hora=2, ghati=6, pranapada=8), feature("05:14", d1=1, d2=3, d11=4, d30=7, bhava=2, hora=2, ghati=6, pranapada=8), ] }) self.assertTrue(scan["d2_candidates_differ"]) self.assertFalse(scan["d11_candidates_differ"]) self.assertTrue(scan["d30_candidates_differ"]) self.assertTrue(scan["bhava_candidates_differ"]) self.assertFalse(scan["pranapada_candidates_differ"]) meanings = [item["user_meaning"] for item in scan["transitions"]] self.assertIn("D2 在 05:14 发生变化", meanings) self.assertIn("D30 在 05:14 发生变化", meanings) self.assertIn("Bhava Lagna 在 05:14 发生变化", meanings) self.assertNotIn("Pranapada Lagna 在 05:14 发生变化", meanings) self.assertEqual(precision_stage(scan, 2)["current"], "ready_to_adopt") self.assertFalse(scan["unique_minute_claim"]) pranapada_scan = window_scan({ "static_contexts": [ feature("05:13", d1=1, d11=4, pranapada=8), feature("05:14", d1=1, d11=5, pranapada=9), ] }) self.assertTrue(pranapada_scan["d11_candidates_differ"]) self.assertTrue(pranapada_scan["pranapada_candidates_differ"]) self.assertIn( "Pranapada Lagna 在 05:14 发生变化", [item["user_meaning"] for item in pranapada_scan["transitions"]], ) self.assertEqual(precision_stage(pranapada_scan, 2)["current"], "ready_to_adopt") def test_packet_ledger_oos_and_nakshatra_without_scores(self): request = request_events() built = { "static_contexts": [ feature("05:13", d1=1, d9=1, d10=4, degree=13.1), feature("05:14", d1=2, d9=1, d10=4, degree=13.2), ], "matrix": { request["events"][0]["id"]: { "05:13": {"points": 6, "rule_ids": ["vim_md_domain_house", "narayana_md_domain_house"]}, "05:14": {"points": 1, "rule_ids": ["no_domain_activation"]}, }, request["events"][1]["id"]: { "05:13": {"points": 2, "rule_ids": ["vim_ad_domain_house"]}, "05:14": {"points": 2, "rule_ids": ["vim_ad_domain_house"]}, }, request["events"][2]["id"]: { "05:13": {"points": 0, "rule_ids": ["no_domain_activation"]}, "05:14": {"points": 0, "rule_ids": ["no_domain_activation"]}, }, }, } packet = build_refinement_packet( request, built, representative_time="05:13", candidate_times=["05:13", "05:14"], ) self.assertEqual(packet["event_dasha_ledger"][0]["match"], "strong") self.assertIn("入职", packet["event_dasha_ledger"][0]["user_meaning"]) self.assertNotIn("points", str(packet["event_dasha_ledger"])) self.assertEqual(packet["lagna_contrast"]["intervals"][0]["lagna"], "金牛座") self.assertTrue(packet["nakshatra_boundary"]["near_boundary"]) self.assertEqual(packet["oos_blind_prompts"][0]["domain"], "family") self.assertFalse(packet["oos_blind_prompts"][0]["used_for_scoring"]) self.assertFalse(packet["confirmation_allowed"]) encoded = str(packet["window_scan"]) self.assertIn("金牛座", encoded) self.assertNotIn("热情冲动", encoded) self.assertIn("事件吻合率", str(packet["event_fit_rate"])) self.assertFalse(packet["event_fit_rate"]["unique_minute_claim"]) def test_decision_receipt_downgrades_confidence_on_dasha_conflict(self): request = request_events() rows = [ {"time": "05:13", "score": 20, "evidence": [], "missing_layers": []}, {"time": "05:14", "score": 8, "evidence": [], "missing_layers": []}, ] decisions = build_candidate_decisions(rows, result_id="00000000-0000-4000-8000-000000000099") built = { "missing_layers": [], "static_contexts": [feature("05:13", d1=1, d9=1, d10=4), feature("05:14", d1=1, d9=1, d10=4)], "matrix": { request["events"][0]["id"]: { "05:13": {"points": 8, "rule_ids": ["vim_md_domain_house"]}, "05:14": {"points": 1, "rule_ids": ["vim_ad_domain_house"]}, }, request["events"][1]["id"]: { "05:13": {"points": 1, "rule_ids": ["narayana_ad_domain_house"]}, "05:14": {"points": 9, "rule_ids": ["narayana_md_domain_house"]}, }, request["events"][2]["id"]: { "05:13": {"points": 1, "rule_ids": ["vim_md_domain_varga"]}, "05:14": {"points": 1, "rule_ids": ["vim_md_domain_varga"]}, }, }, } diagnostics = { "leave_one_event_out_retention_rate": 1, "leave_one_domain_out_retention_rate": 1, "date_sensitivity_retention_rate": 1, "primary_secondary_margin_percent": 50, } receipt = build_decision_receipt(request, decisions, built, diagnostics) self.assertTrue(receipt["acceptance_allowed"]) self.assertFalse(receipt["confirmation_allowed"]) self.assertEqual(receipt["overall_confidence"], "medium") self.assertIn("vimshottari_narayana_conflict", receipt["reasons"]) self.assertEqual(receipt["dasha_agreement"]["status"], "conflict") self.assertEqual(receipt["event_dasha_ledger"][0]["match_label"], "强相关") self.assertTrue(any(row["technique"] == "唯一分钟确认" and row["status"] == "blocked" for row in receipt["technique_audit_table"])) self.assertTrue(any(row["technique"] == "VedAstro 分钟级校验" and row["status"] == "blocked" for row in receipt["technique_audit_table"])) self.assertFalse(receipt["unique_minute_claim"]) def test_receipt_recasts_house_tables_per_candidate_and_blocks_unique_minute(self): request = request_events() rows = [ {"time": "05:13", "score": 20, "evidence": [], "missing_layers": []}, {"time": "05:14", "score": 8, "evidence": [], "missing_layers": []}, ] decisions = build_candidate_decisions(rows, result_id="00000000-0000-4000-8000-000000000099") chart_a = {"ascendant": {"sign": "Taurus", "lon": 40.0}, "planets": {"Sun": {"sign": "Aries", "house": 12}}} chart_b = {"ascendant": {"sign": "Gemini", "lon": 70.0}, "planets": {"Sun": {"sign": "Aries", "house": 11}}} built = { "missing_layers": [], "static_contexts": [ {**feature("05:13", d1=1, d9=1, d10=4), "chart": chart_a}, {**feature("05:14", d1=2, d9=1, d10=4), "chart": chart_b}, ], "matrix": { request["events"][0]["id"]: { "05:13": {"points": 8, "rule_ids": ["vim_md_domain_house"], "technique_layers": ["d1-rashi", "d10-dashamsa"]}, "05:14": {"points": 1, "rule_ids": ["vim_ad_domain_house"], "technique_layers": ["d1-rashi", "d10-dashamsa"]}, }, request["events"][1]["id"]: { "05:13": {"points": 1, "rule_ids": ["narayana_ad_domain_house"], "technique_layers": ["d1-rashi", "d9-navamsa"]}, "05:14": {"points": 9, "rule_ids": ["narayana_md_domain_house"], "technique_layers": ["d1-rashi", "d9-navamsa"]}, }, request["events"][2]["id"]: { "05:13": {"points": 1, "rule_ids": ["vim_md_domain_varga"], "technique_layers": ["d1-rashi", "d5-panchamsha"]}, "05:14": {"points": 1, "rule_ids": ["vim_md_domain_varga"], "technique_layers": ["d1-rashi", "d5-panchamsha"]}, }, }, } receipt = build_decision_receipt(request, decisions, built, { "leave_one_event_out_retention_rate": 1, "leave_one_domain_out_retention_rate": 1, "date_sensitivity_retention_rate": 1, "primary_secondary_margin_percent": 50, }) self.assertEqual(receipt["house_table"]["time"], "05:13") self.assertEqual(receipt["house_table"]["lagna"], "金牛座") self.assertEqual(receipt["house_tables_by_time"]["05:14"]["lagna"], "双子座") self.assertEqual(receipt["natal_recast"]["time"], "05:13") self.assertFalse(receipt["natal_recast"]["unique_minute_claim"]) self.assertFalse(receipt["confirmation_allowed"]) techniques = {row["technique"]: row["status"] for row in receipt["technique_audit_table"]} self.assertEqual(techniques["D1 本命盘"], "executed") self.assertEqual(techniques["D5 成就分盘"], "executed") self.assertEqual(techniques["KP 宫头"], "blocked") self.assertEqual(techniques["唯一分钟确认"], "blocked") if __name__ == "__main__": unittest.main()