Files
Jyotisha/tests/test_rectification_event_probes.py
T
Jesse_Chen f5e73ef326
Independent Staging Quality Gate / validate (push) Failing after 22m30s
Independent Staging Quality Gate / publish (push) Has been skipped
fix(web): rank remaining-minute probes by split and match choice kind
Choice cards used a hardcoded domain menu and always asked existence.
Rank scoring layers by remaining-minute entropy, keep finance and health
volunteer-only, and ask D9/D10 style or exam quality so taps match outcomes.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-25 20:44:55 +08:00

447 lines
17 KiB
Python

from __future__ import annotations
import unittest
from datetime import date, datetime
from scripts.rectification.event_probes import discriminating_event_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,
) -> 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(1, 1),
"D10": _varga(1, 1),
"D5": _varga(1, 1),
"D24": _varga(1, 1),
"D12": _varga(1, 1),
"D7": _varga(1, 1),
"D3": _varga(1, 1),
},
"arudha_padas": {},
"feature": {
"time": time,
"ascendant_sign_index": 0,
"varga_ascendants": {"D4": d4_asc, "D9": 1, "D10": 1, "D5": 1},
},
}
def _request(**extra: object) -> dict:
return {
"birth_date": "1997-08-08",
"events": [],
**extra,
}
class EventProbesTest(unittest.TestCase):
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 = discriminating_event_probes(
{"events": []},
built,
scan=window_scan(built),
candidate_times=["05:13"],
representative_time="05:13",
today=date(2026, 8, 22),
)
self.assertEqual(probes, [])
def test_known_gaokao_event_asks_quality_not_existence(self) -> None:
built = {
"static_contexts": [
_context("05:13", d4_asc=1, sun_house=10, sun_varga_sign=9),
_context("05:14", d4_asc=2, sun_house=10, sun_varga_sign=9),
]
}
probes = discriminating_event_probes(
_request(events=[{
"id": "00000000-0000-4000-8000-000000000001",
"domain": "education",
"summary": "2015年高考",
"date": "2015-06-01",
"precision": "year",
}]),
built,
scan=window_scan(built),
candidate_times=["05:13", "05:14"],
representative_time="05:13",
precision_current="d5_refine",
today=date(2026, 8, 22),
)
self.assertTrue(probes)
quality = next(item for item in probes if item["source"] == "known_event_quality")
self.assertEqual(quality["role"], "distinguish")
self.assertEqual(quality["year"], 2015)
self.assertIn("年份锁定", quality["user_meaning"])
self.assertIn("请写成", quality["user_meaning"])
self.assertIn("发挥失常", quality["user_meaning"])
self.assertNotIn("更像哪一件", quality["user_meaning"])
self.assertNotIn("05:14", quality["user_meaning"])
self.assertNotIn("points", str(probes))
def test_age_band_fallback_without_full_charts(self) -> None:
built = {
"static_contexts": [
{"feature": {"time": "05:13", "varga_ascendants": {"D4": 1, "D9": 1, "D10": 1}}},
{"feature": {"time": "05:14", "varga_ascendants": {"D4": 2, "D9": 1, "D10": 1}}},
]
}
probes = discriminating_event_probes(
_request(),
built,
scan=window_scan(built),
candidate_times=["05:13", "05:14"],
representative_time="05:13",
precision_current="d4_refine",
today=date(2026, 8, 22),
)
self.assertTrue(probes)
self.assertEqual(probes[0]["source"], "age_band")
self.assertEqual(probes[0]["role"], "reverse_verify")
self.assertEqual(probes[0]["domain"], "relocation")
self.assertEqual(probes[0]["year"], 2018)
self.assertIn("年份锁定", probes[0]["user_meaning"])
self.assertIn("请写成", probes[0]["user_meaning"])
self.assertIn("搬家", probes[0]["user_meaning"])
self.assertFalse(probes[0]["unique_minute_claim"])
self.assertNotIn("05:14", probes[0]["user_meaning"])
def test_d4_activation_difference_asks_move_in_that_year(self) -> None:
built = {
"static_contexts": [
_context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3),
_context("05:14", d4_asc=1, sun_house=10, sun_varga_sign=9),
]
}
probes = discriminating_event_probes(
_request(),
built,
scan=window_scan(built),
candidate_times=["05:13", "05:14"],
representative_time="05:13",
precision_current="d4_refine",
today=date(2026, 8, 22),
)
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"], "reverse_verify")
self.assertIn("年份锁定", row["user_meaning"])
self.assertIn("请写成", row["user_meaning"])
self.assertIn("搬家", row["user_meaning"])
self.assertIn(str(row["year"]), row["year_label"])
self.assertNotIn("更像哪一件", row["user_meaning"])
self.assertNotIn("points", str(row))
self.assertNotIn("05:13", row["user_meaning"])
self.assertGreater(row["information_gain"], 0)
self.assertTrue(row["expected_outcomes"])
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:14", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=100.01),
]
}
probes = discriminating_event_probes(
_request(),
built,
scan=window_scan(built),
candidate_times=["05:13", "05:14"],
representative_time="05:13",
precision_current="d4_refine",
today=date(2026, 8, 22),
)
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:14", d4_asc=1, sun_house=10, sun_varga_sign=9, missing_moon=True),
]
}
probes = discriminating_event_probes(
_request(),
built,
scan=window_scan(built),
candidate_times=["05:13", "05:14"],
representative_time="05:13",
precision_current="d4_refine",
today=date(2026, 8, 22),
)
self.assertTrue(probes)
self.assertTrue(all(item["source"] == "age_band" for item in probes))
def test_encoded_exam_quality_does_not_fill_probe_slots(self) -> None:
built = {
"static_contexts": [
_context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3),
_context("05:14", d4_asc=1, sun_house=10, sun_varga_sign=9),
]
}
probes = discriminating_event_probes(
_request(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",
},
{
"id": "00000000-0000-4000-8000-000000000003",
"domain": "education",
"summary": "2016 年 9 月进入大学",
"date": "2016-09-01",
"precision": "month",
},
]),
built,
scan=window_scan(built),
candidate_times=["05:13", "05:14"],
representative_time="05:13",
precision_current="d5_refine",
today=date(2026, 8, 22),
)
self.assertTrue(probes)
self.assertFalse(any(
item["source"] == "known_event_quality" and item["year"] in {2015, 2016}
for item in probes
))
self.assertTrue(any(item["source"] in {"dasha_activation", "dasha_boundary"} for item in probes))
self.assertTrue(any(item["domain"] == "relocation" for item in probes))
self.assertLessEqual(len(probes), 3)
def test_enrollment_quality_does_not_block_dasha(self) -> None:
built = {
"static_contexts": [
_context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3),
_context("05:14", d4_asc=1, sun_house=10, sun_varga_sign=9),
]
}
probes = discriminating_event_probes(
_request(events=[{
"id": "00000000-0000-4000-8000-000000000001",
"domain": "education",
"summary": "2016年9月进入大学",
"date": "2016-09-01",
"precision": "month",
}]),
built,
scan=window_scan(built),
candidate_times=["05:13", "05:14"],
representative_time="05:13",
precision_current="d5_refine",
today=date(2026, 8, 22),
)
self.assertTrue(probes)
self.assertTrue(any(item["source"] in {"dasha_activation", "dasha_boundary"} for item in probes))
self.assertTrue(any(item["domain"] == "relocation" for item in probes))
def test_enrollment_skips_adjacent_education_existence_year(self) -> None:
built = {
"static_contexts": [
{"feature": {"time": "05:13", "varga_ascendants": {"D4": 1, "D9": 1, "D10": 1, "D5": 1}}},
{"feature": {"time": "05:14", "varga_ascendants": {"D4": 2, "D9": 1, "D10": 1, "D5": 2}}},
]
}
probes = discriminating_event_probes(
_request(
birth_date="1998-08-08",
events=[{
"id": "00000000-0000-4000-8000-000000000001",
"domain": "education",
"summary": "2016年9月进入大学",
"date": "2016-09-01",
"precision": "month",
}],
),
built,
scan=window_scan(built),
candidate_times=["05:13", "05:14"],
representative_time="05:13",
precision_current="d5_refine",
today=date(2026, 8, 22),
)
existence = [
item for item in probes
if item["domain"] == "education" and item["source"] != "known_event_quality"
]
self.assertFalse(any(item["year"] in {2015, 2016, 2017} for item in existence))
self.assertFalse(any("高考是 2015" in str(item.get("user_meaning") or "") for item in probes))
def test_representatives_prefer_remaining_candidate_times(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 = discriminating_event_probes(
_request(),
built,
scan=window_scan(built),
candidate_times=["05:00", "05:06", "05:07"],
representative_time="05:00",
precision_current="d4_refine",
today=date(2026, 8, 22),
)
self.assertTrue(probes)
row = next(item for item in probes if item["source"] in {"dasha_activation", "dasha_boundary"})
times = {row.get("left_time"), row.get("right_time")}
self.assertTrue(times <= {"05:00", "05:06", "05:07"})
self.assertIn("05:00", times)
self.assertTrue(times & {"05:06", "05:07"})
covered = set(row["expected_outcomes"][0]["supports"] + row["expected_outcomes"][0]["conflicts"])
self.assertEqual(covered, {"05:00", "05:06", "05:07"})
def test_remaining_family_layer_outranks_stable_relationship(self) -> None:
built = {
"static_contexts": [
{
"feature": {
"time": "05:13",
"varga_ascendants": {"D9": 1, "D10": 1, "D4": 1, "D12": 1, "D7": 1},
}
},
{
"feature": {
"time": "05:14",
"varga_ascendants": {"D9": 1, "D10": 1, "D4": 1, "D12": 2, "D7": 2},
}
},
]
}
probes = discriminating_event_probes(
_request(),
built,
scan=window_scan(built),
candidate_times=["05:13", "05:14"],
representative_time="05:13",
precision_current="d9_refine",
today=date(2026, 8, 22),
)
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": [
{"feature": {"time": "05:13", "varga_ascendants": {"D2": 1, "D9": 1, "D10": 1}}},
{"feature": {"time": "05:14", "varga_ascendants": {"D2": 2, "D9": 1, "D10": 1}}},
]
}
probes = discriminating_event_probes(
_request(),
built,
scan=window_scan(built),
candidate_times=["05:13", "05:14"],
representative_time="05:13",
today=date(2026, 8, 22),
)
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:14", 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 = discriminating_event_probes(
_request(),
built,
scan=window_scan(built),
candidate_times=["05:13", "05:14"],
representative_time="05:13",
today=date(2026, 8, 22),
)
row = next(item for item in probes if item["domain"] == "relocation")
self.assertEqual(row["year"], 2020)
self.assertEqual(row["source"], "dasha_boundary")
if __name__ == "__main__":
unittest.main()