root/Jyotisha
1
0
Fork 0
Code Issues 24 Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
8e0503f9c05c2b11274fe566ac7e842e1a40d98e
Jyotisha/scripts/rectification/event_probes.py
T
jesse-uxandGrok (agent loop) 8e0503f9c0 Speed up rectification scoring without changing results. 
Cache work that depends only on the candidate minute, drop unread response fields, and reject windows above 241 candidates before any chart is built. Eleven canonical scenario hashes match 77057a819. Sealed and reported-offset replays match the 2026-10-08 records (changed_trial_count 8 and 581) and are stored as scoring_perf_2026_10_13.

Co-Authored-By: Grok (agent loop)
2026-10-14 02:06:14 +08:00

2102 lines
74 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Public-safe biographical probes from candidate dasha / varga differences.
Discriminators are feature-signature clusters over the full birth window.
known_event_quality may distinguish when signature groups disagree on the
event's theme varga type, the probe is anchored to confirmed evidence, and
the case cap is respected. Unanchored quality stays clarification-only.
"""
from __future__ import annotations
import re
from datetime import date, datetime, timedelta
from math import log2
from typing import Any, Sequence
from scripts.active_rectification_event_engine import (
DOMAIN_CONFIG,
_active_narayana,
_active_vimshottari,
_score_event,
)
import dasha_analyzer
import narayana_dasha
from scripts.rectification.candidate_contrast import (
PROBE_PHASE_CANDIDATE_DISCRIMINATOR,
PROBE_PHASE_EVENT_CLARIFICATION,
PROBE_PHASE_EVIDENCE_COLLECTION,
SIGNATURE_LAYERS,
assert_distinguish_contract,
candidate_ids_from_outcomes,
candidate_set_version,
candidate_split_hash,
cluster_contexts_by_signature,
discriminator_gate_open,
distinguish_contract_errors,
event_year,
expand_times_through_clusters,
missing_collection_domains,
opportunity_from_probe,
)
from scripts.rectification.case_holdout import holdout_domain_years, holdout_event_ids
from scripts.rectification.probe_question_contract import (
QUESTION_CONTRACT_VERSION,
completed_style_options,
)
from scripts.rectification.refinement_packet import match_level
# 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
# Refresh against a small remaining candidate set may keep a few more years.
REFRESH_MAX_PROBES = 12
REFRESH_MAX_PROBES_PER_DOMAIN = 4
REFRESH_REMAINING_CAP = 5
# Collection asks an age-band cue for every missing catalog domain.
MAX_COLLECTION_PROBES = 7
# 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
REFRESH_MIN_BOUNDARY_DAYS = 30
MAX_QUALITY_DISTINGUISH_PROBES = 2
QUALITY_DISTINGUISH_EVENT_KINDS = frozenset({
"education_start",
"education_change",
"education_interruption",
})
ANSWER_PRIOR_TABLE_VERSION = "rectification-answer-priors-v1"
DOMINANT_ANSWER_PRIOR = 0.8
# Conservative population rates, not fitted from product users.
# Broad family existence (yearless / 3-year window / age band) is near-certain
# in adult life. Specified-year long-distance move is uncommon. Quality answers
# are closer to even because they condition on a known event.
_DEFAULT_EXISTENCE_PRIORS = {"yes": 0.35, "weak_yes": 0.15, "no": 0.40, "unsure": 0.10}
_DEFAULT_QUALITY_PRIORS = {"yes": 0.30, "weak_yes": 0.20, "no": 0.40, "unsure": 0.10}
ANSWER_PRIORS: dict[tuple[str, str], dict[str, float]] = {
("family", "existence"): {"yes": 0.85, "weak_yes": 0.05, "no": 0.05, "unsure": 0.05},
("relocation", "existence"): {"yes": 0.20, "weak_yes": 0.10, "no": 0.60, "unsure": 0.10},
("education", "existence"): {"yes": 0.45, "weak_yes": 0.15, "no": 0.30, "unsure": 0.10},
("relationship", "existence"): {"yes": 0.40, "weak_yes": 0.15, "no": 0.35, "unsure": 0.10},
("career", "existence"): {"yes": 0.40, "weak_yes": 0.15, "no": 0.35, "unsure": 0.10},
("education", "event_quality"): dict(_DEFAULT_QUALITY_PRIORS),
("relationship", "event_quality"): dict(_DEFAULT_QUALITY_PRIORS),
("career", "event_quality"): dict(_DEFAULT_QUALITY_PRIORS),
("relocation", "event_quality"): dict(_DEFAULT_QUALITY_PRIORS),
("family", "event_quality"): dict(_DEFAULT_QUALITY_PRIORS),
}
DOMAIN_QUALITY_LAYER = {
"education": "d24",
"relationship": "d9",
"career": "d10",
"relocation": "d4",
"family": "d12",
}
QUALITY_DISTINGUISH_OPTIONS: dict[str, tuple[dict[str, str], ...]] = {
"education": (
{"label": "发挥明显失常", "answer_class": "yes"},
{"label": "只是将就调剂", "answer_class": "weak_yes"},
{"label": "基本如愿录取", "answer_class": "no"},
{"label": "当时说不清楚", "answer_class": "unsure"},
),
"relationship": (
{"label": "明显受挫变难", "answer_class": "yes"},
{"label": "只是将就相处", "answer_class": "weak_yes"},
{"label": "整体比较顺利", "answer_class": "no"},
{"label": "当时说不清楚", "answer_class": "unsure"},
),
"career": (
{"label": "明显受挫受压", "answer_class": "yes"},
{"label": "只是将就应付", "answer_class": "weak_yes"},
{"label": "整体比较顺利", "answer_class": "no"},
{"label": "当时说不清楚", "answer_class": "unsure"},
),
"relocation": (
{"label": "搬迁特别折腾", "answer_class": "yes"},
{"label": "只是将就安顿", "answer_class": "weak_yes"},
{"label": "整体比较顺利", "answer_class": "no"},
{"label": "当时说不清楚", "answer_class": "unsure"},
),
"family": (
{"label": "家里特别操心", "answer_class": "yes"},
{"label": "只是普通操心", "answer_class": "weak_yes"},
{"label": "整体比较顺利", "answer_class": "no"},
{"label": "当时说不清楚", "answer_class": "unsure"},
),
}
LEVEL_RANK = {"none": 0, "weak": 1, "medium": 2, "strong": 3}
LEVEL_P = {"none": 0.15, "weak": 0.35, "medium": 0.62, "strong": 0.82}
SCORING_LAYERS = ("d1", "d9", "d10", "d4", "d5", "d24", "d7", "d12", "d2", "d11", "d30")
LAYER_DOMAIN = {
"d9": "relationship",
"d10": "career",
"d4": "relocation",
"d5": "education",
"d24": "education",
"d7": "family",
"d12": "family",
"d2": "finance",
"d11": "finance",
"d30": "health_pressure",
}
LAYER_VARGA = {
"d9": "D9",
"d10": "D10",
"d4": "D4",
"d5": "D5",
"d24": "D24",
"d7": "D7",
"d12": "D12",
"d2": "D2",
"d11": "D11",
"d30": "D30",
}
DOMAIN_CATALOG: dict[str, dict[str, Any]] = {
"education": {
"event_family": "升学、转学或换学习环境",
"quality_family": "学业或考试发挥失常、压力特别大",
"kind": "education_milestone",
"age_lo": 16,
"age_hi": 18,
"varga": "D5 / D24",
},
"relocation": {
"event_family": "搬家或长期住到外地",
"quality_family": "搬家或住处特别折腾",
"kind": "home_change",
"age_lo": 18,
"age_hi": 24,
"varga": "D4",
},
"relationship": {
"event_family": "开始一段认真关系、分手或结婚",
"quality_family": "感情里压力特别大或相处明显变难",
"kind": "relationship_change",
"age_lo": 21,
"age_hi": 26,
"varga": "D9",
},
"career": {
"event_family": "入职、换工作或职责加重",
"quality_family": "职责加重或工作压力特别大",
"kind": "career_change",
"age_lo": 22,
"age_hi": 30,
"varga": "D10",
},
"family": {
"event_family": "家人结婚、添丁或住院",
"quality_family": "家人结婚、添丁、住院或家里特别操心",
"kind": "family_event",
"age_lo": 18,
"age_hi": 30,
"varga": "D12 / D7 / D3",
},
"finance": {
"event_family": "收入明显变化、大笔支出或欠债",
"quality_family": "收入或财务压力特别大",
"kind": "finance_change",
"age_lo": 22,
"age_hi": 32,
"varga": "D2 / D11",
},
"health_pressure": {
"event_family": "生病、受伤或压力特别大",
"quality_family": "生病、受伤或压力特别大",
"kind": "self_health_event",
"age_lo": 16,
"age_hi": 40,
"varga": "D30",
},
}
# 入学年与高考通常同一学年或前后一年；感情/事业/搬家的邻近年也常是同一段经历。
EXISTENCE_NEARBY_YEARS = {
"education": 1,
"relationship": 1,
"career": 1,
"relocation": 1,
}
_SEMANTIC_YEAR = re.compile(r"^(?P<domain>[a-z_]+)\.(?P<year>(?:19|20)\d{2})(?:\.|$)")
def canonical_domain(raw: object) -> str:
value = str(raw or "").strip()
if value == "health":
return "health_pressure"
return value
def asked_years_for_domain(asked_probe_keys: Sequence[str] | None, domain: str) -> set[int]:
years: set[int] = set()
prefix = f"{domain}."
for raw in asked_probe_keys or []:
key = str(raw or "").strip()
if not key.startswith(prefix):
continue
match = _SEMANTIC_YEAR.match(key)
if match and match.group("domain") == domain:
years.add(int(match.group("year")))
return years
def _clock(value: str) -> int:
return int(value[:2]) * 60 + int(value[3:5])
def _context_time(context: dict[str, Any]) -> str | None:
feature = context.get("feature") if isinstance(context.get("feature"), dict) else {}
raw = feature.get("time")
if isinstance(raw, str) and len(raw) >= 5:
return raw[:5]
at = context.get("candidate_at")
if isinstance(at, datetime):
return at.strftime("%H:%M")
return None
def _birth_year(value: object) -> int | None:
text = str(value or "").strip()
if len(text) < 4 or not text[:4].isdigit():
return None
year = int(text[:4])
return year if 1900 <= year <= 2100 else None
def _event_year(event: dict[str, Any]) -> int | None:
return event_year(event)
def _period_label(year: int, month: int | None = None) -> str:
if isinstance(month, int) and 1 <= month <= 12:
return f"{year} 年 {month} 月前后"
return f"{year} 年前后"
def _year_label(year: int) -> str:
return _period_label(year)
# Family events (parents marrying, illness) can occur in childhood.
_CHILDHOOD_OK_DOMAINS = frozenset({"family"})
def _domain_year_floor(birth_year: int, domain: str, global_lo: int) -> int:
if domain in _CHILDHOOD_OK_DOMAINS:
return global_lo
catalog = DOMAIN_CATALOG.get(domain)
if not catalog:
return global_lo
return max(global_lo, birth_year + int(catalog["age_lo"]))
def _age_band_year(birth_year: int, domain: str, today: date) -> int | None:
catalog = DOMAIN_CATALOG.get(domain)
if not catalog:
return None
age = (int(catalog["age_lo"]) + int(catalog["age_hi"])) // 2
year = birth_year + age
latest = min(today.year, birth_year + 80)
earliest = birth_year + 5
if year < earliest or year > latest:
return None
return year
def _layer_value(context: dict[str, Any], layer: str) -> int | None:
feature = context.get("feature") if isinstance(context.get("feature"), dict) else {}
if layer == "d1":
raw = feature.get("ascendant_sign_index")
if isinstance(raw, int):
return raw
index = context.get("ascendant_index")
return index if isinstance(index, int) else None
name = LAYER_VARGA.get(layer)
if not name:
return None
vargas = feature.get("varga_ascendants") if isinstance(feature.get("varga_ascendants"), dict) else {}
raw = vargas.get(name)
if isinstance(raw, int):
return raw
charts = context.get("varga_charts") if isinstance(context.get("varga_charts"), dict) else {}
chart = charts.get(name) if isinstance(charts.get(name), dict) else {}
ascendant = chart.get("Ascendant") if isinstance(chart.get("Ascendant"), dict) else {}
index = ascendant.get("sign_idx")
return index if isinstance(index, int) else None
def _differing_layers(contexts: Sequence[dict[str, Any]]) -> set[str]:
values: dict[str, set[int]] = {layer: set() for layer in SCORING_LAYERS}
for context in contexts:
for layer in SCORING_LAYERS:
value = _layer_value(context, layer)
if isinstance(value, int):
values[layer].add(value)
return {layer for layer, bucket in values.items() if len(bucket) > 1}
def _probe_caps(*, refresh: bool, remaining_count: int) -> tuple[int, int]:
if refresh and remaining_count <= REFRESH_REMAINING_CAP:
return REFRESH_MAX_PROBES, REFRESH_MAX_PROBES_PER_DOMAIN
return MAX_PROBES, MAX_PROBES_PER_DOMAIN
def _monthly_family_dasha_boundary(probe: dict[str, Any]) -> bool:
if str(probe.get("source") or "") != "dasha_boundary":
return False
if str(probe.get("domain") or "") != "family":
return False
month = probe.get("month")
return isinstance(month, int) and 1 <= month <= 12
def _probe_domains(
remaining_layers: set[str],
_events: Sequence[dict[str, Any]],
*,
d1_differs: bool = False,
) -> list[str]:
ordered: list[str] = []
for layer in SCORING_LAYERS:
domain = LAYER_DOMAIN.get(layer)
if not domain or domain in ordered:
continue
if layer not in remaining_layers:
continue
ordered.append(domain)
if not ordered and d1_differs:
ordered.append("education")
return ordered
def _static_contexts(built: dict[str, Any]) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
for context in built.get("static_contexts") or []:
if isinstance(context, dict) and _context_time(context):
rows.append(context)
rows.sort(key=lambda item: _clock(str(_context_time(item))))
return rows
def _remaining_contexts(built: dict[str, Any], candidate_times: Sequence[str]) -> list[dict[str, Any]]:
by_time = {_context_time(item): item for item in _static_contexts(built)}
remaining: list[dict[str, Any]] = []
seen: set[str] = set()
for raw in candidate_times:
time = str(raw or "")[:5]
context = by_time.get(time)
if context is None or time in seen:
continue
seen.add(time)
remaining.append(context)
return remaining
def _scoreable(context: dict[str, Any]) -> bool:
chart = context.get("chart")
planets = context.get("planet_longitudes")
vargas = context.get("varga_charts")
return (
isinstance(chart, dict)
and isinstance(planets, dict)
and isinstance(vargas, dict)
and isinstance(planets.get("Moon"), (int, float))
and isinstance(context.get("ascendant_index"), int)
)
def _candidate_at(context: dict[str, Any], birth_date: str) -> datetime | None:
at = context.get("candidate_at")
if isinstance(at, datetime):
return at
time = _context_time(context)
if not time:
return None
try:
return datetime.strptime(f"{birth_date} {time}", "%Y-%m-%d %H:%M")
except ValueError:
return None
def _has_domain_activation(rule_ids: Sequence[str]) -> bool:
return any(
("_domain_" in str(item) or str(item).endswith("_domain_house") or str(item).endswith("_domain_lord") or str(item).endswith("_domain_varga"))
and not str(item).startswith("no_")
for item in rule_ids
)
def _discriminates(left: Sequence[str], right: Sequence[str]) -> bool:
left_rank = LEVEL_RANK.get(match_level(left), 0)
right_rank = LEVEL_RANK.get(match_level(right), 0)
if abs(left_rank - right_rank) >= 2:
return True
return _has_domain_activation(left) != _has_domain_activation(right)
def _tracks_present(rule_ids: Sequence[str]) -> tuple[bool, bool]:
text = [str(item) for item in rule_ids]
return (
any(item.startswith("vim_") for item in text),
any(item.startswith("narayana_") for item in text),
)
def _vim_start_dates(
birth_date: str,
moon_longitude: float,
lo: int,
hi: int,
*,
include_pratyantar: bool = False,
) -> list[date]:
nakshatra, progress, _ = dasha_analyzer.lon_to_nakshatra(float(moon_longitude))
timeline, _, _, _ = dasha_analyzer.build_dasha_timeline(birth_date, nakshatra, progress)
starts: list[date] = []
for major in timeline:
start = major.get("start")
if isinstance(start, datetime) and lo <= start.year <= hi:
starts.append(start.date())
for minor in dasha_analyzer.build_antardasha(major):
minor_start = minor.get("start")
if isinstance(minor_start, datetime) and lo <= minor_start.year <= hi:
starts.append(minor_start.date())
if not include_pratyantar:
continue
for prat in dasha_analyzer.build_antardasha(minor):
prat_start = prat.get("start")
if isinstance(prat_start, datetime) and lo <= prat_start.year <= hi:
starts.append(prat_start.date())
return starts
def _vim_start_years(birth_date: str, moon_longitude: float, lo: int, hi: int) -> list[int]:
return [item.year for item in _vim_start_dates(birth_date, moon_longitude, lo, hi)]
def _narayana_start_dates(
ascendant_index: int,
planet_longitudes: dict[str, float],
birth_date: str,
lo: int,
hi: int,
) -> list[date] | None:
periods = narayana_dasha.calc_narayana_mahadasha(ascendant_index, planet_longitudes)
if not periods:
return None
birth = datetime.strptime(birth_date, "%Y-%m-%d")
starts: list[date] = []
for major in periods:
start_age = major.get("start_age")
if not isinstance(start_age, (int, float)):
return None
at = (birth + timedelta(days=float(start_age) * 365.2425)).date()
if lo <= at.year <= hi:
starts.append(at)
antars = narayana_dasha.calc_narayana_antardasha(periods, int(major["sign_idx"]))
for minor in antars:
minor_age = minor.get("start_age")
if not isinstance(minor_age, (int, float)):
continue
minor_at = (birth + timedelta(days=float(minor_age) * 365.2425)).date()
if lo <= minor_at.year <= hi:
starts.append(minor_at)
return starts
def _narayana_start_years(
ascendant_index: int,
planet_longitudes: dict[str, float],
birth_date: str,
lo: int,
hi: int,
) -> list[int] | None:
starts = _narayana_start_dates(ascendant_index, planet_longitudes, birth_date, lo, hi)
return None if starts is None else [item.year for item in starts]
def _as_start_date(value: date | datetime | int) -> date | None:
if isinstance(value, datetime):
return value.date()
if isinstance(value, date):
return value
if isinstance(value, int) and 1900 <= value <= 2100:
return date(value, 7, 1)
return None
def _boundary_windows(
left: Sequence[date | datetime | int],
right: Sequence[date | datetime | int],
*,
min_days: int | None = None,
) -> list[date]:
windows: list[date] = []
seen: set[tuple[int, int]] = set()
threshold = MIN_BOUNDARY_DAYS if min_days is None else min_days
for raw_left, raw_right in zip(left, right):
one = _as_start_date(raw_left)
two = _as_start_date(raw_right)
if one is None or two is None:
continue
if one.year == two.year and abs((one - two).days) < threshold:
continue
for item in (one, two):
key = (item.year, item.month)
if key in seen:
continue
seen.add(key)
windows.append(item)
return windows
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,
min_boundary_days: int | None = None,
include_pratyantar: bool = False,
varga_narayana: bool = False,
) -> list[date]:
threshold = MIN_BOUNDARY_DAYS if min_boundary_days is None else min_boundary_days
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), include_pratyantar)
right_vim_key = (*_vim_cache_key(birth_date, right_moon, lo, hi), include_pratyantar)
if left_vim_key not in vim_cache:
vim_cache[left_vim_key] = _vim_start_dates(
birth_date, left_moon, lo, hi, include_pratyantar=include_pratyantar,
)
if right_vim_key not in vim_cache:
vim_cache[right_vim_key] = _vim_start_dates(
birth_date, right_moon, lo, hi, include_pratyantar=include_pratyantar,
)
windows = list(_boundary_windows(
vim_cache[left_vim_key],
vim_cache[right_vim_key],
min_days=threshold,
))
layers: list[str | None] = [None]
if varga_narayana:
layers.extend(["d9", "d10"])
for layer in layers:
if layer is None:
left_asc = int(left["ascendant_index"])
right_asc = int(right["ascendant_index"])
else:
left_raw = _layer_value(left, layer)
right_raw = _layer_value(right, layer)
if not isinstance(left_raw, int) or not isinstance(right_raw, int):
continue
left_asc = left_raw
right_asc = right_raw
left_nara_key = (*_narayana_cache_key(
left_asc, left["planet_longitudes"], birth_date, lo, hi,
), layer)
right_nara_key = (*_narayana_cache_key(
right_asc, right["planet_longitudes"], birth_date, lo, hi,
), layer)
if left_nara_key not in narayana_cache:
narayana_cache[left_nara_key] = _narayana_start_dates(
left_asc, left["planet_longitudes"], birth_date, lo, hi,
)
if right_nara_key not in narayana_cache:
narayana_cache[right_nara_key] = _narayana_start_dates(
right_asc, 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,
min_days=threshold,
))
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],
*,
birth_date: str,
domain: str,
year: int,
month: int | None = None,
) -> dict[str, Any] | None:
catalog = DOMAIN_CATALOG[domain]
prefixes, _ = DOMAIN_CONFIG[domain]
varga_charts = context.get("varga_charts") or {}
domain_vargas = [varga_charts.get(prefix) for prefix in prefixes]
if any(item is None for item in domain_vargas):
return None
candidate_at = _candidate_at(context, birth_date)
moon = (context.get("planet_longitudes") or {}).get("Moon")
if candidate_at is None or not isinstance(moon, (int, float)):
return None
month_value = month if isinstance(month, int) and 1 <= month <= 12 else None
event_at = datetime(year, month_value, 15) if month_value else datetime(year, 7, 1)
event = {
"id": f"probe-{domain}-{year}" + (f"-{month_value:02d}" if month_value else ""),
"domain": domain,
"event_kind": catalog["kind"],
"date": f"{year}-{month_value:02d}-15" if month_value else f"{year}-07-01",
"precision": "month" if month_value else "year",
"summary": catalog["event_family"],
}
try:
vimshottari = _active_vimshottari(birth_date, float(moon), event_at)
narayana = _active_narayana(
int(context["ascendant_index"]),
context["planet_longitudes"],
candidate_at,
event_at,
)
except (KeyError, TypeError, ValueError):
return None
if narayana[0] is None or narayana[1] is None:
return None
return _score_event(
candidate_time=str(_context_time(context)),
event=event,
natal_chart=context["chart"],
varga_charts=[item for item in domain_vargas if item is not None],
vimshottari=vimshottari,
narayana=narayana,
arudha_padas=context.get("arudha_padas") or {},
functional=context.get("functional_profile"),
)
def _agent_brief(
*,
year_label: str,
domain: str,
family: str,
quality: bool = False,
exam: bool = False,
nearby_note: str = "",
) -> str:
nearby = nearby_note.strip()
if exam:
return (
f"时间范围锁定 {year_label}；领域锁定 {domain}。"
"语义目标是那次考试的实际体验。结合最近对话，只选一个容易回答的口语入口，"
"问是否明显失常或压力很大；不要堆叠例子，不得改时间范围。"
+ (f"{nearby}" if nearby else "")
)
if quality:
return (
f"时间范围锁定 {year_label}；领域锁定 {domain}；语义目标是 {family}。"
"结合最近对话，只选一个容易回答的口语入口来核对体验；"
"不要逐字复述语义目标，不要堆叠例子，不得改时间范围。"
+ (f"{nearby}" if nearby else "")
)
lead = (
f"{nearby}时间范围锁定 {year_label}；请问用户那段时间身上发生了什么变化；"
if nearby
else f"时间范围锁定 {year_label}；领域锁定 {domain}；语义目标是 {family}。"
"结合最近对话，只选一个容易回答的口语入口，写一句自然的是/否题；"
)
if nearby:
return (
lead
+ f"领域锁定 {domain}；语义目标是 {family}。"
"选项由服务端给出；不要发明年份，不得改时间范围。"
)
return (
lead
+ "不要逐字复述语义目标，不要把所有例子堆进一句，不得改时间范围。"
)
def _binary_entropy(probability: float) -> float:
if probability <= 0.0 or probability >= 1.0:
return 0.0
return -(probability * log2(probability) + (1.0 - probability) * log2(1.0 - probability))
def _pair_entropy(left: float, right: float) -> float:
total = left + right
if total <= 0:
return 0.0
return _binary_entropy(left / total)
def _group_entropy(sizes: Sequence[int]) -> float:
total = sum(int(item) for item in sizes)
if total <= 0:
return 0.0
return round(-sum((item / total) * log2(item / total) for item in sizes if item > 0), 4)
def _information_gain(left_level: str, right_level: str) -> float:
left_p = LEVEL_P.get(left_level, 0.5)
right_p = LEVEL_P.get(right_level, 0.5)
yes_p = 0.5 * left_p + 0.5 * right_p
after = yes_p * _pair_entropy(left_p, right_p) + (1.0 - yes_p) * _pair_entropy(1.0 - left_p, 1.0 - right_p)
return round(max(0.0, 1.0 - after), 4)
def _broad_existence_window(probe: dict[str, Any]) -> bool:
if str(probe.get("source") or "") == "age_band":
return True
year = probe.get("year")
if not isinstance(year, int) or year <= 0:
return True
span = probe.get("window_span_years")
return isinstance(span, int) and span >= 3
def _answer_priors_for(probe: dict[str, Any]) -> dict[str, float]:
kind = str(probe.get("choice_kind") or "existence")
if kind not in {"existence", "event_quality"}:
kind = "existence"
domain = str(probe.get("domain") or "")
if _monthly_family_dasha_boundary(probe):
return dict(ANSWER_PRIORS[("family", "existence")])
if domain == "family" and kind == "existence" and not _broad_existence_window(probe):
return dict(_DEFAULT_EXISTENCE_PRIORS)
priors = ANSWER_PRIORS.get((domain, kind))
if priors:
return dict(priors)
return dict(_DEFAULT_QUALITY_PRIORS if kind == "event_quality" else _DEFAULT_EXISTENCE_PRIORS)
def _expected_information_gain(raw_split_gain: float, priors: dict[str, float]) -> float:
weight = (
float(priors.get("yes") or 0)
+ float(priors.get("no") or 0)
+ 0.5 * float(priors.get("weak_yes") or 0)
)
return round(max(0.0, raw_split_gain * weight), 4)
def _apply_prior_ranking(probe: dict[str, Any]) -> dict[str, Any]:
raw = float(probe.get("information_gain") or 0)
priors = _answer_priors_for(probe)
probe["raw_split_gain"] = raw
probe["answer_priors"] = priors
probe["information_gain"] = _expected_information_gain(raw, priors)
return probe
def _dominant_existence_prior(probe: dict[str, Any], priors: dict[str, float]) -> bool:
if str(probe.get("choice_kind") or "existence") != "existence":
return False
if str(probe.get("source") or "") == "known_event_quality":
return False
if _monthly_family_dasha_boundary(probe):
return False
return max(priors.values()) > DOMINANT_ANSWER_PRIOR
def _event_month(event: dict[str, Any]) -> int | None:
raw = str(event.get("date") or event.get("date_start") or "")
if len(raw) >= 7 and raw[4] == "-":
try:
month = int(raw[5:7])
except ValueError:
return None
if 1 <= month <= 12:
return month
return None
def _display_date_label(event: dict[str, Any]) -> str:
year = _event_year(event)
# A year-precision event is stored as YYYY-01-01; its month is not known (BUG-1088).
month = None if str(event.get("precision") or "") == "year" else _event_month(event)
if year is None:
return "那次"
if month:
return f"{year} 年 {month} 月"
return f"{year} 年"
def _event_month_index(year: int, month: int | None) -> int | None:
if month is None or not 1 <= month <= 12:
return None
return year * 12 + month
def nearby_ledger_note(
events: Sequence[dict[str, Any]],
*,
domain: str,
year: int,
month: int | None,
) -> str:
if year <= 0:
return ""
probe_index = _event_month_index(year, month)
best: dict[str, Any] | None = None
best_delta = 99
for event in events:
if not isinstance(event, dict):
continue
other_domain = str(event.get("domain") or "")
if other_domain == domain or other_domain not in DOMAIN_CATALOG:
continue
other_year = _event_year(event)
if other_year is None:
continue
other_month = _event_month(event)
other_index = _event_month_index(other_year, other_month)
if probe_index is not None and other_index is not None:
delta = abs(probe_index - other_index)
if delta > 2:
continue
elif other_year != year:
continue
else:
delta = 2 if probe_index is not None or other_index is not None else 0
if delta < best_delta:
best_delta = delta
best = event
if best is None:
return ""
family = str(DOMAIN_CATALOG[str(best.get("domain") or "")]["event_family"])
return f"账本里 { _display_date_label(best) } 有{family}；题干先提那件事再问。"
def _annotate_nearby_ledger(
probes: Sequence[dict[str, Any]],
events: Sequence[dict[str, Any]],
) -> None:
for probe in probes:
if not isinstance(probe, dict):
continue
if str(probe.get("choice_kind") or "existence") != "existence":
continue
if str(probe.get("source") or "") not in {"dasha_boundary", "dasha_activation"}:
continue
year = probe.get("year")
if not isinstance(year, int) or year <= 0:
continue
note = nearby_ledger_note(
events,
domain=str(probe.get("domain") or ""),
year=year,
month=int(probe["month"]) if isinstance(probe.get("month"), int) else None,
)
if not note:
continue
meaning = str(probe.get("user_meaning") or "")
if note not in meaning:
probe["user_meaning"] = f"{note}{meaning}"
def _event_kind_name(event: dict[str, Any]) -> str:
return str(event.get("event_kind") or event.get("kind") or "")
def _quality_event_allowed(event: dict[str, Any], domain: str) -> bool:
if domain != "education":
return False
return _event_kind_name(event) in QUALITY_DISTINGUISH_EVENT_KINDS
def _quality_split_sets(probe: dict[str, Any]) -> tuple[frozenset[str], frozenset[str]]:
outcomes = probe.get("expected_outcomes") or []
yes = next((row for row in outcomes if isinstance(row, dict) and row.get("answer_class") == "yes"), {})
no = next((row for row in outcomes if isinstance(row, dict) and row.get("answer_class") == "no"), {})
yes_times = yes.get("supports") if isinstance(yes, dict) else ()
no_times = no.get("supports") if isinstance(no, dict) else ()
return frozenset(str(item) for item in (yes_times or ())), frozenset(str(item) for item in (no_times or ()))
def _quality_probe_rank(probe: dict[str, Any]) -> tuple[float, int, int]:
year = int(probe["year"]) if isinstance(probe.get("year"), int) else 9999
month = int(probe["month"]) if isinstance(probe.get("month"), int) else 12
return (-float(probe.get("information_gain") or 0), year, month)
def _select_quality_distinguish_rows(candidates: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
best: dict[tuple[str, frozenset[str], frozenset[str]], dict[str, Any]] = {}
for probe in candidates:
key = (str(probe.get("domain") or ""), *_quality_split_sets(probe))
current = best.get(key)
if current is None or _quality_probe_rank(probe) < _quality_probe_rank(current):
best[key] = probe
selected = list(best.values())
selected.sort(key=_quality_probe_rank)
return selected[:MAX_QUALITY_DISTINGUISH_PROBES]
#: What the education quality question calls the event, by kind (BUG-1088). Only the
#: kinds in `QUALITY_DISTINGUISH_EVENT_KINDS` reach the quality probe.
EDUCATION_QUALITY_EVENT_NAME = {
"education_start": "升学",
"education_change": "学业变动",
"education_interruption": "学业中断",
}
def _quality_user_meaning(event: dict[str, Any], domain: str) -> str:
label = _display_date_label(event)
if domain == "education":
name = EDUCATION_QUALITY_EVENT_NAME.get(_event_kind_name(event), "学业经历")
return (
f"{label}那次{name}，结果更接近如愿、将就调剂、发挥失常还是说不清。"
"只问那次经历的实际体验，不得改时间范围。"
)
family = str(DOMAIN_CATALOG[domain]["event_family"])
return (
f"{label}那次{family}，当时更接近顺利、将就、明显受挫还是说不清。"
"只问那次经历的实际体验，不得改时间范围。"
)
def _quality_distinguish_probes(
events: Sequence[dict[str, Any]],
clusters: Sequence[dict[str, Any]],
*,
set_version: str,
holdout_ids: set[str],
holdout_keys: set[str],
) -> list[dict[str, Any]]:
if len(clusters) < 2:
return []
rows: list[dict[str, Any]] = []
for event in events:
if not isinstance(event, dict):
continue
event_id = str(event.get("id") or "")
domain = str(event.get("domain") or "")
year = _event_year(event)
layer = DOMAIN_QUALITY_LAYER.get(domain)
if not event_id or year is None or layer is None or domain not in DOMAIN_CATALOG:
continue
if not _quality_event_allowed(event, domain):
continue
if event_id in holdout_ids or f"{domain}:{year}" in holdout_keys:
continue
if _quality_encoded(event, domain):
continue
groups: dict[int, list[str]] = {}
for cluster in clusters:
representative = cluster.get("representative") if isinstance(cluster, dict) else None
if not isinstance(representative, dict):
continue
sign = _layer_value(representative, layer)
if not isinstance(sign, int):
continue
bucket = groups.setdefault(sign, [])
for time in cluster.get("times") or []:
clock = str(time)[:5]
if len(clock) == 5 and clock not in bucket:
bucket.append(clock)
if len(groups) < 2:
continue
signs = sorted(groups)
yes_times = sorted(groups[signs[-1]], key=_clock)
no_times = sorted(groups[signs[0]], key=_clock)
overlap = set(yes_times) & set(no_times)
yes_times = [time for time in yes_times if time not in overlap]
no_times = [time for time in no_times if time not in overlap]
if len(yes_times) < 1 or len(no_times) < 1:
continue
month = _event_month(event)
outcomes = [
{"answer_class": "yes", "supports": yes_times, "conflicts": no_times},
{"answer_class": "weak_yes", "supports": yes_times, "conflicts": no_times},
{"answer_class": "no", "supports": no_times, "conflicts": yes_times},
{"answer_class": "unsure", "supports": [], "conflicts": []},
]
split = candidate_split_hash(
candidate_set_version_value=set_version,
domain=domain,
year=year,
month=month,
groups=[yes_times, no_times],
)
gain = round(_group_entropy([len(yes_times), len(no_times)]), 4)
if gain <= 0:
continue
label = _display_date_label(event)
probe = _public_probe(
year=year,
month=month,
domain=domain,
source="known_event_quality",
tracks=("vimshottari", "narayana"),
tracks_agree=True,
user_meaning=_quality_user_meaning(event, domain),
event_family=str(DOMAIN_CATALOG[domain]["quality_family"]),
information_gain=gain,
semantic_key=f"{domain}.{year}.known_event_quality",
candidate_split_hash=split,
candidate_set_version=set_version,
expected_outcomes=outcomes,
candidate_ids=candidate_ids_from_outcomes(outcomes),
left_time=yes_times[0],
right_time=no_times[0],
target_evidence_id=event_id,
display_date_label=label,
role="distinguish",
phase=PROBE_PHASE_CANDIDATE_DISCRIMINATOR,
style_options=list(QUALITY_DISTINGUISH_OPTIONS.get(domain) or ()),
)
if distinguish_contract_errors(probe):
continue
rows.append(_apply_prior_ranking(probe))
return _select_quality_distinguish_rows(rows)
def _year_activated(rule_ids: Sequence[str]) -> bool:
return _has_domain_activation(rule_ids) or LEVEL_RANK.get(match_level(rule_ids), 0) >= 2
def _public_probe(
*,
year: int,
domain: str,
source: str,
tracks: Sequence[str],
tracks_agree: bool,
user_meaning: str,
event_family: str,
month: int | None = None,
**extra: Any,
) -> dict[str, Any]:
if source == "known_event_quality":
phase = PROBE_PHASE_EVENT_CLARIFICATION
role = "clarify"
elif source == "age_band":
phase = PROBE_PHASE_EVIDENCE_COLLECTION
role = "collect"
else:
phase = PROBE_PHASE_CANDIDATE_DISCRIMINATOR
role = "distinguish"
month_value = month if isinstance(month, int) and 1 <= month <= 12 else None
payload = {
"year": year,
"year_label": _period_label(year, month_value),
"domain": domain,
"event_family": event_family,
"source": source,
"tracks": list(tracks),
"tracks_agree": tracks_agree,
"unique_minute_claim": False,
"user_meaning": user_meaning,
"role": role,
"phase": phase,
"semantic_key": f"{domain}.{year}.{month_value:02d}" if month_value else f"{domain}.{year}",
"information_gain": 0.0,
"candidate_split_hash": f"{domain}:{year}" + (f"-{month_value:02d}" if month_value else ""),
"expected_outcomes": [],
"candidate_ids": [],
"choice_kind": "event_quality" if source == "known_event_quality" else "existence",
}
if month_value:
payload["month"] = month_value
payload.update(extra)
if payload["role"] == "distinguish":
payload["candidate_ids"] = candidate_ids_from_outcomes(payload.get("expected_outcomes") or [])
style_options = completed_style_options(payload.get("choice_kind"), payload.get("style_options"))
if style_options:
payload["style_options"] = style_options
payload["question_contract_version"] = QUESTION_CONTRACT_VERSION
return payload
QUALITY_HINTS: dict[str, tuple[str, ...]] = {
"education": ("失利", "失常", "压力", "复读", "考砸", "发挥不好", "发挥异常"),
}
# Only education has a quality dimension distinct from existence.
# Career/family "quality_family" copies event_family and must not jump the
# adoption gate as known_event_quality.
def _quality_encoded(event: dict[str, Any], domain: str) -> bool:
summary = str(event.get("summary") or "")
return any(hint in summary for hint in QUALITY_HINTS.get(domain, ()))
def _year_quality_encoded(events: Sequence[dict[str, Any]], domain: str, year: int) -> bool:
return any(
isinstance(event, dict)
and canonical_domain(event.get("domain")) == canonical_domain(domain)
and _event_year(event) == year
and _quality_encoded(event, canonical_domain(domain))
for event in events
)
def _event_years(events: Sequence[dict[str, Any]], domain: str) -> set[int]:
years: set[int] = set()
wanted = canonical_domain(domain)
for event in events:
if not isinstance(event, dict) or canonical_domain(event.get("domain")) != wanted:
continue
year = _event_year(event)
if year is not None:
years.add(year)
return years
def _existence_blocked_years(domain: str, known_years: set[int]) -> set[int]:
blocked = set(known_years)
span = int(EXISTENCE_NEARBY_YEARS.get(domain, 0))
if span <= 0:
return blocked
for year in known_years:
blocked.update(range(year - span, year + span + 1))
return blocked
def _quality_probes(
events: Sequence[dict[str, Any]],
domains: Sequence[str],
) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
seen_domains: set[str] = set()
for event in events:
if not isinstance(event, dict):
continue
domain = str(event.get("domain") or "")
year = _event_year(event)
if domain not in domains or year is None or domain in seen_domains:
continue
if domain not in DOMAIN_CATALOG or domain not in QUALITY_HINTS:
continue
if _year_quality_encoded(events, domain, year):
continue
seen_domains.add(domain)
summary = str(event.get("summary") or "")
family = str(DOMAIN_CATALOG[domain]["quality_family"])
exam = domain == "education" and ("高考" in summary or "考试" in summary)
rows.append(_public_probe(
year=year,
domain=domain,
source="known_event_quality",
tracks=("vimshottari", "narayana"),
tracks_agree=True,
user_meaning=_agent_brief(
year_label=_year_label(year),
domain=domain,
family=family,
quality=True,
exam=exam,
),
event_family=family,
))
return rows
def _evaluate_contexts(
contexts: Sequence[dict[str, Any]],
*,
birth_date: str,
domain: str,
year: int,
source: str,
month: int | None = None,
clusters: Sequence[dict[str, Any]] | None = None,
set_version: str | None = None,
) -> dict[str, Any] | None:
scored_rows: list[tuple[str, list[str]]] = []
month_value = month if isinstance(month, int) and 1 <= month <= 12 else None
for context in contexts:
time = _context_time(context)
if not time:
continue
scored = _score_year(
context,
birth_date=birth_date,
domain=domain,
year=year,
month=month_value,
)
if scored is None:
continue
scored_rows.append((time, list(scored.get("rule_ids") or [])))
if len(scored_rows) < 2:
return None
yes: list[tuple[str, list[str]]] = []
no: list[tuple[str, list[str]]] = []
for time, rules in scored_rows:
if _year_activated(rules):
yes.append((time, rules))
else:
no.append((time, rules))
if not yes or not no:
ranks = [
(time, LEVEL_RANK.get(match_level(rules), 0), rules)
for time, rules in scored_rows
]
highest = max(item[1] for item in ranks)
lowest = min(item[1] for item in ranks)
if highest - lowest < 2:
return None
yes = [(time, rules) for time, rank, rules in ranks if rank == highest]
no = [(time, rules) for time, rank, rules in ranks if rank < highest]
if not yes or not no:
return None
if not any(_discriminates(left, right) for _, left in yes for _, right in no):
return None
yes_times = sorted((time for time, _ in yes), key=_clock)
no_times = sorted((time for time, _ in no), key=_clock)
if clusters:
yes_times = expand_times_through_clusters(yes_times, clusters)
no_times = expand_times_through_clusters(no_times, clusters)
overlap = set(yes_times) & set(no_times)
if overlap:
yes_times = [time for time in yes_times if time not in overlap]
no_times = [time for time in no_times if time not in overlap]
if len(yes_times) < 1 or len(no_times) < 1:
return None
yes_level = max((match_level(rules) for _, rules in yes), key=lambda item: LEVEL_RANK[item])
no_level = min((match_level(rules) for _, rules in no), key=lambda item: LEVEL_RANK[item])
stronger = next(rules for _, rules in yes if match_level(rules) == yes_level)
vim_hit, narayana_hit = _tracks_present(stronger)
gain = round(
_group_entropy([len(yes_times), len(no_times)]) + _information_gain(yes_level, no_level),
4,
)
if gain <= 0:
return None
version = set_version or candidate_set_version([yes_times, no_times])
outcomes = [
{"answer_class": "yes", "supports": yes_times, "conflicts": no_times},
{"answer_class": "weak_yes", "supports": yes_times, "conflicts": no_times},
{"answer_class": "no", "supports": no_times, "conflicts": yes_times},
{"answer_class": "unsure", "supports": [], "conflicts": []},
]
split = candidate_split_hash(
candidate_set_version_value=version,
domain=domain,
year=year,
month=month_value,
groups=[yes_times, no_times],
)
period = _period_label(year, month_value)
probe = _public_probe(
year=year,
month=month_value,
domain=domain,
source=source,
tracks=("vimshottari", "narayana"),
tracks_agree=vim_hit and narayana_hit,
user_meaning=_agent_brief(
year_label=period,
domain=domain,
family=str(DOMAIN_CATALOG[domain]["event_family"]),
),
event_family=str(DOMAIN_CATALOG[domain]["event_family"]),
information_gain=gain,
semantic_key=f"{domain}.{year}.{month_value:02d}.{source}" if month_value else f"{domain}.{year}.{source}",
candidate_split_hash=split,
candidate_set_version=version,
expected_outcomes=outcomes,
candidate_ids=candidate_ids_from_outcomes(outcomes),
left_time=yes_times[0],
right_time=no_times[0],
source_features=[{
"technique": source,
"layers": list(SIGNATURE_LAYERS),
"calculationResultId": None,
}],
)
if not probe or distinguish_contract_errors(probe):
return None
return probe
def _evaluate_year(
left: dict[str, Any],
right: dict[str, Any],
*,
birth_date: str,
domain: str,
year: int,
source: str,
) -> dict[str, Any] | None:
return _evaluate_contexts(
[left, right],
birth_date=birth_date,
domain=domain,
year=year,
source=source,
)
def event_clarification_probes(
request: dict[str, Any],
*,
domains: Sequence[str] | None = None,
) -> list[dict[str, Any]]:
events = [item for item in (request.get("events") or []) if isinstance(item, dict)]
known = [
str(event.get("domain"))
for event in events
if str(event.get("domain") or "") in DOMAIN_CATALOG
]
return _quality_probes(events, domains or known)
def evidence_collection_probes(
request: dict[str, Any],
*,
today: date | None = None,
) -> list[dict[str, Any]]:
birth_year = _birth_year(request.get("birth_date"))
if birth_year is None:
return []
now = today or date.today()
events = [item for item in (request.get("events") or []) if isinstance(item, dict)]
missing = missing_collection_domains(events, list(DOMAIN_CATALOG), frozenset())
rows: list[dict[str, Any]] = []
for domain in missing:
year = _age_band_year(birth_year, domain, now)
if year is None:
continue
rows.append(_public_probe(
year=year,
domain=domain,
source="age_band",
tracks=("vimshottari", "narayana"),
tracks_agree=False,
user_meaning=_agent_brief(
year_label=_year_label(year),
domain=domain,
family=str(DOMAIN_CATALOG[domain]["event_family"]),
),
event_family=str(DOMAIN_CATALOG[domain]["event_family"]),
))
if len(rows) >= MAX_COLLECTION_PROBES:
break
return rows
def candidate_contrast_opportunities(
request: dict[str, Any],
built: dict[str, Any],
*,
scan: dict[str, Any],
candidate_times: Sequence[str],
representative_time: str | None,
today: date | None = None,
) -> list[dict[str, Any]]:
probes = discriminating_event_probes(
request,
built,
scan=scan,
candidate_times=candidate_times,
representative_time=representative_time,
today=today,
)
return [opportunity_from_probe(probe) for probe in probes]
def _discriminating_event_probe_lists(
request: dict[str, Any],
built: dict[str, Any],
*,
scan: dict[str, Any],
candidate_times: Sequence[str],
representative_time: str | None,
precision_current: str | None = None,
today: date | None = None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
del precision_current, representative_time
birth_date = str(request.get("birth_date") or "").strip()
birth_year = _birth_year(birth_date)
empty: tuple[list[dict[str, Any]], list[dict[str, Any]]] = ([], [])
if birth_year is None:
return empty
try:
datetime.strptime(birth_date, "%Y-%m-%d")
except ValueError:
return empty
events = [item for item in (request.get("events") or []) if isinstance(item, dict)]
asked_probe_keys = [
str(item).strip()
for item in (request.get("asked_probe_keys") or [])
if isinstance(item, str) and str(item).strip()
]
if not discriminator_gate_open(events):
return empty
holdout_keys = holdout_domain_years(events)
now = today or date.today()
full = _static_contexts(built)
if len(full) < 2:
return empty
refresh = request.get("refresh_probes") is True
remaining = _remaining_contexts(built, candidate_times) if candidate_times else []
if refresh and remaining:
work = remaining
clusters = cluster_contexts_by_signature(work)
else:
work = full
clusters = cluster_contexts_by_signature(full)
if len(clusters) < 2:
work = remaining or full
clusters = cluster_contexts_by_signature(work)
if len(clusters) < 2:
return empty
reps = [cluster["representative"] for cluster in clusters if _scoreable(cluster["representative"])]
if len(reps) < 2:
reps = [item for item in work if _scoreable(item)]
if len(reps) < 2:
return empty
max_probes, max_per_domain = _probe_caps(
refresh=refresh,
remaining_count=len(remaining) if refresh else len(full),
)
set_version = candidate_set_version([cluster["times"] for cluster in clusters])
remaining_layers = _differing_layers(work if refresh else full)
if not remaining_layers:
remaining_layers = {
layer for layer in SCORING_LAYERS
if scan.get(f"{layer}_candidates_differ")
}
domains = _probe_domains(
remaining_layers,
events,
d1_differs="d1" in remaining_layers or bool(scan.get("d1_candidates_differ")),
)
lo, hi = birth_year + 5, min(now.year, birth_year + 80)
union_kw: dict[str, Any] = {}
if refresh:
union_kw = {
"min_boundary_days": REFRESH_MIN_BOUNDARY_DAYS,
"include_pratyantar": True,
"varga_narayana": True,
}
boundary_dates = _union_boundary_dates(
reps,
birth_date=birth_date,
lo=lo,
hi=hi,
**union_kw,
) if domains else []
probes: list[dict[str, Any]] = []
for domain in domains:
if domain not in DOMAIN_CATALOG:
continue
known_years = _event_years(events, domain) | asked_years_for_domain(asked_probe_keys, domain)
blocked_years = _existence_blocked_years(domain, known_years)
domain_lo = _domain_year_floor(birth_year, domain, lo)
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,
year=at.year,
month=at.month,
source="dasha_boundary",
clusters=clusters,
set_version=set_version,
)
evaluated += 1
if row is None or distinguish_contract_errors(row):
continue
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_per_domain]
if len(kept) < max_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"]),
)
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.extend(_quality_distinguish_probes(
events,
clusters,
set_version=set_version,
holdout_ids=set(holdout_event_ids(events)),
holdout_keys=set(holdout_keys),
))
_annotate_nearby_ledger(probes, events)
probes.sort(key=_probe_sort_key)
public, dropped = _partition_ranked_probes(probes, max_probes=max_probes)
assert_distinguish_contract(public)
return public, dropped
def _partition_ranked_probes(
probes: Sequence[dict[str, Any]],
*,
max_probes: int = MAX_PROBES,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
public: list[dict[str, Any]] = []
dropped: list[dict[str, Any]] = []
seen: set[tuple[str, int, int, str]] = set()
for row in probes:
anchored_quality = (
row.get("source") == "known_event_quality"
and row.get("role") == "distinguish"
and row.get("target_evidence_id")
)
if row.get("source") == "known_event_quality" and not anchored_quality:
continue
if row.get("phase") != PROBE_PHASE_CANDIDATE_DISCRIMINATOR:
continue
if distinguish_contract_errors(row):
continue
ranked = row if "raw_split_gain" in row else _apply_prior_ranking(dict(row))
priors = ranked.get("answer_priors") or _answer_priors_for(ranked)
if _dominant_existence_prior(ranked, priors):
dropped.append({
"semantic_key": ranked.get("semantic_key"),
"reason": "dominant_answer_prior",
"domain": ranked.get("domain"),
"year": ranked.get("year"),
"source": ranked.get("source"),
"answer_priors": priors,
})
continue
if not isinstance(row.get("year"), int) or int(row["year"]) <= 0:
continue
key = (str(ranked["domain"]), int(ranked["year"]), int(ranked.get("month") or 0), str(ranked["source"]))
encoded = str(ranked)
if key in seen or "points" in encoded:
continue
seen.add(key)
public.append(ranked)
if len(public) >= max_probes:
break
public.sort(key=_probe_sort_key)
return public, dropped
def discriminating_event_probe_set(
request: dict[str, Any],
built: dict[str, Any],
*,
scan: dict[str, Any],
candidate_times: Sequence[str],
representative_time: str | None,
precision_current: str | None = None,
today: date | None = None,
) -> dict[str, list[dict[str, Any]]]:
probes, dropped = _discriminating_event_probe_lists(
request,
built,
scan=scan,
candidate_times=candidate_times,
representative_time=representative_time,
precision_current=precision_current,
today=today,
)
return {"probes": probes, "dropped": dropped}
def discriminating_event_probes(
request: dict[str, Any],
built: dict[str, Any],
*,
scan: dict[str, Any],
candidate_times: Sequence[str],
representative_time: str | None,
precision_current: str | None = None,
today: date | None = None,
) -> list[dict[str, Any]]:
probes, _dropped = _discriminating_event_probe_lists(
request,
built,
scan=scan,
candidate_times=candidate_times,
representative_time=representative_time,
precision_current=precision_current,
today=today,
)
return probes
GUIDED_COLLECT_LIMIT = 6
GUIDED_DOMAIN_ORDER = (
"education",
"career",
"relocation",
"relationship",
"family",
"finance",
"health_pressure",
)
_GUIDED_TRACK_DOMAIN = {
"nara:d9": "relationship",
"nara:d10": "career",
}
#: Windows whose track carries no domain are asked open (BUG-749).
GUIDED_ANY_DOMAIN = "any"
def _guided_track_starts(
context: dict[str, Any],
*,
birth_date: str,
lo: int,
hi: int,
include_pratyantar: bool,
varga_narayana: bool,
vim_cache: dict[tuple[Any, ...], list[date]],
narayana_cache: dict[tuple[Any, ...], list[date] | None],
) -> dict[str, list[date]]:
tracks: dict[str, list[date]] = {}
moon = float(context["planet_longitudes"]["Moon"])
vim_key = (*_vim_cache_key(birth_date, moon, lo, hi), include_pratyantar)
if vim_key not in vim_cache:
vim_cache[vim_key] = _vim_start_dates(
birth_date, moon, lo, hi, include_pratyantar=include_pratyantar,
)
tracks["vim"] = vim_cache[vim_key]
natal_key = (*_narayana_cache_key(
int(context["ascendant_index"]), context["planet_longitudes"], birth_date, lo, hi,
), None)
if natal_key not in narayana_cache:
narayana_cache[natal_key] = _narayana_start_dates(
int(context["ascendant_index"]), context["planet_longitudes"], birth_date, lo, hi,
)
natal = narayana_cache[natal_key]
if natal is not None:
tracks["nara"] = natal
if not varga_narayana:
return tracks
for layer in ("d9", "d10"):
raw = _layer_value(context, layer)
if not isinstance(raw, int):
continue
layer_key = (*_narayana_cache_key(
raw, context["planet_longitudes"], birth_date, lo, hi,
), layer)
if layer_key not in narayana_cache:
narayana_cache[layer_key] = _narayana_start_dates(
raw, context["planet_longitudes"], birth_date, lo, hi,
)
dates = narayana_cache[layer_key]
if dates is not None:
tracks[f"nara:{layer}"] = dates
return tracks
def _guided_year_windows(dates: Sequence[date]) -> list[tuple[int, int, int]]:
by_year: dict[int, list[date]] = {}
for item in dates:
by_year.setdefault(item.year, []).append(item)
windows: list[tuple[int, int, int]] = []
for year in sorted(by_year):
months = [item.month for item in by_year[year]]
windows.append((year, min(months), max(months)))
return windows
def guided_collect_windows(
request: dict[str, Any],
built: dict[str, Any],
*,
candidate_times: Sequence[str],
today: date | None = None,
) -> list[dict[str, Any]]:
"""Boundary windows for guided collect. Does not change auto probe generation."""
birth_date = str(request.get("birth_date") or "").strip()
birth_year = _birth_year(birth_date)
if birth_year is None:
return []
try:
datetime.strptime(birth_date, "%Y-%m-%d")
except ValueError:
return []
remaining = _remaining_contexts(built, candidate_times)
if len(remaining) < 2:
remaining = _static_contexts(built)
work = [item for item in remaining if _scoreable(item)]
if len(work) < 2:
return []
declined = {
str(item).strip()
for item in (request.get("declined_domains") or [])
if str(item).strip() in GUIDED_DOMAIN_ORDER
}
now = today or date.today()
lo, hi = birth_year + 5, min(now.year, birth_year + 80)
vim_cache: dict[tuple[Any, ...], list[date]] = {}
narayana_cache: dict[tuple[Any, ...], list[date] | None] = {}
starts_by_time: dict[str, dict[str, list[date]]] = {}
for context in work:
time = _context_time(context)
if not time:
continue
starts_by_time[time] = _guided_track_starts(
context,
birth_date=birth_date,
lo=lo,
hi=hi,
include_pratyantar=True,
varga_narayana=True,
vim_cache=vim_cache,
narayana_cache=narayana_cache,
)
# BUG-1323: gathered depends only on (track, boundary index). Prefetch once
# per candidate and skip a boundary after the one==two check has passed.
work_tracks = [starts_by_time.get(_context_time(context) or "") or {} for context in work]
rows: list[dict[str, Any]] = []
seen: set[tuple[int, int, int, str]] = set()
done: set[tuple[str, int]] = set()
for left, right in _representative_pairs(work):
left_time = str(_context_time(left) or "")
right_time = str(_context_time(right) or "")
left_tracks = starts_by_time.get(left_time) or {}
right_tracks = starts_by_time.get(right_time) or {}
for track in ("vim", "nara", "nara:d9", "nara:d10"):
left_dates = left_tracks.get(track) or []
right_dates = right_tracks.get(track) or []
if not left_dates or not right_dates:
continue
for index, (one, two) in enumerate(zip(left_dates, right_dates)):
if one == two:
continue
if (track, index) in done:
continue
done.add((track, index))
gathered: list[date] = []
for tracks in work_tracks:
dates = tracks.get(track) or []
if index < len(dates):
gathered.append(dates[index])
if len(gathered) < 2 or min(gathered) == max(gathered):
continue
ordered = sorted(gathered)
cut = ordered[0] + (ordered[-1] - ordered[0]) / 2
left_n = sum(1 for item in gathered if item <= cut)
right_n = len(gathered) - left_n
if left_n < 1 or right_n < 1:
continue
# BUG-749: only the D9 / D10 Narayana tracks carry a domain.
# Vimshottari and natal Narayana boundaries say nothing about
# which area of life moved, so the window is asked open
# ("any") instead of being round-robined onto one domain.
track_domain = _GUIDED_TRACK_DOMAIN.get(track)
domain = (
track_domain
if track_domain and track_domain not in declined
else GUIDED_ANY_DOMAIN
)
for year, month_lo, month_hi in _guided_year_windows(gathered):
if year < lo or year > hi:
continue
key = (year, month_lo, month_hi, domain)
if key in seen:
continue
seen.add(key)
rows.append({
"year": year,
"month_lo": month_lo,
"month_hi": month_hi,
"domain": domain,
"split": {"left": left_n, "right": right_n},
})
rows.sort(key=lambda row: (
abs(row["split"]["left"] - row["split"]["right"]),
abs(now.year - int(row["year"])),
-int(row["year"]),
int(row["month_lo"]),
str(row["domain"]),
))
return rows[:GUIDED_COLLECT_LIMIT]
def prospective_windows_for_time(
request: dict[str, Any],
built: dict[str, Any],
time: str,
*,
today: date | None = None,
) -> list[dict[str, Any]]:
"""1–2 event windows in the next 12 months for one candidate minute."""
birth_date = str(request.get("birth_date") or "").strip()
clock = str(time or "")[:5]
if not birth_date or len(clock) < 5:
return []
now = today or date.today()
horizon = now + timedelta(days=366)
by_time = {_context_time(item): item for item in _static_contexts(built)}
context = by_time.get(clock)
if not isinstance(context, dict):
return []
moon = (context.get("planet_longitudes") or {}).get("Moon")
if not isinstance(moon, (int, float)):
return []
try:
starts = [
item
for item in _vim_start_dates(
birth_date,
float(moon),
now.year,
horizon.year,
include_pratyantar=True,
)
if now <= item <= horizon
]
except (KeyError, TypeError, ValueError):
return []
rows: list[dict[str, Any]] = []
seen: set[str] = set()
for start in sorted(set(starts)):
key = f"{start.year:04d}-{start.month:02d}"
if key in seen:
continue
seen.add(key)
rows.append({
"domain": "career",
"from": key,
"to": key,
})
if len(rows) >= 2:
break
return rows
def prospective_windows_by_time(
request: dict[str, Any],
built: dict[str, Any],
times: Sequence[str],
*,
today: date | None = None,
) -> dict[str, list[dict[str, Any]]]:
out: dict[str, list[dict[str, Any]]] = {}
for raw in times:
clock = str(raw or "")[:5]
if len(clock) < 5 or clock in out:
continue
out[clock] = prospective_windows_for_time(request, built, clock, today=today)
return out
def prospective_event_windows(
request: dict[str, Any],
built: dict[str, Any],
*,
candidate_times: Sequence[str],
today: date | None = None,
) -> list[dict[str, Any]]:
birth_date = str(request.get("birth_date") or "").strip()
if not birth_date:
return []
now = today or date.today()
lo, hi = now.year + 1, now.year + 3
full = _static_contexts(built)
if len(full) < 2:
return []
clusters = cluster_contexts_by_signature(full)
if len(clusters) < 2:
remaining = _remaining_contexts(built, candidate_times) or full
clusters = cluster_contexts_by_signature(remaining)
if len(clusters) < 2:
return []
labels = ("A", "B", "C")
rows: list[dict[str, Any]] = []
for index, cluster in enumerate(clusters[:3]):
context = cluster.get("representative") if isinstance(cluster, dict) else None
if not isinstance(context, dict):
continue
moon = (context.get("planet_longitudes") or {}).get("Moon")
if not isinstance(moon, (int, float)):
continue
starts = _vim_start_dates(birth_date, float(moon), lo, hi, include_pratyantar=True)
future = [item for item in starts if item.year >= lo]
if not future:
continue
start = min(future)
label = labels[index]
window = f"{start.year} 年 {start.month} 月附近"
rows.append({
"candidate_label": label,
"domain": "career",
"window_label": window,
"user_meaning": (
f"候选 {label} 预测下一次事业变动更可能在 {window}。"
"这是预测窗口，不是承诺；下次发生时回来补一条，可进一步分辨。"
),
"used_for_scoring": False,
})
return rows
Reference in New Issue View Git Blame Copy Permalink