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 ( MAX_BOUNDARY_CANDIDATES_PER_DOMAIN, MAX_COLLECTION_PROBES, MAX_PROBES, MAX_PROBES_PER_DOMAIN, _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, ) def _multi_layer_window_built() -> dict: """Two signature clusters, moons far enough apart that Vimshottari windows exist. Before supply expansion this fixture published exactly three probes (career.2024.02 / relationship.2024.02 / relocation.2015.03). """ early = dict(d4_asc=0, sun_house=4, sun_varga_sign=3, moon=100.0, d9_asc=1, d10_asc=1, d12_asc=1, d24_asc=1) late = dict(d4_asc=1, sun_house=10, sun_varga_sign=9, moon=101.0, d9_asc=2, d10_asc=2, d12_asc=2, d24_asc=2) return { "static_contexts": [ _context("04:47", **early), _context("04:48", **late), _context("05:00", **early), _context("05:06", **late), _context("05:07", **late), ] } 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.assertEqual(len(collection), MAX_COLLECTION_PROBES) 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"]) self.assertEqual(len(row["style_options"]), 4) self.assertEqual( {item["answer_class"] for item in row["style_options"]}, {"yes", "weak_yes", "no", "unsure"}, ) self.assertIn("这段记不清楚", {item["label"] for item in row["style_options"]}) 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_same_year_month_apart_boundary_uses_engine_month(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[date]: return [date(2018, 3, 15)] if moon <= 100.0 else [date(2018, 9, 20)] def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]: moon = float(planets.get("Moon") or 0) return [date(2018, 3, 15)] if moon <= 100.0 else [date(2018, 9, 20)] def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict: del birth_date, domain early = probes_mod._context_time(context) == "05:13" if year == 2018 and month == 3: return {"rule_ids": ["vim_ad_domain_lord"] if early else ["no_domain_activation"]} if year == 2018 and month == 9: 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_dates", side_effect=fake_vim), patch.object(probes_mod, "_narayana_start_dates", 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["source"], "dasha_boundary") self.assertEqual(row["year"], 2018) self.assertIn(row["month"], {3, 9}) self.assertRegex(row["year_label"], r"2018 年 [39] 月前后") self.assertIn("2018 年", row["user_meaning"]) self.assertIn("月前后", row["user_meaning"]) self.assertNotIn("points", str(row)) def test_vim_start_dates_keep_engine_month(self) -> None: from scripts.rectification import event_probes as probes_mod starts = probes_mod._vim_start_dates("1997-08-08", 100.0, 2005, 2025) self.assertTrue(starts) self.assertTrue(all(isinstance(item, date) and not isinstance(item, datetime) for item in starts)) self.assertTrue(any(item.month != 7 or item.day != 1 for item in starts)) def test_age_band_fallback_stays_year_precision(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=100.0), ] } def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict: del birth_date, domain, year, month 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_dates", return_value=[]), patch.object(probes_mod, "_narayana_start_dates", return_value=[]), 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["source"] == "dasha_activation") self.assertNotIn("month", row) self.assertRegex(row["year_label"], r"^\d{4} 年前后$") 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), MAX_PROBES) 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[date]: return [date(2010, 3, 15), date(2020, 3, 15)] if moon <= 100.0 else [date(2009, 9, 15), date(2019, 9, 15)] def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]: moon = float(planets.get("Moon") or 0) return [date(2010, 3, 15), date(2020, 3, 15)] if moon <= 100.0 else [date(2009, 9, 15), date(2019, 9, 15)] def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict: del birth_date, domain, month 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_dates", side_effect=fake_vim), patch.object(probes_mod, "_narayana_start_dates", 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[date]: return [date(2003, 3, 15), date(2019, 3, 15)] if moon <= 100.0 else [date(2002, 9, 15), date(2018, 9, 15)] def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]: moon = float(planets.get("Moon") or 0) return [date(2003, 3, 15), date(2019, 3, 15)] if moon <= 100.0 else [date(2002, 9, 15), date(2018, 9, 15)] def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict: del birth_date, domain, year, month 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_dates", side_effect=fake_vim), patch.object(probes_mod, "_narayana_start_dates", 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)) def test_career_and_relocation_boundaries_skip_childhood_years(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, d10_asc=1, moon=100.0), _context("05:40", d4_asc=1, sun_house=10, sun_varga_sign=9, d10_asc=2, moon=101.0), ] } def fake_vim(_birth_date: str, moon: float, _lo: int, _hi: int) -> list[date]: early = [date(2005, 5, 15), date(2012, 11, 15), date(2019, 4, 15), date(2023, 5, 15)] late = [date(2005, 11, 15), date(2012, 5, 15), date(2019, 10, 15), date(2023, 11, 15)] return early if moon <= 100.0 else late def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]: moon = float(planets.get("Moon") or 0) return fake_vim(_birth_date, moon, _lo, _hi) def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict: del birth_date, domain, year, month 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_dates", side_effect=fake_vim), patch.object(probes_mod, "_narayana_start_dates", side_effect=fake_narayana), patch.object(probes_mod, "_score_year", side_effect=fake_score), ): probes = _probes(_request(), built, ["05:13", "05:40"], "05:13") career = [item for item in probes if item["domain"] == "career"] relocation = [item for item in probes if item["domain"] == "relocation"] self.assertTrue(career) self.assertTrue(all(item["year"] >= 2019 for item in career)) self.assertFalse(any(item["year"] in {2005, 2012} for item in career)) self.assertTrue(relocation) self.assertTrue(all(item["year"] >= 2015 for item in relocation)) self.assertFalse(any(item["year"] == 2005 for item in relocation)) def test_recorded_relationship_skips_adjacent_existence_year(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), ] } request = _request(events=_gate_events() + [{ "id": "00000000-0000-4000-8000-000000000021", "domain": "relationship", "event_kind": "relationship_start", "summary": "相识", "date": "2024-05-01", "precision": "month", }]) def fake_vim(_birth_date: str, moon: float, _lo: int, _hi: int) -> list[date]: return [date(2023, 5, 15), date(2019, 3, 15)] if moon <= 100.0 else [date(2023, 11, 15), date(2018, 9, 15)] def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]: moon = float(planets.get("Moon") or 0) return fake_vim(_birth_date, moon, _lo, _hi) def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict: del birth_date, domain, month return { "rule_ids": ["vim_md_domain_house"] if probes_mod._context_time(context) == "05:13" and year != 2024 else ["no_domain_activation"] } with ( patch.object(probes_mod, "_vim_start_dates", side_effect=fake_vim), patch.object(probes_mod, "_narayana_start_dates", 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.assertFalse(any(item["year"] == 2023 for item in relationship)) def test_discriminating_probes_expand_beyond_previous_cap(self) -> None: probes = _probes(_request(), _multi_layer_window_built(), ["05:00", "05:06", "05:07"], "05:00") keys = [item["semantic_key"] for item in probes] self.assertGreater(len(probes), 3) self.assertLessEqual(len(probes), MAX_PROBES) self.assertTrue(keys) for probe in probes: self.assertIsInstance(probe.get("year"), int) self.assertGreaterEqual(int(probe["year"]), 1900) self.assertNotEqual(int(probe["year"]), 0) self.assertEqual(distinguish_contract_errors(probe), []) self.assertIn(probe["source"], {"dasha_boundary", "dasha_activation"}) self.assertGreaterEqual(len(probe["style_options"]), 4) self.assertGreaterEqual(len(probe["candidate_ids"]), 2) self.assertGreaterEqual(len(probe["expected_outcomes"]), 2) def test_per_domain_boundary_evaluation_respects_k_budget(self) -> None: from unittest.mock import patch from scripts.rectification import event_probes as probes_mod real = probes_mod._evaluate_contexts calls_by_domain: dict[str, int] = {} def wrapped( contexts, *, birth_date: str, domain: str, year: int, source: str, month: int | None = None, clusters=None, set_version=None, ): if source == "dasha_boundary": calls_by_domain[domain] = calls_by_domain.get(domain, 0) + 1 return real( contexts, birth_date=birth_date, domain=domain, year=year, source=source, month=month, clusters=clusters, set_version=set_version, ) with patch.object(probes_mod, "_evaluate_contexts", side_effect=wrapped): probes = _probes(_request(), _multi_layer_window_built(), ["05:00", "05:06", "05:07"], "05:00") self.assertTrue(probes) self.assertTrue(calls_by_domain) filled: dict[str, set[int]] = {} for probe in probes: if probe["source"] != "dasha_boundary": continue filled.setdefault(str(probe["domain"]), set()).add(int(probe["year"])) for domain, count in calls_by_domain.items(): if len(filled.get(domain, set())) >= 2: self.assertLessEqual( count, MAX_BOUNDARY_CANDIDATES_PER_DOMAIN, {domain: count, "filled": filled.get(domain)}, ) def test_adjacent_pair_boundary_is_used_when_first_last_has_none(self) -> None: from unittest.mock import patch from scripts.rectification import event_probes as probes_mod built = { "static_contexts": [ _context("04:50", d4_asc=0, sun_house=4, sun_varga_sign=3, moon=100.0, d9_asc=1), _context("05:10", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=101.0, d9_asc=1), _context("05:40", d4_asc=0, sun_house=4, sun_varga_sign=3, moon=100.0, d9_asc=2), ] } def fake_vim(_birth_date: str, moon: float, _lo: int, _hi: int) -> list[date]: return [date(2018, 3, 15)] if moon <= 100.0 else [date(2018, 9, 20)] def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]: moon = float(planets.get("Moon") or 0) return fake_vim(_birth_date, moon, _lo, _hi) def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict: del birth_date, domain if year == 2018 and month in {3, 9}: return { "rule_ids": ["vim_md_domain_house"] if probes_mod._context_time(context) == "04:50" else ["no_domain_activation"] } return {"rule_ids": ["no_domain_activation"]} with ( patch.object(probes_mod, "_vim_start_dates", side_effect=fake_vim), patch.object(probes_mod, "_narayana_start_dates", side_effect=fake_narayana), patch.object(probes_mod, "_score_year", side_effect=fake_score), ): probes = _probes(_request(), built, ["04:50", "05:10", "05:40"], "04:50") boundary = [item for item in probes if item["source"] == "dasha_boundary"] self.assertTrue(boundary) self.assertTrue(all(item["year"] == 2018 for item in boundary)) self.assertTrue(all(item.get("month") in {3, 9} for item in boundary)) self.assertTrue(all(item.get("year") for item in probes)) def test_same_domain_keeps_multiple_boundary_years(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[date]: if moon <= 100.0: return [date(2016, 3, 15), date(2022, 3, 15)] return [date(2016, 9, 20), date(2022, 9, 20)] def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]: moon = float(planets.get("Moon") or 0) return fake_vim(_birth_date, moon, _lo, _hi) def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict: del birth_date, domain, month early = probes_mod._context_time(context) == "05:13" if year in {2016, 2022}: 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_dates", side_effect=fake_vim), patch.object(probes_mod, "_narayana_start_dates", side_effect=fake_narayana), patch.object(probes_mod, "_score_year", side_effect=fake_score), ): probes = _probes(_request(), built, ["05:13", "05:40"], "05:13") relocation = [item for item in probes if item["domain"] == "relocation" and item["source"] == "dasha_boundary"] years = {int(item["year"]) for item in relocation} keys = { (item["domain"], int(item["year"]), int(item.get("month") or 0), item["source"]) for item in relocation } self.assertGreaterEqual(len(years), 2) self.assertEqual(years, {2016, 2022}) self.assertEqual(len(keys), len(relocation)) def test_activation_supplements_boundary_without_exceeding_domain_cap(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[date]: return [date(2020, 3, 15)] if moon <= 100.0 else [date(2020, 9, 20)] def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]: moon = float(planets.get("Moon") or 0) return fake_vim(_birth_date, moon, _lo, _hi) def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict: del birth_date, domain early = probes_mod._context_time(context) == "05:13" if (year == 2020 and month in {3, 9}) or (year == 2018 and month is None): 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_dates", side_effect=fake_vim), patch.object(probes_mod, "_narayana_start_dates", side_effect=fake_narayana), patch.object(probes_mod, "_score_year", side_effect=fake_score), ): probes = _probes(_request(), built, ["05:13", "05:40"], "05:13") relocation = [item for item in probes if item["domain"] == "relocation"] sources = {item["source"] for item in relocation} self.assertIn("dasha_boundary", sources) self.assertIn("dasha_activation", sources) self.assertLessEqual(len(relocation), MAX_PROBES_PER_DOMAIN) self.assertLessEqual(len(probes), MAX_PROBES) self.assertTrue(all(isinstance(item.get("year"), int) and item["year"] for item in relocation)) activation = next(item for item in relocation if item["source"] == "dasha_activation") self.assertEqual(activation["year"], 2018) self.assertNotIn("month", activation) def test_domain_at_n_does_not_add_activation_past_cap(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[date]: if moon <= 100.0: return [date(2016, 3, 15), date(2019, 3, 15), date(2022, 3, 15)] return [date(2016, 9, 20), date(2019, 9, 20), date(2022, 9, 20)] def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]: moon = float(planets.get("Moon") or 0) return fake_vim(_birth_date, moon, _lo, _hi) def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict: del birth_date, domain, month early = probes_mod._context_time(context) == "05:13" if year in {2016, 2019, 2022, 2018}: 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_dates", side_effect=fake_vim), patch.object(probes_mod, "_narayana_start_dates", side_effect=fake_narayana), patch.object(probes_mod, "_score_year", side_effect=fake_score), ): probes = _probes(_request(), built, ["05:13", "05:40"], "05:13") relocation = [item for item in probes if item["domain"] == "relocation"] self.assertLessEqual(len(relocation), MAX_PROBES_PER_DOMAIN) self.assertFalse(any(item["source"] == "dasha_activation" for item in relocation)) self.assertEqual({int(item["year"]) for item in relocation}, {2016, 2019, 2022}) if __name__ == "__main__": unittest.main()