Jesse_Chen
04b2ccdbe2
Distinguish-stage reverse-verify questions were exhausting after 1–3 dated probes. Union boundary windows across representative pairs, keep multiple years per domain, raise the public cap, and allow activation fallback without relaxing MIN_BOUNDARY_DAYS or scoring. Co-authored-by: Cursor <cursoragent@cursor.com>
902 lines
40 KiB
Python
902 lines
40 KiB
Python
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()
|