from __future__ import annotations import unittest from datetime import date, datetime from scripts.rectification.candidate_contrast import distinguish_contract_errors from scripts.rectification.event_probes import ( _agent_brief, discriminating_event_probes, event_clarification_probes, evidence_collection_probes, ) from scripts.rectification.refinement_packet import window_scan PLANETS = { "Sun": 12.0, "Moon": 100.0, "Mars": 40.0, "Mercury": 20.0, "Jupiter": 80.0, "Venus": 50.0, "Saturn": 200.0, "Rahu": 310.0, "Ketu": 130.0, } def _varga(asc: int, planet_sign: int) -> dict: return { "Ascendant": {"sign_idx": asc}, **{name: {"sign_idx": planet_sign} for name in PLANETS}, } def _context( time: str, *, d4_asc: int, sun_house: int, sun_varga_sign: int, moon: float = 100.0, missing_moon: bool = False, d9_asc: int = 1, d10_asc: int = 1, d12_asc: int = 1, d24_asc: int = 1, ) -> dict: hour, minute = (int(part) for part in time.split(":")) planets = {**PLANETS, "Moon": moon} natal_planets = { name: {"house": sun_house if name != "Moon" else 4, "lon": lon} for name, lon in planets.items() } return { "candidate_at": datetime(1997, 8, 8, hour, minute), "chart": {"ascendant": {"lon": 10.0, "sign": "Aries"}, "planets": natal_planets}, "planet_longitudes": {name: lon for name, lon in planets.items() if not (missing_moon and name == "Moon")}, "ascendant_index": 0, "varga_charts": { "D4": _varga(d4_asc, sun_varga_sign), "D9": _varga(d9_asc, 1), "D10": _varga(d10_asc, 1), "D5": _varga(1, 1), "D24": _varga(d24_asc, 1), "D12": _varga(d12_asc, 1), "D7": _varga(1, 1), "D3": _varga(1, 1), }, "arudha_padas": {}, "feature": { "time": time, "ascendant_sign_index": 0, "varga_ascendants": { "D4": d4_asc, "D9": d9_asc, "D10": d10_asc, "D5": 1, "D24": d24_asc, "D12": d12_asc, }, }, } def _gate_events() -> list[dict]: return [ { "id": "00000000-0000-4000-8000-000000000011", "domain": "education", "event_kind": "education_start", "summary": "入学", "date": "2014-09-01", "precision": "month", }, { "id": "00000000-0000-4000-8000-000000000012", "domain": "education", "event_kind": "education_completion", "summary": "毕业", "date": "2017-06-01", "precision": "month", }, { "id": "00000000-0000-4000-8000-000000000013", "domain": "career", "event_kind": "career_entry", "summary": "入职", "date": "2018-07-01", "precision": "month", }, { "id": "00000000-0000-4000-8000-000000000014", "domain": "relationship", "event_kind": "relationship_start", "summary": "相识", "date": "2021-08-01", "precision": "month", }, ] def _request(**extra: object) -> dict: return { "birth_date": "1997-08-08", "events": _gate_events(), **extra, } def _probes(request: dict, built: dict, times: list[str], representative: str, **kwargs: object): return discriminating_event_probes( request, built, scan=window_scan(built), candidate_times=times, representative_time=representative, today=date(2026, 8, 22), **kwargs, ) class EventProbesTest(unittest.TestCase): def test_agent_brief_locks_meaning_without_forcing_template_copy(self) -> None: brief = _agent_brief( year_label="2023 年前后", domain="career", family="入职、升职或职责明显加重", ) self.assertIn("2023 年前后", brief) self.assertIn("career", brief) self.assertIn("只选一个", brief) self.assertIn("口语", brief) self.assertNotIn("事件家族：入职、升职或职责明显加重", brief) def test_missing_birth_date_emits_no_probes(self) -> None: built = {"static_contexts": [_context("05:13", d4_asc=1, sun_house=4, sun_varga_sign=3)]} probes = _probes({"events": []}, built, ["05:13"], "05:13") self.assertEqual(probes, []) def test_known_exam_quality_stays_in_clarification(self) -> None: built = { "static_contexts": [ _context("05:13", d4_asc=1, sun_house=10, sun_varga_sign=9), _context("05:40", d4_asc=2, sun_house=10, sun_varga_sign=9), ] } request = _request(events=[ { "id": "00000000-0000-4000-8000-000000000013", "domain": "career", "event_kind": "career_entry", "summary": "入职", "date": "2018-07-01", "precision": "month", }, { "id": "00000000-0000-4000-8000-000000000014", "domain": "relationship", "event_kind": "relationship_start", "summary": "关系开始", "date": "2016-03-01", "precision": "month", }, { "id": "00000000-0000-4000-8000-000000000001", "domain": "education", "summary": "2015年入学考试", "date": "2015-06-01", "precision": "year", }, ]) probes = _probes(request, built, ["05:13", "05:40"], "05:13", precision_current="d5_refine") self.assertFalse(any(item["source"] == "known_event_quality" for item in probes)) self.assertFalse(any(distinguish_contract_errors(item) for item in probes)) clarification = event_clarification_probes(request) quality = next(item for item in clarification if item["source"] == "known_event_quality") self.assertEqual(quality["role"], "clarify") self.assertEqual(quality["phase"], "event_clarification") self.assertEqual(quality["year"], 2015) self.assertEqual(quality["information_gain"], 0.0) self.assertEqual(quality["expected_outcomes"], []) def test_career_events_do_not_emit_quality_probes(self) -> None: built = { "static_contexts": [ _context("05:13", d4_asc=1, sun_house=10, sun_varga_sign=9), _context("05:40", d4_asc=2, sun_house=10, sun_varga_sign=9), ] } request = _request(events=[ { "id": "00000000-0000-4000-8000-000000000001", "domain": "career", "summary": "2020 年 4 月开始实习（第一份工作）", "date": "2020-04-01", "precision": "month", }, { "id": "00000000-0000-4000-8000-000000000002", "domain": "career", "summary": "2020 年 10 月实习结束离职", "date": "2020-10-01", "precision": "month", }, ]) probes = _probes(request, built, ["05:13", "05:40"], "05:13", precision_current="d10_refine") self.assertEqual(probes, []) self.assertFalse(any(item["source"] == "known_event_quality" for item in event_clarification_probes(request))) def test_spoken_exam_anomaly_encodes_quality(self) -> None: request = _request(events=[{ "id": "00000000-0000-4000-8000-000000000001", "domain": "education", "summary": "2015年高考发挥异常", "date": "2015-06-01", "precision": "year", }]) self.assertFalse(any( item["source"] == "known_event_quality" and item["year"] == 2015 for item in event_clarification_probes(request) )) def test_gate_closed_collects_missing_domains_instead_of_discriminators(self) -> None: built = { "static_contexts": [ {"feature": {"time": "05:13", "varga_ascendants": {"D4": 1, "D9": 1, "D10": 1}}}, {"feature": {"time": "05:40", "varga_ascendants": {"D4": 2, "D9": 1, "D10": 1}}}, ] } request = {"birth_date": "1997-08-08", "events": []} probes = _probes(request, built, ["05:13", "05:40"], "05:13", precision_current="d4_refine") self.assertEqual(probes, []) collection = evidence_collection_probes(request, today=date(2026, 8, 22)) self.assertTrue(collection) self.assertTrue(all(item["phase"] == "evidence_collection" for item in collection)) self.assertTrue(all(item["role"] == "collect" for item in collection)) self.assertTrue(all(item["source"] == "age_band" for item in collection)) def test_d4_activation_difference_emits_valid_discriminator(self) -> None: built = { "static_contexts": [ _context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3), _context("05:40", d4_asc=1, sun_house=10, sun_varga_sign=9), ] } probes = _probes(_request(), built, ["05:13", "05:40"], "05:13", precision_current="d4_refine") self.assertTrue(probes) row = next(item for item in probes if item["domain"] == "relocation") self.assertIn(row["source"], {"dasha_activation", "dasha_boundary"}) self.assertEqual(row["role"], "distinguish") self.assertEqual(row["phase"], "candidate_discriminator") self.assertEqual(distinguish_contract_errors(row), []) self.assertGreater(row["information_gain"], 0) self.assertGreaterEqual(len(row["candidate_ids"]), 2) self.assertGreaterEqual(len(row["expected_outcomes"]), 2) self.assertEqual(row["tracks"], ["vimshottari", "narayana"]) self.assertFalse(row["unique_minute_claim"]) def test_same_calendar_year_shift_is_not_a_boundary_year(self) -> None: built = { "static_contexts": [ _context("05:13", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=100.0), _context("05:40", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=100.01), ] } probes = _probes(_request(), built, ["05:13", "05:40"], "05:13", precision_current="d4_refine") self.assertTrue(all(item["source"] != "dasha_boundary" for item in probes)) def test_missing_narayana_inputs_do_not_claim_dasha_year(self) -> None: built = { "static_contexts": [ _context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3, missing_moon=True), _context("05:40", d4_asc=1, sun_house=10, sun_varga_sign=9, missing_moon=True), ] } probes = _probes(_request(), built, ["05:13", "05:40"], "05:13", precision_current="d4_refine") self.assertEqual(probes, []) def test_encoded_exam_quality_does_not_enter_discriminators(self) -> None: built = { "static_contexts": [ _context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3), _context("05:40", d4_asc=1, sun_house=10, sun_varga_sign=9), ] } request = _request(events=_gate_events() + [ { "id": "00000000-0000-4000-8000-000000000001", "domain": "education", "summary": "2015 年第一次参加高考，发挥失利", "date": "2015-06-01", "precision": "year", }, { "id": "00000000-0000-4000-8000-000000000002", "domain": "education", "summary": "2016 年复读一年后再次参加高考", "date": "2016-06-01", "precision": "year", }, ]) probes = _probes(request, built, ["05:13", "05:40"], "05:13", precision_current="d5_refine") self.assertFalse(any(item["source"] == "known_event_quality" for item in probes)) self.assertTrue(any(item["source"] in {"dasha_activation", "dasha_boundary"} for item in probes)) self.assertLessEqual(len(probes), 3) for probe in probes: self.assertEqual(distinguish_contract_errors(probe), []) def test_enrollment_skips_adjacent_education_existence_year(self) -> None: built = { "static_contexts": [ _context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3, d24_asc=1), _context("05:40", d4_asc=1, sun_house=10, sun_varga_sign=9, d24_asc=2), ] } request = _request( birth_date="1998-08-08", events=_gate_events() + [{ "id": "00000000-0000-4000-8000-000000000001", "domain": "education", "summary": "2016年9月进入大学", "date": "2016-09-01", "precision": "month", }], ) probes = _probes(request, built, ["05:13", "05:40"], "05:13", precision_current="d5_refine") existence = [item for item in probes if item["domain"] == "education"] self.assertFalse(any(item["year"] in {2015, 2016, 2017} for item in existence)) def test_signature_clusters_use_full_birth_window(self) -> None: built = { "static_contexts": [ _context("04:47", d4_asc=0, sun_house=4, sun_varga_sign=3), _context("04:48", d4_asc=1, sun_house=10, sun_varga_sign=9), _context("05:00", d4_asc=0, sun_house=4, sun_varga_sign=3), _context("05:06", d4_asc=1, sun_house=10, sun_varga_sign=9), _context("05:07", d4_asc=1, sun_house=10, sun_varga_sign=9), ] } probes = _probes(_request(), built, ["05:00", "05:06", "05:07"], "05:00", precision_current="d4_refine") self.assertTrue(probes) row = next(item for item in probes if item["source"] in {"dasha_activation", "dasha_boundary"}) covered = set(row["expected_outcomes"][0]["supports"] + row["expected_outcomes"][0]["conflicts"]) self.assertGreaterEqual(len(covered), 2) self.assertTrue(covered & {"04:47", "04:48"}) self.assertNotEqual(covered, {"05:00", "05:06", "05:07"}) self.assertEqual(distinguish_contract_errors(row), []) def test_remaining_family_layer_outranks_stable_relationship(self) -> None: built = { "static_contexts": [ _context("05:13", d4_asc=1, sun_house=10, sun_varga_sign=9, d9_asc=1, d12_asc=1, moon=100.0), _context("05:40", d4_asc=1, sun_house=4, sun_varga_sign=3, d9_asc=1, d12_asc=2, moon=101.5), ] } probes = _probes(_request(), built, ["05:13", "05:40"], "05:13", precision_current="d9_refine") self.assertTrue(probes) self.assertEqual(probes[0]["domain"], "family") self.assertFalse(any(item["domain"] == "relationship" for item in probes)) self.assertFalse(any(item["domain"] == "finance" for item in probes)) def test_finance_layer_stays_volunteer_only(self) -> None: built = { "static_contexts": [ _context("05:13", d4_asc=1, sun_house=10, sun_varga_sign=9), _context("05:40", d4_asc=1, sun_house=10, sun_varga_sign=9), ] } probes = _probes(_request(), built, ["05:13", "05:40"], "05:13") self.assertFalse(any(item["domain"] == "finance" for item in probes)) def test_highest_gain_year_is_kept_not_first_hit(self) -> None: from unittest.mock import patch from scripts.rectification import event_probes as probes_mod built = { "static_contexts": [ _context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3, moon=100.0), _context("05:40", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=101.0), ] } def fake_vim(_birth_date: str, moon: float, _lo: int, _hi: int) -> list[int]: return [2010, 2020] if moon <= 100.0 else [2009, 2019] def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[int]: moon = float(planets.get("Moon") or 0) return [2010, 2020] if moon <= 100.0 else [2009, 2019] def fake_score(context: dict, *, birth_date: str, domain: str, year: int) -> dict: del birth_date, domain early = probes_mod._context_time(context) == "05:13" if year == 2010: return {"rule_ids": ["vim_ad_domain_lord"] if early else ["no_domain_activation"]} if year == 2020: return {"rule_ids": ["vim_md_domain_house"] if early else ["no_domain_activation"]} return {"rule_ids": ["no_domain_activation"]} with ( patch.object(probes_mod, "_vim_start_years", side_effect=fake_vim), patch.object(probes_mod, "_narayana_start_years", side_effect=fake_narayana), patch.object(probes_mod, "_score_year", side_effect=fake_score), ): probes = _probes(_request(), built, ["05:13", "05:40"], "05:13") row = next(item for item in probes if item["domain"] == "relocation") self.assertEqual(row["year"], 2020) self.assertEqual(row["source"], "dasha_boundary") self.assertGreater(row["information_gain"], 0) def test_relationship_boundary_starts_at_adult_age_band(self) -> None: from unittest.mock import patch from scripts.rectification import event_probes as probes_mod built = { "static_contexts": [ _context("05:13", d4_asc=1, sun_house=4, sun_varga_sign=3, d9_asc=1, moon=100.0), _context("05:40", d4_asc=1, sun_house=4, sun_varga_sign=3, d9_asc=2, moon=101.0), ] } def fake_vim(_birth_date: str, moon: float, _lo: int, _hi: int) -> list[int]: return [2003, 2019] if moon <= 100.0 else [2002, 2018] def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[int]: moon = float(planets.get("Moon") or 0) return [2003, 2019] if moon <= 100.0 else [2002, 2018] def fake_score(context: dict, *, birth_date: str, domain: str, year: int) -> dict: del birth_date, domain, year return { "rule_ids": ["vim_md_domain_house"] if probes_mod._context_time(context) == "05:13" else ["no_domain_activation"] } with ( patch.object(probes_mod, "_vim_start_years", side_effect=fake_vim), patch.object(probes_mod, "_narayana_start_years", side_effect=fake_narayana), patch.object(probes_mod, "_score_year", side_effect=fake_score), ): probes = _probes(_request(), built, ["05:13", "05:40"], "05:13") relationship = [item for item in probes if item["domain"] == "relationship"] self.assertTrue(relationship) self.assertTrue(all(item["year"] >= 2018 for item in relationship)) self.assertFalse(any(item["year"] in {2002, 2003} for item in relationship)) if __name__ == "__main__": unittest.main()