from __future__ import annotations import random import unittest from datetime import date, datetime from scripts.rectification.candidate_contrast import ( MIN_DISCRIMINATOR_DOMAINS, MIN_DISCRIMINATOR_EVENTS, SIGNATURE_LAYERS, discriminator_gate_open, distinguish_contract_errors, feature_signature, select_signature_representatives, training_scoreable_stats, ) from scripts.rectification.event_probes import ( candidate_contrast_opportunities, 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, d9_asc: int = 1, d10_asc: int = 1, d12_asc: int = 1, d24_asc: int = 1, sun_house: int = 10, sun_varga_sign: int = 9, moon: float = 100.0, ) -> 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()}, "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": "e1", "domain": "education", "event_kind": "education_start", "date": "2014-09-01", "precision": "month"}, {"id": "e2", "domain": "education", "event_kind": "education_completion", "date": "2017-06-01", "precision": "month"}, {"id": "e3", "domain": "career", "event_kind": "career_entry", "date": "2018-07-01", "precision": "month"}, {"id": "e4", "domain": "relationship", "event_kind": "relationship_start", "date": "2021-08-01", "precision": "month"}, ] def _request(**extra: object) -> dict: return { "birth_date": "1997-08-08", "events": _gate_events(), **extra, } class DiscriminatorContractTest(unittest.TestCase): def test_ci_forbids_invalid_distinguish_payloads(self) -> None: self.assertEqual( distinguish_contract_errors({ "role": "distinguish", "information_gain": 0, "candidate_ids": ["05:00", "05:20"], "expected_outcomes": [ {"answer_class": "yes", "supports": ["05:00"], "conflicts": ["05:20"]}, {"answer_class": "no", "supports": ["05:20"], "conflicts": ["05:00"]}, ], }), ["distinguish_non_positive_information_gain"], ) self.assertEqual( distinguish_contract_errors({ "role": "distinguish", "information_gain": 0.4, "candidate_ids": [], "expected_outcomes": [ {"answer_class": "yes", "supports": [], "conflicts": []}, {"answer_class": "no", "supports": [], "conflicts": []}, ], }), ["distinguish_empty_candidate_ids"], ) self.assertEqual( distinguish_contract_errors({ "role": "distinguish", "information_gain": 0.4, "candidate_ids": ["05:00", "05:20"], "expected_outcomes": [], }), ["distinguish_empty_expected_outcomes"], ) def test_quality_never_enters_discriminating_event_probes(self) -> None: built = { "static_contexts": [ _context("05:13", d4_asc=1, d9_asc=1), _context("05:40", d4_asc=2, d9_asc=4), ] } events = _gate_events() + [{ "id": "exam", "domain": "education", "event_kind": "education_milestone", "summary": "入学考试", "date": "2015-06-01", "precision": "year", }] request = _request(events=events) probes = discriminating_event_probes( request, built, scan=window_scan(built), candidate_times=["05:13", "05:40"], representative_time="05:13", today=date(2026, 8, 22), ) self.assertFalse(any(item.get("source") == "known_event_quality" for item in probes)) self.assertFalse(any(item.get("role") == "distinguish" and distinguish_contract_errors(item) for item in probes)) clarification = event_clarification_probes(request) self.assertTrue(any(item.get("source") == "known_event_quality" for item in clarification)) self.assertTrue(all(item.get("phase") == "event_clarification" for item in clarification)) self.assertFalse(any(item.get("role") == "distinguish" for item in clarification)) def test_gate_blocks_discriminator_until_three_events_two_domains(self) -> None: built = { "static_contexts": [ _context("05:13", d4_asc=1), _context("05:40", d4_asc=2), ] } too_few = discriminating_event_probes( {"birth_date": "1997-08-08", "events": _gate_events()[:2]}, built, scan=window_scan(built), candidate_times=["05:13", "05:40"], representative_time="05:13", today=date(2026, 8, 22), ) self.assertEqual(too_few, []) collection = evidence_collection_probes({"birth_date": "1997-08-08", "events": _gate_events()[:2]}) self.assertTrue(collection) self.assertTrue(all(item.get("phase") == "evidence_collection" for item in collection)) self.assertGreaterEqual(MIN_DISCRIMINATOR_EVENTS, 3) self.assertGreaterEqual(MIN_DISCRIMINATOR_DOMAINS, 2) def test_training_gate_needs_four_events_when_one_is_holdout(self) -> None: two = _gate_events()[:2] three = _gate_events()[:3] four = _gate_events() self.assertFalse(discriminator_gate_open(two)) self.assertFalse(discriminator_gate_open(three)) two_count, _, _ = training_scoreable_stats(two) three_count, three_domains, _ = training_scoreable_stats(three) four_count, four_domains, _ = training_scoreable_stats(four) self.assertLess(two_count, MIN_DISCRIMINATOR_EVENTS) self.assertLess(three_count, MIN_DISCRIMINATOR_EVENTS) self.assertGreaterEqual(four_count, MIN_DISCRIMINATOR_EVENTS) self.assertGreaterEqual(four_domains, MIN_DISCRIMINATOR_DOMAINS) 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), ] } three_probes = discriminating_event_probes( {"birth_date": "1997-08-08", "events": three}, built, scan=window_scan(built), candidate_times=["05:13", "05:40"], representative_time="05:13", today=date(2026, 8, 22), ) four_probes = discriminating_event_probes( {"birth_date": "1997-08-08", "events": four}, built, scan=window_scan(built), candidate_times=["05:13", "05:40"], representative_time="05:13", today=date(2026, 8, 22), ) self.assertEqual(three_probes, []) self.assertTrue(four_probes) self.assertGreater(three_domains, 0) def test_signature_clusters_are_not_three_adjacent_minutes(self) -> None: rows = [ {"time": "05:13", "score": 20}, {"time": "05:14", "score": 19}, {"time": "05:15", "score": 18}, {"time": "05:40", "score": 12}, ] contexts = [ _context("05:13", d4_asc=1, d9_asc=1), _context("05:14", d4_asc=1, d9_asc=1), _context("05:15", d4_asc=1, d9_asc=1), _context("05:40", d4_asc=2, d9_asc=4), ] public = select_signature_representatives(rows, contexts) times = [row["time"] for row in public] self.assertIn("05:40", times) self.assertLessEqual(sum(1 for time in times if time in {"05:13", "05:14", "05:15"}), 1) self.assertNotEqual(feature_signature(contexts[0]), feature_signature(contexts[3])) self.assertEqual(SIGNATURE_LAYERS[:6], ("d1", "d9", "d10", "d24", "d4", "d12")) self.assertIn("md", SIGNATURE_LAYERS) def test_staging_quick_gate_runs_this_contract(self) -> None: from pathlib import Path text = Path("scripts/run_quality_gate.py").read_text(encoding="utf-8") self.assertIn('"tests/test_candidate_discriminator_contract.py"', text) def test_randomized_hidden_mutated_splits_keep_mapping_and_gain(self) -> None: rng = random.Random(20260826) built = { "static_contexts": [ _context("04:50", d4_asc=0, d9_asc=1, d10_asc=2, moon=99.0), _context("05:20", d4_asc=3, d9_asc=6, d10_asc=8, moon=101.5), ] } for _ in range(12): events = list(_gate_events()) rng.shuffle(events) for event in events: event = dict(event) event["summary"] = rng.choice(["记不清细节", "家里提过", "档案上有"]) request = _request(events=events) probes = discriminating_event_probes( request, built, scan=window_scan(built), candidate_times=["04:50", "05:20"], representative_time="04:50", today=date(2026, 8, 22), ) self.assertFalse(any(item.get("source") == "known_event_quality" for item in probes)) for probe in probes: self.assertEqual(distinguish_contract_errors(probe), []) self.assertGreater(float(probe["information_gain"]), 0) self.assertGreaterEqual(len(probe["candidate_ids"]), 2) self.assertGreaterEqual(len(probe["expected_outcomes"]), 2) self.assertTrue(probe["candidate_set_version"]) self.assertTrue(probe["candidate_split_hash"]) self.assertNotEqual(probe["candidate_split_hash"], f"{probe['domain']}:{probe['year']}") opportunities = candidate_contrast_opportunities( request, built, scan=window_scan(built), candidate_times=["04:50", "05:20"], representative_time="04:50", today=date(2026, 8, 22), ) for opportunity in opportunities: self.assertGreater(float(opportunity["information_gain"]), 0) self.assertGreaterEqual(len(opportunity["candidate_groups"]), 2) self.assertGreaterEqual(len(opportunity["expected_outcomes"]), 2) self.assertTrue(opportunity["domain"]) self.assertTrue(opportunity["source_features"]) def test_collection_reserves_holdout_out_of_scoring_and_probes(self) -> None: from scripts.rectification.case_holdout import holdout_domain_years, holdout_event_ids from scripts.rectification.scoring_service import score_from_matrix events = _gate_events() holdout = holdout_event_ids(events) self.assertEqual(len(holdout), 1) holdout_id = next(iter(holdout)) built = { "candidate_times": ["05:00", "05:20"], "matrix": { "e1": { "05:00": {"points": 10, "rule_ids": []}, "05:20": {"points": 1, "rule_ids": []}, }, "e2": { "05:00": {"points": 10, "rule_ids": []}, "05:20": {"points": 1, "rule_ids": []}, }, "e3": { "05:00": {"points": 100, "rule_ids": []}, "05:20": {"points": 0, "rule_ids": []}, }, "e4": { "05:00": {"points": 4, "rule_ids": []}, "05:20": {"points": 1, "rule_ids": []}, }, }, "missing_layers": [], } request = { "birth_date": "1997-08-08", "start_time": "04:50", "end_time": "05:30", "lat": 31.2, "lon": 121.5, "tz": 8.0, "events": [ { "id": event["id"], "domain": event["domain"], "event_kind": event["event_kind"], "date_start": event["date"], "date_end": event["date"], "precision": event["precision"], "summary": "dated", } for event in events ], } rows = score_from_matrix(request, built) by_time = {row["time"]: row for row in rows} training_ids = {event["id"] for event in events} - holdout expected = sum( built["matrix"][event_id]["05:00"]["points"] for event_id in training_ids if event_id in built["matrix"] ) self.assertEqual(by_time["05:00"]["score"], expected) self.assertFalse(any(item["event_id"] == holdout_id for item in by_time["05:00"]["evidence"])) self.assertIn(holdout_id, {event["id"] for event in events}) probes = discriminating_event_probes( _request(events=events), { "static_contexts": [ _context("05:13", d4_asc=1, d9_asc=1), _context("05:40", d4_asc=2, d9_asc=4), ] }, scan=window_scan({ "static_contexts": [ _context("05:13", d4_asc=1, d9_asc=1), _context("05:40", d4_asc=2, d9_asc=4), ] }), candidate_times=["05:13", "05:40"], representative_time="05:13", today=date(2026, 8, 22), ) blocked = holdout_domain_years(events) self.assertTrue(blocked) for probe in probes: self.assertNotIn(f"{probe['domain']}:{probe['year']}", blocked) if __name__ == "__main__": unittest.main()