fix(web): rank remaining-minute probes by split and match choice kind

Choice cards used a hardcoded domain menu and always asked existence.
Rank scoring layers by remaining-minute entropy, keep finance and health
volunteer-only, and ask D9/D10 style or exam quality so taps match outcomes.

Co-authored-by: Cursor <cursoragent@cursor.com>

scripts/rectification/event_probes.py

@@ -37,14 +37,19 @@ LAYER_DOMAIN = {
     "d11": "finance",
     "d30": "health_pressure",
 }
-STAGE_DOMAIN = {
-    "d9_refine": "relationship",
-    "d10_refine": "career",
-    "d4_refine": "relocation",
-    "theme_refine": "relocation",
-    "d5_refine": "education",
-}
 VOLUNTEER_ONLY = frozenset({"finance", "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": "升学、高考、转学或学习环境变化",
@@ -158,10 +163,43 @@ def _age_band_year(birth_year: int, domain: str, today: date) -> int | 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_domains(
-    scan: dict[str, Any],
-    precision_current: str | None,
+    remaining_layers: set[str],
     events: Sequence[dict[str, Any]],
+    *,
+    d1_differs: bool = False,
 ) -> list[str]:
     volunteered = {
         str(event.get("domain"))
@@ -169,19 +207,16 @@ def _probe_domains(
         if isinstance(event, dict) and event.get("domain")
     }
     ordered: list[str] = []
-    stage_domain = STAGE_DOMAIN.get(str(precision_current or ""))
-    if stage_domain:
-        ordered.append(stage_domain)
     for layer in SCORING_LAYERS:
         domain = LAYER_DOMAIN.get(layer)
         if not domain or domain in ordered:
             continue
-        if not scan.get(f"{layer}_candidates_differ"):
+        if layer not in remaining_layers:
             continue
         if domain in VOLUNTEER_ONLY and domain not in volunteered:
             continue
         ordered.append(domain)
-    if not ordered and scan.get("d1_candidates_differ"):
+    if not ordered and d1_differs:
         ordered.append("education")
     return ordered
 
@@ -195,6 +230,20 @@ def _static_contexts(built: dict[str, Any]) -> list[dict[str, Any]]:
     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 _pick_representatives(
     built: dict[str, Any],
     scan: dict[str, Any],
@@ -440,6 +489,13 @@ def _pair_entropy(left: float, right: float) -> float:
     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)
@@ -448,6 +504,10 @@ def _information_gain(left_level: str, right_level: str) -> float:
     return round(max(0.0, 1.0 - after), 4)
 
 
+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,
@@ -474,6 +534,7 @@ def _public_probe(
         "information_gain": 0.0,
         "candidate_split_hash": f"{domain}:{year}",
         "expected_outcomes": [],
+        "choice_kind": "event_quality" if source == "known_event_quality" else "existence",
     }
     payload.update(extra)
     return payload
@@ -558,33 +619,54 @@ def _quality_probes(
     return rows
 
 
-def _evaluate_year(
-    left: dict[str, Any],
-    right: dict[str, Any],
+def _evaluate_contexts(
+    contexts: Sequence[dict[str, Any]],
     *,
     birth_date: str,
     domain: str,
     year: int,
     source: str,
 ) -> dict[str, Any] | None:
-    scored_left = _score_year(left, birth_date=birth_date, domain=domain, year=year)
-    scored_right = _score_year(right, birth_date=birth_date, domain=domain, year=year)
-    if scored_left is None or scored_right is None:
+    scored_rows: list[tuple[str, list[str]]] = []
+    for context in contexts:
+        time = _context_time(context)
+        if not time:
+            continue
+        scored = _score_year(context, birth_date=birth_date, domain=domain, year=year)
+        if scored is None:
+            continue
+        scored_rows.append((time, list(scored.get("rule_ids") or [])))
+    if len(scored_rows) < 2:
         return None
-    left_rules = scored_left.get("rule_ids") or []
-    right_rules = scored_right.get("rule_ids") or []
-    if not _discriminates(left_rules, right_rules):
+    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
-    left_level = match_level(left_rules)
-    right_level = match_level(right_rules)
-    stronger = left_rules if LEVEL_RANK[left_level] >= LEVEL_RANK[right_level] else right_rules
+    yes_times = sorted((time for time, _ in yes), key=_clock)
+    no_times = sorted((time for time, _ in no), key=_clock)
+    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)
-    left_time = _context_time(left)
-    right_time = _context_time(right)
-    left_stronger = LEVEL_RANK[left_level] >= LEVEL_RANK[right_level]
-    yes_supports = [time for time in ([left_time] if left_stronger else [right_time]) if time]
-    yes_conflicts = [time for time in ([right_time] if left_stronger else [left_time]) if time]
-    split = f"{domain}:{year}:{ '|'.join(sorted(yes_supports + yes_conflicts)) }"
+    split = f"{domain}:{year}:{ '|'.join(sorted(yes_times + no_times)) }"
     return _public_probe(
         year=year,
         domain=domain,
@@ -596,16 +678,37 @@ def _evaluate_year(
             family=str(DOMAIN_CATALOG[domain]["event_family"]),
         ),
         event_family=str(DOMAIN_CATALOG[domain]["event_family"]),
-        information_gain=_information_gain(left_level, right_level),
+        information_gain=round(
+            _group_entropy([len(yes_times), len(no_times)]) + _information_gain(yes_level, no_level),
+            4,
+        ),
         semantic_key=f"{domain}.{year}.{source}",
         candidate_split_hash=split,
         expected_outcomes=[
-            {"answer_class": "yes", "supports": yes_supports, "conflicts": yes_conflicts},
-            {"answer_class": "no", "supports": yes_conflicts, "conflicts": yes_supports},
+            {"answer_class": "yes", "supports": yes_times, "conflicts": no_times},
+            {"answer_class": "no", "supports": no_times, "conflicts": yes_times},
             {"answer_class": "unsure", "supports": [], "conflicts": []},
         ],
-        left_time=left_time,
-        right_time=right_time,
+        left_time=yes_times[0],
+        right_time=no_times[0],
+    )
+
+
+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,
     )
 
 
@@ -619,6 +722,7 @@ def discriminating_event_probes(
     precision_current: str | None = None,
     today: date | None = None,
 ) -> list[dict[str, Any]]:
+    del precision_current
     birth_date = str(request.get("birth_date") or "").strip()
     birth_year = _birth_year(birth_date)
     if birth_year is None:
@@ -629,21 +733,42 @@ def discriminating_event_probes(
         return []
     now = today or date.today()
     events = [item for item in (request.get("events") or []) if isinstance(item, dict)]
-    domains = _probe_domains(scan, precision_current, events)
-    if not domains:
+    remaining = _remaining_contexts(built, candidate_times)
+    if len(remaining) < 2:
+        remaining = _static_contexts(built)
+    remaining_layers = _differing_layers(remaining) if len(remaining) >= 2 else set()
+    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")),
+    )
+    known_domains = [
+        str(event.get("domain"))
+        for event in events
+        if str(event.get("domain") or "") in DOMAIN_CATALOG
+    ]
+    if not domains and not known_domains:
         return []
     probes: list[dict[str, Any]] = []
     covered_domains: set[str] = set()
     pair = _pick_representatives(built, scan, candidate_times, representative_time)
+    scoreable_remaining = [item for item in remaining if _scoreable(item)]
+    if len(scoreable_remaining) >= 2:
+        score_contexts = scoreable_remaining
+    elif pair is not None and _scoreable(pair[0]) and _scoreable(pair[1]):
+        score_contexts = [pair[0], pair[1]]
+    else:
+        score_contexts = []
     lo, hi = birth_year + 5, min(now.year, birth_year + 80)
-    can_score = (
-        pair is not None
-        and _scoreable(pair[0])
-        and _scoreable(pair[1])
-    )
+    can_score = len(score_contexts) >= 2
     dasha_domains: set[str] = set()
-    if can_score and pair is not None:
-        left, right = pair
+    if can_score:
+        left, right = score_contexts[0], score_contexts[-1]
         left_moon = float(left["planet_longitudes"]["Moon"])
         right_moon = float(right["planet_longitudes"]["Moon"])
         vim_years = _boundary_years(
@@ -661,26 +786,36 @@ def discriminating_event_probes(
             known_years = _event_years(events, domain)
             blocked_years = _existence_blocked_years(domain, known_years)
             boundary = sorted((vim_years | narayana_years) & set(range(lo, hi + 1)))
-            found = None
+            best = None
             for year in boundary:
                 if year in blocked_years:
                     continue
-                found = _evaluate_year(
-                    left, right, birth_date=birth_date, domain=domain, year=year, source="dasha_boundary",
+                found = _evaluate_contexts(
+                    score_contexts,
+                    birth_date=birth_date,
+                    domain=domain,
+                    year=year,
+                    source="dasha_boundary",
                 )
-                if found:
-                    break
-            if found is None:
+                if found is None:
+                    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:
-                    found = _evaluate_year(
-                        left, right, birth_date=birth_date, domain=domain, year=midpoint, source="dasha_activation",
+                    best = _evaluate_contexts(
+                        score_contexts,
+                        birth_date=birth_date,
+                        domain=domain,
+                        year=midpoint,
+                        source="dasha_activation",
                     )
-            if found:
-                probes.append(found)
+            if best:
+                probes.append(best)
                 dasha_domains.add(domain)
                 covered_domains.add(domain)
-    quality = _quality_probes(events, domains)
+    quality = _quality_probes(events, known_domains or domains)
     for row in quality:
         if row["domain"] in dasha_domains:
             continue

scripts/rectification/refinement_packet.py

@@ -77,11 +77,18 @@ def _features(built: dict[str, Any]) -> list[dict[str, Any]]:
     return rows
 
 
+def _sign_name(index: int | None) -> str | None:
+    if isinstance(index, int) and 0 <= index <= 11:
+        return SIGNS_CN[SIGNS[index]]
+    return None
+
+
 def _sign_names(indices: set[int]) -> list[str]:
     names: list[str] = []
     for idx in sorted(indices):
-        if isinstance(idx, int) and 0 <= idx <= 11:
-            names.append(SIGNS_CN[SIGNS[idx]])
+        name = _sign_name(idx)
+        if name:
+            names.append(name)
     return names
 
 
@@ -207,11 +214,17 @@ def window_scan(
                 before = previous[layer]
                 after = current[layer]
                 if isinstance(before, int) and isinstance(after, int) and before != after:
-                    transitions.append({
+                    row = {
                         "layer": layer,
                         "at": time,
                         "user_meaning": f"{label} 在 {time} 发生变化",
-                    })
+                    }
+                    from_sign = _sign_name(before)
+                    to_sign = _sign_name(after)
+                    if from_sign and to_sign:
+                        row["from_sign"] = from_sign
+                        row["to_sign"] = to_sign
+                    transitions.append(row)
         previous = current
     payload: dict[str, Any] = {
         "scanned": True,