Jesse_ChenandCursor
ab1ade59b7
After six dated answers, refresh can use 30-day same-year windows plus pratyantar and D9/D10 Narayana. First-round probe rules stay at 45 days. Co-authored-by: Cursor <cursoragent@cursor.com>
244 lines
8.6 KiB
Python
244 lines
8.6 KiB
Python
from __future__ import annotations
|
||
|
||
import unittest
|
||
from datetime import date, datetime
|
||
from unittest.mock import patch
|
||
|
||
from scripts.rectification.event_probes import (
|
||
MIN_BOUNDARY_DAYS,
|
||
REFRESH_MIN_BOUNDARY_DAYS,
|
||
_boundary_windows,
|
||
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,
|
||
}
|
||
|
||
# Relocation age_lo=18 → 2015; asked relocation.2015 blocks 2014–2016.
|
||
# Relationship age_lo=21 → 2018; asked relationship.2023 / event 2021 leave 2019 open.
|
||
ADULT_YEAR = 2019
|
||
|
||
TIMES = ["04:48", "04:53", "04:59", "05:06", "05:07"]
|
||
ASKED = [
|
||
"career.2020",
|
||
"career.2023",
|
||
"education.2016",
|
||
"education.2020",
|
||
"relationship.2023",
|
||
"relocation.2015",
|
||
]
|
||
|
||
|
||
def _varga(asc: int) -> dict:
|
||
return {
|
||
"Ascendant": {"sign_idx": asc},
|
||
**{name: {"sign_idx": 1} for name in PLANETS},
|
||
}
|
||
|
||
|
||
def _context(time: str, *, moon: float, d4_asc: int, d9_asc: int, d10_asc: int) -> dict:
|
||
hour, minute = (int(part) for part in time.split(":"))
|
||
planets = {**PLANETS, "Moon": moon}
|
||
return {
|
||
"candidate_at": datetime(1997, 8, 8, hour, minute),
|
||
"chart": {
|
||
"ascendant": {"lon": 10.0, "sign": "Aries"},
|
||
"planets": {
|
||
name: {"house": 4 if name == "Moon" else 10, "lon": lon}
|
||
for name, lon in planets.items()
|
||
},
|
||
},
|
||
"planet_longitudes": dict(planets),
|
||
"ascendant_index": 0,
|
||
"varga_charts": {
|
||
"D4": _varga(d4_asc),
|
||
"D9": _varga(d9_asc),
|
||
"D10": _varga(d10_asc),
|
||
"D5": _varga(1),
|
||
"D24": _varga(1),
|
||
"D12": _varga(1),
|
||
"D7": _varga(1),
|
||
"D3": _varga(1),
|
||
},
|
||
"arudha_padas": {},
|
||
"feature": {
|
||
"time": time,
|
||
"ascendant_sign_index": 0,
|
||
"varga_ascendants": {
|
||
"D4": d4_asc, "D9": d9_asc, "D10": d10_asc, "D5": 1, "D24": 1, "D12": 1,
|
||
},
|
||
},
|
||
}
|
||
|
||
|
||
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 _built(*, d4_left: int = 0, d4_right: int = 1, d9_left: int = 1, d9_right: int = 1, d10_left: int = 1, d10_right: int = 1) -> dict:
|
||
early = dict(moon=100.0, d4_asc=d4_left, d9_asc=d9_left, d10_asc=d10_left)
|
||
late = dict(moon=101.0, d4_asc=d4_right, d9_asc=d9_right, d10_asc=d10_right)
|
||
return {
|
||
"static_contexts": [
|
||
_context("04:48", **early),
|
||
_context("04:53", **early),
|
||
_context("04:59", **early),
|
||
_context("05:06", **late),
|
||
_context("05:07", **late),
|
||
]
|
||
}
|
||
|
||
|
||
def _score_months(target_year: int, months: set[int], rule: str = "vim_ad_domain_lord"):
|
||
def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict:
|
||
del birth_date, domain
|
||
early = str(context.get("feature", {}).get("time") or "")[:5] < "05:00"
|
||
if year == target_year and month in months:
|
||
return {"rule_ids": [rule] if early else ["no_domain_activation"]}
|
||
return {"rule_ids": ["no_domain_activation"]}
|
||
|
||
return fake_score
|
||
|
||
|
||
def _probes(*, refresh: bool, built: dict, vim, narayana, score):
|
||
from scripts.rectification import event_probes as probes_mod
|
||
|
||
request = {
|
||
"birth_date": "1997-08-08",
|
||
"events": _gate_events(),
|
||
"asked_probe_keys": ASKED,
|
||
"refresh_probes": refresh,
|
||
}
|
||
with (
|
||
patch.object(probes_mod, "_vim_start_dates", side_effect=vim),
|
||
patch.object(probes_mod, "_narayana_start_dates", side_effect=narayana),
|
||
patch.object(probes_mod, "_score_year", side_effect=score),
|
||
):
|
||
return discriminating_event_probes(
|
||
request,
|
||
built,
|
||
scan=window_scan(built),
|
||
candidate_times=TIMES,
|
||
representative_time="04:53",
|
||
today=date(2026, 9, 13),
|
||
)
|
||
|
||
|
||
class RefreshR3R4Test(unittest.TestCase):
|
||
def test_default_boundary_threshold_stays_45(self) -> None:
|
||
self.assertEqual(MIN_BOUNDARY_DAYS, 45)
|
||
self.assertEqual(REFRESH_MIN_BOUNDARY_DAYS, 30)
|
||
left = [date(ADULT_YEAR, 3, 1)]
|
||
right = [date(ADULT_YEAR, 4, 10)]
|
||
self.assertEqual(_boundary_windows(left, right), [])
|
||
self.assertEqual(
|
||
[item.month for item in _boundary_windows(left, right, min_days=30)],
|
||
[3, 4],
|
||
)
|
||
|
||
def test_first_round_drops_a_40_day_same_year_window(self) -> None:
|
||
def fake_vim(_birth: str, moon: float, _lo: int, _hi: int, **_kwargs: object) -> list[date]:
|
||
return [date(ADULT_YEAR, 3, 1)] if moon <= 100.0 else [date(ADULT_YEAR, 4, 10)]
|
||
|
||
def fake_narayana(_asc: int, _planets: dict, _birth: str, _lo: int, _hi: int) -> list[date]:
|
||
return []
|
||
|
||
score = _score_months(ADULT_YEAR, {3, 4})
|
||
built = _built()
|
||
first = _probes(refresh=False, built=built, vim=fake_vim, narayana=fake_narayana, score=score)
|
||
refresh = _probes(refresh=True, built=built, vim=fake_vim, narayana=fake_narayana, score=score)
|
||
self.assertFalse(any(item.get("year") == ADULT_YEAR for item in first))
|
||
self.assertTrue(any(
|
||
item.get("year") == ADULT_YEAR and item.get("source") == "dasha_boundary"
|
||
for item in refresh
|
||
), [item.get("semantic_key") for item in refresh])
|
||
|
||
def test_refresh_adds_pratyantar_windows_first_round_does_not(self) -> None:
|
||
def fake_vim(_birth: str, moon: float, _lo: int, _hi: int, *, include_pratyantar: bool = False) -> list[date]:
|
||
md = [date(2011, 1, 1)]
|
||
if not include_pratyantar:
|
||
return md
|
||
extra = date(ADULT_YEAR, 2, 1) if moon <= 100.0 else date(ADULT_YEAR, 4, 1)
|
||
return [*md, extra]
|
||
|
||
def fake_narayana(_asc: int, _planets: dict, _birth: str, _lo: int, _hi: int) -> list[date]:
|
||
return []
|
||
|
||
score = _score_months(ADULT_YEAR, {2, 4})
|
||
built = _built()
|
||
first = _probes(refresh=False, built=built, vim=fake_vim, narayana=fake_narayana, score=score)
|
||
refresh = _probes(refresh=True, built=built, vim=fake_vim, narayana=fake_narayana, score=score)
|
||
self.assertFalse(any(item.get("year") == ADULT_YEAR for item in first))
|
||
self.assertTrue(
|
||
any(item.get("year") == ADULT_YEAR for item in refresh),
|
||
[item.get("semantic_key") for item in refresh],
|
||
)
|
||
|
||
def test_refresh_adds_d9_narayana_when_natal_lagna_matches(self) -> None:
|
||
def fake_vim(_birth: str, _moon: float, _lo: int, _hi: int, **_kwargs: object) -> list[date]:
|
||
return [date(2011, 1, 1)]
|
||
|
||
def fake_narayana(asc: int, _planets: dict, _birth: str, _lo: int, _hi: int) -> list[date]:
|
||
if asc == 1:
|
||
return [date(ADULT_YEAR, 6, 1)]
|
||
if asc == 2:
|
||
return [date(ADULT_YEAR, 7, 21)]
|
||
return [date(2011, 1, 1)]
|
||
|
||
score = _score_months(ADULT_YEAR, {6, 7}, rule="narayana_ad_domain_lord")
|
||
built = _built(d4_left=0, d4_right=0, d9_left=1, d9_right=2)
|
||
first = _probes(refresh=False, built=built, vim=fake_vim, narayana=fake_narayana, score=score)
|
||
refresh = _probes(refresh=True, built=built, vim=fake_vim, narayana=fake_narayana, score=score)
|
||
self.assertFalse(any(item.get("year") == ADULT_YEAR for item in first))
|
||
self.assertTrue(
|
||
any(item.get("year") == ADULT_YEAR for item in refresh),
|
||
[item.get("semantic_key") for item in refresh],
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|