Jesse_Chen
a88467ffa8
Keep unique-top and width on confirmation only, and stop lagna-frame follow-ups from blocking cards on an already-scored cluster. Co-authored-by: Cursor <cursoragent@cursor.com>
366 lines
17 KiB
Python
366 lines
17 KiB
Python
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()
|