Files
Jyotisha/tests/test_rectification_refinement_packet.py
T
jesse-ux 227a75719b
Independent Staging Quality Gate / validate (push) Failing after 13m34s
Independent Staging Quality Gate / publish (push) Skipped
fix(rectification): D9 判别题不再借 D24 探针,性格描述一列一句
2026-09-17 01:06:17 +08:00

483 lines
23 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.
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"},
{"id": "00000000-0000-4000-8000-000000000004", "domain": "family", "summary": "家人变化", "event_kind": "family_event", "precision": "day", "date_start": "2020-01-01", "date_end": "2020-01-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 发生变化",
"from_sign": "金牛座",
"to_sign": "天蝎座",
}])
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_dasha_agreement_same_cluster_peaks_agree(self):
built = {
"matrix": {
"00000000-0000-4000-8000-000000000001": {
"04:54": {"points": 8, "rule_ids": ["vim_md_domain_house"]},
"04:59": {"points": 1, "rule_ids": ["vim_ad_domain_house"]},
},
"00000000-0000-4000-8000-000000000002": {
"04:54": {"points": 1, "rule_ids": ["narayana_ad_domain_house"]},
"04:59": {"points": 9, "rule_ids": ["narayana_md_domain_house"]},
},
}
}
clusters = [{
"time": "04:54",
"cluster_start": "04:54",
"cluster_end": "04:59",
"cluster_times": ["04:54", "04:59"],
}]
without_clusters = dasha_agreement(built, ["04:54", "04:59"])
self.assertEqual(without_clusters["status"], "conflict")
agreement = dasha_agreement(built, ["04:54", "04:59"], clusters=clusters)
self.assertEqual(agreement["status"], "agree")
self.assertEqual(agreement["vimshottari_top"], "04:54")
self.assertEqual(agreement["narayana_top"], "04:59")
def test_dasha_agreement_cross_cluster_peaks_conflict(self):
built = {
"matrix": {
"00000000-0000-4000-8000-000000000001": {
"05:13": {"points": 8, "rule_ids": ["vim_md_domain_house"]},
"05:20": {"points": 1, "rule_ids": ["vim_ad_domain_house"]},
},
"00000000-0000-4000-8000-000000000002": {
"05:13": {"points": 1, "rule_ids": ["narayana_ad_domain_house"]},
"05:20": {"points": 9, "rule_ids": ["narayana_md_domain_house"]},
},
}
}
clusters = [
{"time": "05:13", "cluster_start": "05:13", "cluster_end": "05:13", "cluster_times": ["05:13"]},
{"time": "05:20", "cluster_start": "05:20", "cluster_end": "05:20", "cluster_times": ["05:20"]},
]
agreement = dasha_agreement(built, ["05:13", "05:20"], clusters=clusters)
self.assertEqual(agreement["status"], "conflict")
def test_dasha_agreement_one_track_is_partial(self):
built = {
"matrix": {
"00000000-0000-4000-8000-000000000001": {
"05:13": {"points": 8, "rule_ids": ["vim_md_domain_house"]},
"05:20": {"points": 1, "rule_ids": ["vim_ad_domain_house"]},
},
}
}
agreement = dasha_agreement(built, ["05:13", "05:20"])
self.assertEqual(agreement["status"], "partial")
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(len(packet["nakshatra_boundary"]["options"][0]["traits"]), 1)
self.assertEqual(len(packet["nakshatra_boundary"]["options"][1]["traits"]), 1)
self.assertNotEqual(
packet["nakshatra_boundary"]["options"][0]["traits"][0],
packet["nakshatra_boundary"]["options"][1]["traits"][0],
)
self.assertIn("A", packet["nakshatra_boundary"]["user_meaning"])
self.assertNotIn("", packet["nakshatra_boundary"]["user_meaning"])
self.assertEqual(packet["oos_blind_prompts"][0]["domain"], "finance")
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"])
self.assertEqual(packet["discriminating_event_probes"], [])
def test_decision_receipt_downgrades_confidence_on_dasha_conflict(self):
request = request_events()
rows = [
{"time": "05:13", "score": 20, "evidence": [], "missing_layers": []},
{"time": "05:20", "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:20", d1=2, d9=1, d10=4)],
"matrix": {
request["events"][0]["id"]: {
"05:13": {"points": 8, "rule_ids": ["vim_md_domain_house"]},
"05:20": {"points": 1, "rule_ids": ["vim_ad_domain_house"]},
},
request["events"][1]["id"]: {
"05:13": {"points": 1, "rule_ids": ["narayana_ad_domain_house"]},
"05:20": {"points": 9, "rule_ids": ["narayana_md_domain_house"]},
},
request["events"][2]["id"]: {
"05:13": {"points": 1, "rule_ids": ["vim_md_domain_varga"]},
"05:20": {"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_same_cluster_dasha_peaks_agree_without_opening_confirmation(self):
request = request_events()
rows = [
{"time": "04:54", "score": 20, "evidence": [], "missing_layers": []},
{"time": "04:59", "score": 18, "evidence": [], "missing_layers": []},
]
contexts = [
feature("04:54", d1=1, d9=1, d10=4),
feature("04:59", d1=1, d9=1, d10=4),
]
decisions = build_candidate_decisions(
rows,
result_id="00000000-0000-4000-8000-000000000099",
static_contexts=contexts,
)
built = {
"missing_layers": [],
"static_contexts": contexts,
"candidate_times": ["04:54", "04:59"],
"matrix": {
request["events"][0]["id"]: {
"04:54": {"points": 8, "rule_ids": ["vim_md_domain_house"]},
"04:59": {"points": 1, "rule_ids": ["vim_ad_domain_house"]},
},
request["events"][1]["id"]: {
"04:54": {"points": 1, "rule_ids": ["narayana_ad_domain_house"]},
"04:59": {"points": 9, "rule_ids": ["narayana_md_domain_house"]},
},
request["events"][2]["id"]: {
"04:54": {"points": 1, "rule_ids": ["vim_md_domain_varga"]},
"04:59": {"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.assertEqual(receipt["dasha_agreement"]["status"], "agree")
self.assertNotIn("vimshottari_narayana_conflict", receipt["reasons"])
self.assertFalse(receipt["confirmation_allowed"])
width = receipt["gates"]["exact_confirmation"]["indistinguishable_width_minutes"]
self.assertGreaterEqual(width, 6)
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()