From 04b2ccdbe2f175936efdc65aec10e1742d8a72b0 Mon Sep 17 00:00:00 2001 From: Jesse_Chen Date: Sun, 30 Aug 2026 11:46:30 +0800 Subject: [PATCH] feat(rectification): expand dated dasha_boundary probe supply MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- CHANGELOG.md | 14 + scripts/rectification/event_probes.py | 359 ++++++++++++++++------- tests/test_rectification_event_probes.py | 259 +++++++++++++++- 3 files changed, 530 insertions(+), 102 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c9012fc1..85196999 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,19 @@ # 印度占星 Skill 更新日志 +## 2026-08-30 — 反推前事探针供给扩容（Skill 10.0.13 不变） + +区分阶段在收窄无年份分盘对比题之后，带年份的 `dasha_boundary` 反推题每个 case 只剩 1–3 条。本轮只加供给，不放宽判定门槛、不改计分口径。 + +- 边界窗改为多对代表分钟求并集（时间排序后的相邻对 + 首尾对），仍执行 `MIN_BOUNDARY_DAYS = 45`。 +- 同领域按 `information_gain` 保留前 N=3 条年份（去重键仍是 `(domain, year, month, source)`）。 +- 公开上限 `MAX_PROBES` 提到 8：约 4–6 个领域竞争时，可让约 3 个领域各留两年并再加少量 activation，前端出题层仍按信息量全局排序。 +- 采集上限仍是 `MAX_COLLECTION_PROBES = 3`，避免采集题被连带放宽。 +- 即使该领域已有边界探针，只要还没到每域 N，也可以再补一条带年份的 `dasha_activation`（仍走 blocked_years / holdout / 契约校验）。 +- 每域评估预算 K=8：先评均匀抽样的 K 个年月，按信息量保留；若 K 次全部未命中再扫描剩余窗直到第一击，避免漏掉唯一能区分的月份。K 与 N、MAX_PROBES 的关系：N ≤ K，公开列表再按全局信息量截到 MAX_PROBES。 +- 未改 `_evaluate_contexts` / `LEVEL_P` / `match_level` / information_gain 算法，未放宽 confirmation gate，未新增「已确认唯一分钟」路径。 + +同一 fixture（两簇、Moon 100 vs 101、多层分盘差）实测：公开探针 3 → 7 条；三次打分中位 0.2878s → 0.0108s（K 截断减少 `_evaluate_contexts` 次数）。 + ## v6.9.14（2026-06-21）—— 发布卫生、CI门禁与包产物校验 > **验证**：65 techniques registry validate PASS（55 covered / 10 complete / 0 partial / 0 missing）；475 pytest PASS；legacy runner 102/102 PASS；frontend Vite build PASS；wheel/sdist build PASS；twine check PASS。 diff --git a/scripts/rectification/event_probes.py b/scripts/rectification/event_probes.py index 5e246566..132126fa 100644 --- a/scripts/rectification/event_probes.py +++ b/scripts/rectification/event_probes.py @@ -42,7 +42,22 @@ from scripts.rectification.probe_question_contract import ( ) from scripts.rectification.refinement_packet import match_level -MAX_PROBES = 3 +# Public discriminator list. Four to six domains typically compete; 8 lets about +# three domains keep two years plus a couple of activation fallbacks without +# flooding the ask layer, which still ranks globally by information_gain. +MAX_PROBES = 8 +# Collection still asks at most three missing-domain age-band questions. +MAX_COLLECTION_PROBES = 3 +# N: keep the top scored probes per domain (boundary years, plus at most one +# activation if the domain is still under this cap). +MAX_PROBES_PER_DOMAIN = 3 +# K: per-domain evaluation budget of unique (year, month) windows. +# N <= K. Endpoints and evenly spaced months are tried first (at most K). +# If those K miss every discriminator, scan the remainder until the first +# hit so a lone month is not dropped; do not keep scanning to fill N. +# Shrink K if runtime more than doubles; do not drop multi-pair union. +# MAX_PROBES is the published cap after a global information_gain sort. +MAX_BOUNDARY_CANDIDATES_PER_DOMAIN = 8 MIN_BOUNDARY_DAYS = 45 LEVEL_RANK = {"none": 0, "weak": 1, "medium": 2, "strong": 3} LEVEL_P = {"none": 0.15, "weak": 0.35, "medium": 0.62, "strong": 0.82} @@ -284,59 +299,6 @@ def _remaining_contexts(built: dict[str, Any], candidate_times: Sequence[str]) - return remaining -def _pick_representatives( - built: dict[str, Any], - scan: dict[str, Any], - candidate_times: Sequence[str], - representative_time: str | None, -) -> tuple[dict[str, Any], dict[str, Any]] | None: - contexts = _static_contexts(built) - by_time = {_context_time(item): item for item in contexts} - times = [str(_context_time(item)) for item in contexts] - remaining: list[str] = [] - for raw in candidate_times: - time = str(raw or "")[:5] - if len(time) >= 5 and time in by_time and time not in remaining: - remaining.append(time) - if len(remaining) >= 2: - remaining_set = set(remaining) - for transition in scan.get("transitions") or []: - if not isinstance(transition, dict): - continue - layer = transition.get("layer") - at = str(transition.get("at") or "")[:5] - if layer not in SCORING_LAYERS or at not in by_time or at not in remaining_set: - continue - left = None - for time in remaining: - if _clock(time) < _clock(at): - left = time - if left and left != at: - return by_time[left], by_time[at] - return by_time[remaining[0]], by_time[remaining[-1]] - for transition in scan.get("transitions") or []: - if not isinstance(transition, dict): - continue - layer = transition.get("layer") - at = str(transition.get("at") or "")[:5] - if layer not in SCORING_LAYERS or at not in by_time: - continue - index = times.index(at) - left = times[index - 1] if index > 0 else at - if left != at: - return by_time[left], by_time[at] - picked: list[str] = [] - for raw in [*candidate_times, representative_time]: - time = str(raw or "")[:5] - if len(time) >= 5 and time in by_time and time not in picked: - picked.append(time) - if len(picked) >= 2: - return by_time[picked[0]], by_time[picked[-1]] - if len(times) >= 2: - return by_time[times[0]], by_time[times[-1]] - return None - - def _scoreable(context: dict[str, Any]) -> bool: chart = context.get("chart") planets = context.get("planet_longitudes") @@ -480,6 +442,196 @@ def _boundary_years(left: list[int], right: list[int]) -> set[int]: return {item.year for item in _boundary_windows(left, right)} +def _representative_pairs( + reps: Sequence[dict[str, Any]], +) -> list[tuple[dict[str, Any], dict[str, Any]]]: + ordered = sorted( + [item for item in reps if _scoreable(item) and _context_time(item)], + key=lambda item: _clock(str(_context_time(item))), + ) + if len(ordered) < 2: + return [] + pairs: list[tuple[dict[str, Any], dict[str, Any]]] = [] + seen: set[tuple[str, str]] = set() + + def add(left: dict[str, Any], right: dict[str, Any]) -> None: + left_time = str(_context_time(left) or "") + right_time = str(_context_time(right) or "") + if not left_time or not right_time or left_time == right_time: + return + key = (left_time, right_time) if left_time < right_time else (right_time, left_time) + if key in seen: + return + seen.add(key) + pairs.append((left, right)) + + for index in range(len(ordered) - 1): + add(ordered[index], ordered[index + 1]) + if len(ordered) > 2: + add(ordered[0], ordered[-1]) + return pairs + + +def _vim_cache_key(birth_date: str, moon: float, lo: int, hi: int) -> tuple[Any, ...]: + return (birth_date, round(float(moon), 6), lo, hi) + + +def _narayana_cache_key( + ascendant_index: int, + planet_longitudes: dict[str, Any], + birth_date: str, + lo: int, + hi: int, +) -> tuple[Any, ...]: + planet_key = tuple( + sorted( + (str(name), round(float(lon), 6)) + for name, lon in planet_longitudes.items() + if isinstance(lon, (int, float)) + ) + ) + return (int(ascendant_index), planet_key, birth_date, lo, hi) + + +def _union_boundary_dates( + reps: Sequence[dict[str, Any]], + *, + birth_date: str, + lo: int, + hi: int, +) -> list[date]: + vim_cache: dict[tuple[Any, ...], list[date]] = {} + narayana_cache: dict[tuple[Any, ...], list[date] | None] = {} + dates_by_key: dict[tuple[int, int], date] = {} + for left, right in _representative_pairs(reps): + left_moon = float(left["planet_longitudes"]["Moon"]) + right_moon = float(right["planet_longitudes"]["Moon"]) + left_vim_key = _vim_cache_key(birth_date, left_moon, lo, hi) + right_vim_key = _vim_cache_key(birth_date, right_moon, lo, hi) + if left_vim_key not in vim_cache: + vim_cache[left_vim_key] = _vim_start_dates(birth_date, left_moon, lo, hi) + if right_vim_key not in vim_cache: + vim_cache[right_vim_key] = _vim_start_dates(birth_date, right_moon, lo, hi) + windows = list(_boundary_windows(vim_cache[left_vim_key], vim_cache[right_vim_key])) + left_nara_key = _narayana_cache_key( + int(left["ascendant_index"]), + left["planet_longitudes"], + birth_date, + lo, + hi, + ) + right_nara_key = _narayana_cache_key( + int(right["ascendant_index"]), + right["planet_longitudes"], + birth_date, + lo, + hi, + ) + if left_nara_key not in narayana_cache: + narayana_cache[left_nara_key] = _narayana_start_dates( + int(left["ascendant_index"]), + left["planet_longitudes"], + birth_date, + lo, + hi, + ) + if right_nara_key not in narayana_cache: + narayana_cache[right_nara_key] = _narayana_start_dates( + int(right["ascendant_index"]), + right["planet_longitudes"], + birth_date, + lo, + hi, + ) + left_narayana = narayana_cache[left_nara_key] + right_narayana = narayana_cache[right_nara_key] + if left_narayana is not None and right_narayana is not None: + windows.extend(_boundary_windows(left_narayana, right_narayana)) + for item in windows: + dates_by_key.setdefault((item.year, item.month), item) + return sorted(dates_by_key.values(), key=lambda item: (item.year, item.month)) + + +def _bounded_candidate_dates(dates: Sequence[date], limit: int) -> list[date]: + items = list(dates) + if limit <= 0 or len(items) <= limit: + return items + if limit == 1: + return items[:1] + picked: list[date] = [] + seen: set[tuple[int, int]] = set() + last_index = len(items) - 1 + for step in range(limit): + index = (step * last_index + (limit - 1) // 2) // (limit - 1) + item = items[index] + key = (item.year, item.month) + if key in seen: + continue + seen.add(key) + picked.append(item) + if len(picked) < limit: + for item in items: + key = (item.year, item.month) + if key in seen: + continue + seen.add(key) + picked.append(item) + if len(picked) >= limit: + break + return picked + + +def _evaluation_order(dates: Sequence[date], limit: int) -> list[date]: + sampled = _bounded_candidate_dates(dates, limit) + sampled_keys = {(item.year, item.month) for item in sampled} + remainder = [item for item in dates if (item.year, item.month) not in sampled_keys] + return sampled + remainder + + +def _probe_sort_key(row: dict[str, Any]) -> tuple[float, str]: + return (-float(row.get("information_gain") or 0), str(row.get("semantic_key") or "")) + + +def _best_probe_per_year(rows: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: + by_year: dict[int, dict[str, Any]] = {} + for row in sorted(rows, key=_probe_sort_key): + year = int(row["year"]) + if year not in by_year: + by_year[year] = row + return list(by_year.values()) + + +def _try_activation_probe( + *, + reps: Sequence[dict[str, Any]], + birth_date: str, + birth_year: int, + domain: str, + now: date, + blocked_years: set[int], + holdout_keys: set[str], + clusters: Sequence[dict[str, Any]], + set_version: str, +) -> dict[str, Any] | None: + midpoint = _age_band_year(birth_year, domain, now) + if midpoint is None or midpoint in blocked_years or f"{domain}:{midpoint}" in holdout_keys: + return None + found = _evaluate_contexts( + reps, + birth_date=birth_date, + domain=domain, + year=midpoint, + source="dasha_activation", + clusters=clusters, + set_version=set_version, + ) + if found is None or distinguish_contract_errors(found): + return None + if not isinstance(found.get("year"), int) or int(found["year"]) <= 0: + return None + return found + + def _score_year( context: dict[str, Any], *, @@ -906,7 +1058,7 @@ def evidence_collection_probes( ), event_family=str(DOMAIN_CATALOG[domain]["event_family"]), )) - if len(rows) >= MAX_PROBES: + if len(rows) >= MAX_COLLECTION_PROBES: break return rows @@ -984,27 +1136,7 @@ def discriminating_event_probes( if not domains: return [] lo, hi = birth_year + 5, min(now.year, birth_year + 80) - left, right = reps[0], reps[-1] - left_moon = float(left["planet_longitudes"]["Moon"]) - right_moon = float(right["planet_longitudes"]["Moon"]) - vim_windows = _boundary_windows( - _vim_start_dates(birth_date, left_moon, lo, hi), - _vim_start_dates(birth_date, right_moon, lo, hi), - ) - left_narayana = _narayana_start_dates(int(left["ascendant_index"]), left["planet_longitudes"], birth_date, lo, hi) - right_narayana = _narayana_start_dates(int(right["ascendant_index"]), right["planet_longitudes"], birth_date, lo, hi) - narayana_windows: list[date] = [] - if left_narayana is not None and right_narayana is not None: - narayana_windows = _boundary_windows(left_narayana, right_narayana) - boundary_dates: list[date] = [] - seen_windows: set[tuple[int, int]] = set() - for item in [*vim_windows, *narayana_windows]: - key = (item.year, item.month) - if key in seen_windows: - continue - seen_windows.add(key) - boundary_dates.append(item) - boundary_dates.sort() + boundary_dates = _union_boundary_dates(reps, birth_date=birth_date, lo=lo, hi=hi) probes: list[dict[str, Any]] = [] for domain in domains: if domain not in DOMAIN_CATALOG: @@ -1012,14 +1144,20 @@ def discriminating_event_probes( known_years = _event_years(events, domain) blocked_years = _existence_blocked_years(domain, known_years) domain_lo = _domain_year_floor(birth_year, domain, lo) - boundary = [item for item in boundary_dates if domain_lo <= item.year <= hi] - best = None - for at in boundary: - if at.year in blocked_years: - continue - if f"{domain}:{at.year}" in holdout_keys: - continue - found = _evaluate_contexts( + eligible = [ + item + for item in boundary_dates + if domain_lo <= item.year <= hi + and item.year not in blocked_years + and f"{domain}:{item.year}" not in holdout_keys + ] + found: list[dict[str, Any]] = [] + evaluated = 0 + sample_size = min(MAX_BOUNDARY_CANDIDATES_PER_DOMAIN, len(eligible)) + for at in _evaluation_order(eligible, MAX_BOUNDARY_CANDIDATES_PER_DOMAIN): + if evaluated >= sample_size and found: + break + row = _evaluate_contexts( reps, birth_date=birth_date, domain=domain, @@ -1029,25 +1167,42 @@ def discriminating_event_probes( clusters=clusters, set_version=set_version, ) - if found is None: + evaluated += 1 + if row is None or distinguish_contract_errors(row): continue - if best is None or float(found["information_gain"]) > float(best["information_gain"]): - best = found - if best is None: - midpoint = _age_band_year(birth_year, domain, now) - if midpoint is not None and midpoint not in blocked_years and f"{domain}:{midpoint}" not in holdout_keys: - best = _evaluate_contexts( - reps, - birth_date=birth_date, - domain=domain, - year=midpoint, - source="dasha_activation", - clusters=clusters, - set_version=set_version, + if not isinstance(row.get("year"), int) or int(row["year"]) <= 0: + continue + found.append(row) + if evaluated > sample_size: + break + kept = _best_probe_per_year(found)[:MAX_PROBES_PER_DOMAIN] + if len(kept) < MAX_PROBES_PER_DOMAIN: + activation = _try_activation_probe( + reps=reps, + birth_date=birth_date, + birth_year=birth_year, + domain=domain, + now=now, + blocked_years=blocked_years, + holdout_keys=set(holdout_keys), + clusters=clusters, + set_version=set_version, + ) + if activation is not None: + activation_key = ( + str(activation["domain"]), + int(activation["year"]), + int(activation.get("month") or 0), + str(activation["source"]), ) - if best and not distinguish_contract_errors(best): - probes.append(best) - probes.sort(key=lambda row: (-float(row.get("information_gain") or 0), str(row.get("semantic_key") or ""))) + existing = { + (str(row["domain"]), int(row["year"]), int(row.get("month") or 0), str(row["source"])) + for row in kept + } + if activation_key not in existing: + kept.append(activation) + probes.extend(kept) + probes.sort(key=_probe_sort_key) public: list[dict[str, Any]] = [] seen: set[tuple[str, int, int, str]] = set() for row in probes: @@ -1055,6 +1210,8 @@ def discriminating_event_probes( continue if distinguish_contract_errors(row): continue + if not isinstance(row.get("year"), int) or int(row["year"]) <= 0: + continue key = (str(row["domain"]), int(row["year"]), int(row.get("month") or 0), str(row["source"])) encoded = str(row) if key in seen or "points" in encoded: diff --git a/tests/test_rectification_event_probes.py b/tests/test_rectification_event_probes.py index 1885b733..5a2f5c2d 100644 --- a/tests/test_rectification_event_probes.py +++ b/tests/test_rectification_event_probes.py @@ -5,6 +5,10 @@ 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, @@ -134,6 +138,25 @@ def _probes(request: dict, built: dict, times: list[str], representative: str, * ) +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( @@ -247,6 +270,7 @@ class EventProbesTest(unittest.TestCase): 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)) @@ -404,7 +428,7 @@ class EventProbesTest(unittest.TestCase): 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), 3) + self.assertLessEqual(len(probes), MAX_PROBES) for probe in probes: self.assertEqual(distinguish_contract_errors(probe), []) @@ -639,6 +663,239 @@ class EventProbesTest(unittest.TestCase): 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()