Jesse_Chen
b1173f7245
Three collected events with a reserved holdout were stalling because the discriminator door counted holdout. Public selection_allowed still had snapshot fallbacks, and health only proved the image SHA. Co-authored-by: Cursor <cursoragent@cursor.com>
396 lines
15 KiB
Python
396 lines
15 KiB
Python
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()
|