Files
Jyotisha/scripts/research/minute_resolution_sweep.py
T
jesse-ux 2d2467dca1
Independent Staging Quality Gate / validate (push) Failing after 9m27s
Independent Staging Quality Gate / publish (push) Skipped
docs(research): measure minute-resolution scoring; no variant clears all radii
Build holdout v4 from the public AA set, correct the two v3 dates, and sweep
R1–R5 plus pairs offline. No production scoring defaults change. No
implementation brief: delivered width stays the full window on every radius.
2026-09-14 20:43:27 +08:00

600 lines
22 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.
#!/usr/bin/env python3
"""Offline sweep of minute-resolution scoring changes.
Reads production scoring modules. Does not change their defaults.
"""
from __future__ import annotations
import argparse
import json
import statistics
import sys
import traceback
from collections import defaultdict
from dataclasses import dataclass
from datetime import date, datetime
from pathlib import Path
from typing import Any, Sequence
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from scripts.active_rectification_event_engine import ( # noqa: E402
AYANAMSA,
DOMAIN_CONFIG,
NODE_MODE,
_active_vimshottari,
compute_candidate_static_contexts,
compute_event_candidate_rows,
)
from scripts.rectification.event_probes import discriminating_event_probes # noqa: E402
from scripts.rectification.refinement_packet import window_scan # noqa: E402
from scripts.rectification.scoring_service import ( # noqa: E402
_kind_adjusted_evidence,
_legacy_request,
sample_event_dates,
score_from_matrix,
scoreable_request,
)
from scripts.research.minute_resolution_lib import ( # noqa: E402
SIGNATURE_LAYERS,
aggregate_samples,
cluster_contexts,
cluster_of,
dynamic_signature_layers,
kp_changes_in_window,
kp_event_points,
metrics_from_public,
public_rows,
rescale_varga_points,
shannon_entropy,
subtract_event_floor,
)
from scripts.research.probe_supply_after_six import ( # noqa: E402
ASK_COUNT,
apply_answer,
optimal_answer,
remaining_after_six,
request_from_case,
top1_hit,
)
HOLDOUT = ROOT / "references" / "real_case_calibration" / "minute_rectification_holdout_v4.json"
REPORT_MD = ROOT / "docs" / "research" / "minute_resolution_2026_09_14.md"
REPORT_JSON = ROOT / "docs" / "research" / "minute_resolution_2026_09_14.json"
TODAY = date(2026, 9, 14)
RADII = (10, 30, 60)
MINUTE_STEP = 2
ASK = ASK_COUNT
@dataclass(frozen=True)
class Variant:
name: str
r1: str = "2len"
r2: bool = False
r3: float | None = None
r4: str = "mean"
r5: bool = False
@property
def uses_r3(self) -> bool:
return self.r3 is not None
def all_variants() -> list[Variant]:
rows = [
Variant("baseline"),
Variant("R1@len", r1="len"),
Variant("R1@sqrt", r1="sqrt"),
Variant("R1@fixed2", r1="fixed2"),
Variant("R2", r2=True),
Variant("R3@0.5", r3=0.5),
Variant("R3@1.0", r3=1.0),
Variant("R3@2.0", r3=2.0),
Variant("R4@max", r4="max"),
Variant("R4@lse", r4="lse"),
Variant("R5", r5=True),
]
keys = ("R1", "R2", "R3", "R4", "R5")
for i, left in enumerate(keys):
for right in keys[i + 1:]:
pair = {left, right}
rows.append(Variant(
f"{left}+{right}",
r1="len" if "R1" in pair else "2len",
r2="R2" in pair,
r3=(1.0 if "R3" in pair else None),
r4="lse" if "R4" in pair else "mean",
r5="R5" in pair,
))
return rows
def scoring_request_for(case: dict[str, Any], radius: int) -> dict[str, Any]:
previous = case.get("candidate_radius_minutes")
case["candidate_radius_minutes"] = radius
try:
request = request_from_case(case)
finally:
if previous is None:
case.pop("candidate_radius_minutes", None)
else:
case["candidate_radius_minutes"] = previous
request["minute_step"] = MINUTE_STEP
request["ayanamsa"] = AYANAMSA
request["node_mode"] = NODE_MODE
return scoreable_request(request)
def event_at(sample: str) -> datetime:
return datetime.strptime(f"{sample} 12:00:00", "%Y-%m-%d %H:%M:%S")
def collect_samples(
request: dict[str, Any],
static_contexts: Sequence[dict[str, Any]],
) -> dict[str, Any]:
times = [
item["feature"]["time"] if isinstance(item.get("feature"), dict) else item["candidate_at"].strftime("%H:%M")
for item in static_contexts
]
events = list(request["events"])
payload: dict[str, dict[str, list[dict[str, Any]]]] = {event["id"]: {time: [] for time in times} for event in events}
kp_by_event: dict[str, dict[str, list[float]]] = {event["id"]: {time: [] for time in times} for event in events}
domains = [str(event["domain"]) for event in events]
for event in events:
samples = sample_event_dates(event)
for sampled in samples:
rows = list(compute_event_candidate_rows(
_legacy_request(request, event, sampled),
static_contexts=static_contexts,
))
by_time = {str(row["time"])[:5]: row for row in rows}
event_dt = event_at(sampled)
for context in static_contexts:
time = str(context["feature"]["time"])[:5]
row = by_time.get(time)
if row is None or not row.get("evidence"):
continue
adjusted = _kind_adjusted_evidence(event, row["evidence"][0])
payload[event["id"]][time].append({
"points": float(adjusted["points"]),
"rule_ids": list(adjusted["rule_ids"]),
})
moon = context["planet_longitudes"]["Moon"]
try:
vim = _active_vimshottari(request["birth_date"], float(moon), event_dt)
except (KeyError, TypeError, ValueError):
vim = ("", "", "")
kp_by_event[event["id"]][time].append(
kp_event_points(context, event["domain"], vim, 1.0),
)
return {
"times": times,
"events": events,
"domains": domains,
"samples": payload,
"kp_samples": kp_by_event,
"static_contexts": list(static_contexts),
}
def apply_variant(bundle: dict[str, Any], variant: Variant) -> dict[str, dict[str, dict[str, Any]]]:
matrix: dict[str, dict[str, dict[str, Any]]] = {}
for event in bundle["events"]:
event_id = event["id"]
domain = event["domain"]
kind = event["event_kind"]
precision = event["precision"]
by_time: dict[str, float] = {}
rules_by_time: dict[str, list[str]] = {}
for time, samples in bundle["samples"][event_id].items():
if not samples:
continue
scaled = [
rescale_varga_points(item["points"], item["rule_ids"], domain, kind, precision, variant.r1)
for item in samples
]
kp_scaled = [
round(value * float(variant.r3), 4)
for value in bundle["kp_samples"][event_id].get(time, [])
] if variant.r3 is not None else [0.0] * len(scaled)
combined = [left + right for left, right in zip(scaled, kp_scaled)] or scaled
by_time[time] = aggregate_samples(combined, variant.r4)
rules_by_time[time] = sorted({rule for item in samples for rule in item["rule_ids"]})
if variant.r2:
by_time = subtract_event_floor(by_time)
matrix[event_id] = {
time: {"points": score, "rule_ids": rules_by_time.get(time, [])}
for time, score in by_time.items()
}
return matrix
def built_payload(bundle: dict[str, Any], matrix: dict[str, dict[str, dict[str, Any]]]) -> dict[str, Any]:
return {
"candidate_times": list(bundle["times"]),
"matrix": matrix,
"date_sensitivity": [],
"missing_layers": [],
"static_contexts": bundle["static_contexts"],
}
def entropy_curve(
*,
request: dict[str, Any],
built: dict[str, Any],
prior: dict[str, float],
true_time: str,
clusters: Sequence[dict[str, Any]],
) -> dict[str, Any]:
times = list(prior)
initial = discriminating_event_probes(
{**request, "refresh_probes": False},
built,
scan=window_scan(built),
candidate_times=times,
representative_time=true_time,
today=TODAY,
)
asked = initial[:ASK]
scores = dict(prior)
conflicts = {time: 0 for time in times}
eliminated: set[str] = set()
curve = [shannon_entropy(max(scores[time], 0.0) for time in times if time not in eliminated)]
coverage_curve = []
width_curve = []
for probe in asked:
answer = optimal_answer(probe, true_time)
if answer is None:
curve.append(curve[-1] if curve else 0.0)
continue
scores, conflicts, eliminated = apply_answer(
scores, conflicts, eliminated, probe, answer, times,
)
remaining, _mode, true_alive = remaining_after_six(
all_times=times,
scores=scores,
eliminated=eliminated,
clusters=clusters,
true_time=true_time,
)
active_scores = [max(scores.get(time, 0.0), 0.0) for time in remaining]
curve.append(shannon_entropy(active_scores))
coverage_curve.append(true_alive)
width_curve.append(
None if not remaining else (max(len(remaining), 0)),
)
remaining, remaining_mode, true_alive = remaining_after_six(
all_times=times,
scores=scores,
eliminated=eliminated,
clusters=clusters,
true_time=true_time,
)
return {
"entropy": curve,
"asked": len(asked),
"probes": len(initial),
"remaining_count": len(remaining),
"remaining_mode": remaining_mode,
"true_alive": true_alive,
"replay_top1": top1_hit(scores, remaining, true_time, clusters),
"coverage_after_answers": coverage_curve,
}
def score_variant(
*,
request: dict[str, Any],
bundle: dict[str, Any],
variant: Variant,
true_time: str,
) -> dict[str, Any]:
matrix = apply_variant(bundle, variant)
built = built_payload(bundle, matrix)
rows = score_from_matrix(request, built)
layers = dynamic_signature_layers(bundle["domains"]) if variant.r5 else SIGNATURE_LAYERS
clusters = cluster_contexts(bundle["static_contexts"], layers)
public = public_rows(rows, bundle["static_contexts"], layers)
prior = {str(row["time"])[:5]: float(row["score"]) for row in rows}
engine = metrics_from_public(public, true_time, bundle["times"])
replay = entropy_curve(
request=request,
built=built,
prior=prior,
true_time=true_time,
clusters=clusters,
)
true_cluster = cluster_of(true_time, clusters)
return {
"engine": engine,
"replay": replay,
"true_cluster_key": None if true_cluster is None else true_cluster.get("signature_key"),
"true_cluster_size": 0 if true_cluster is None else len(true_cluster.get("times") or []),
"score_range": round(
max(prior.values()) - min(prior.values()),
4,
) if prior else 0.0,
}
def kp_window_changes(bundle: dict[str, Any]) -> dict[str, int]:
changes = {}
for domain in sorted(set(bundle["domains"])):
changes[domain] = kp_changes_in_window(bundle["static_contexts"], domain)
changes["any"] = max(changes.values()) if changes else 0
return changes
def summarize(rows: Sequence[dict[str, Any]], *, r3_only: bool = False) -> dict[str, Any]:
usable = [row for row in rows if (not r3_only or row.get("kp_eligible"))]
if not usable:
return {
"n": 0,
"top1": None,
"coverage": None,
"width_median": None,
"tie": None,
"entropy0": None,
"entropy6": None,
"squeezed": None,
"too_narrow": None,
"replay_top1": None,
}
widths = [row["engine"]["width"] for row in usable if row["engine"]["width"] is not None]
entropy0 = [row["engine"]["entropy"] for row in usable]
entropy6 = [
(row["replay"]["entropy"][-1] if row["replay"]["entropy"] else None)
for row in usable
]
entropy6 = [item for item in entropy6 if item is not None]
return {
"n": len(usable),
"top1": round(sum(1 for row in usable if row["engine"]["top1_hit"]) / len(usable), 4),
"coverage": round(sum(1 for row in usable if row["engine"]["coverage"]) / len(usable), 4),
"width_median": statistics.median(widths) if widths else None,
"tie": round(sum(1 for row in usable if row["engine"]["tie"]) / len(usable), 4),
"entropy0": round(sum(entropy0) / len(entropy0), 4) if entropy0 else None,
"entropy6": round(sum(entropy6) / len(entropy6), 4) if entropy6 else None,
"squeezed": sum(1 for row in usable if row["engine"]["truth_squeezed"]),
"too_narrow": sum(1 for row in usable if row["engine"]["too_narrow"]),
"replay_top1": round(sum(1 for row in usable if row["replay"]["replay_top1"]) / len(usable), 4),
}
def verdict(baseline: dict[str, Any], candidate: dict[str, Any]) -> str:
if baseline["n"] == 0 or candidate["n"] == 0:
return "uncertain"
if any(candidate[key] is None or baseline[key] is None for key in ("top1", "coverage", "tie", "width_median")):
return "uncertain"
hit_ok = candidate["top1"] + 1e-9 >= baseline["top1"]
cover_ok = candidate["coverage"] + 1e-9 >= baseline["coverage"]
tie_down = candidate["tie"] <= baseline["tie"] + 1e-9
width_down = candidate["width_median"] <= baseline["width_median"] + 1e-9
squeezed_ok = candidate["squeezed"] <= baseline["squeezed"]
if hit_ok and cover_ok and tie_down and width_down and squeezed_ok:
if (
candidate["top1"] > baseline["top1"] + 1e-9
or candidate["tie"] < baseline["tie"] - 1e-9
or candidate["width_median"] < baseline["width_median"] - 1e-9
):
return "benefit"
return "no_benefit"
if not hit_ok or not cover_ok or candidate["squeezed"] > baseline["squeezed"]:
return "no_benefit"
return "uncertain"
def md_table(headers: Sequence[str], rows: Sequence[Sequence[Any]]) -> str:
def cell(value: Any) -> str:
if value is None:
return "—"
if isinstance(value, float):
return f"{value:.3f}".rstrip("0").rstrip(".")
return str(value)
line = "| " + " | ".join(headers) + " |"
sep = "| " + " | ".join("---" if index == 0 else "---:" for index in range(len(headers))) + " |"
body = ["| " + " | ".join(cell(item) for item in row) + " |" for row in rows]
return "\n".join([line, sep, *body])
def write_report(payload: dict[str, Any]) -> None:
lines = [
"# 分钟分辨率打分尺度测量(2026-09-14",
"",
f"- 口径:ayanamsa `{payload['ayanamsa']}`node mode `{payload['node_mode']}`。不得与上游 true-node 数字直接对比。",
f"- 数据:`{payload['holdout']}`{payload['case_count']} 例公开 Rodden-AA,每例 ≥7 件事、≥4 个领域。",
f"- 半径:{', '.join(f{item}' for item in payload['radii'])},步长 {MINUTE_STEP} 分钟。",
"- 性质:离线测量。生产 `active_rectification_event_engine.py` / `scoring_service.py` / `candidate_contrast.py` 默认值未改。",
"- D1:KP 宫头子主计分方向已拍板;本文件只报告权重收益,不改线上默认。",
"",
"## 结论",
"",
]
overall = payload["verdicts"]
lines.append("| 改法 | 判定 | 说明 |")
lines.append("| --- | --- | --- |")
for name, row in overall.items():
lines.append(f"| {name} | **{row['verdict']}** | {row['note']} |")
lines.extend(["", "## 基线成绩单", ""])
base_rows = []
for radius, summary in payload["baseline"].items():
base_rows.append([
f{radius}",
summary["n"],
summary["top1"],
summary["coverage"],
summary["width_median"],
summary["tie"],
summary["entropy0"],
summary["squeezed"],
])
lines.append(md_table(
["半径", "N", "真分钟在头名簇", "真分钟在交付区间", "区间宽度中位", "并列率", "引擎熵", "真值被挤出"],
base_rows,
))
lines.extend(["", "## 分半径指标", ""])
for radius, variants in payload["by_radius"].items():
lines.append(f"### ±{radius} 分钟")
lines.append("")
table = []
for name, summary in variants.items():
table.append([
name,
summary["top1"],
summary["coverage"],
summary["width_median"],
summary["tie"],
summary["entropy0"],
summary["entropy6"],
summary["squeezed"],
summary["too_narrow"],
])
lines.append(md_table(
["方案", "头名簇命中", "区间覆盖", "宽度中位", "并列率", "熵0", "熵6", "挤出", "过窄"],
table,
))
lines.append("")
lines.extend([
"## R3 KP 窗内变化",
"",
"只统计 ±10 窗内目标宫宫头子主至少变化一次的例子。",
"",
])
kp = payload["kp"]
lines.append(md_table(
["例子", "窗内变化次数", "计入 R3"],
[[row["case_id"], row["changes"], "yes" if row["eligible"] else "no"] for row in kp],
))
lines.extend([
"",
"## 方法",
"",
"1. 每个例子先算生产静态盘(含 KP 观察),再按事件抽样日打生产分。",
"2. R1 按 `*_domain_varga` 规则把 `2*len` 换成 `len` / `sqrt(len)` / 固定 2。",
"3. R2 在事件级减去当前窗口最小值。",
"4. R3 把目标宫 KP 子主/次子主与当时 Vim MD/AD/PD 比对,权重 0.5 / 1.0 / 2.0。",
"5. R4 对年精度 12 个月样本取 max 或 log-sum-expT=1)。",
"6. R5 按本例有证据的领域动态取签名层,去掉同日恒定的 `md`。",
"7. 交付区间 = 公开簇成员分钟的并集。并列 = 头名簇分数并列。",
"8. 熵曲线用生产探针 + 真值方向最优答,最多 6 题。",
"",
"配对组合用 R1=`len`、R3=1.0、R4=`lse`。",
"",
])
REPORT_MD.write_text("\n".join(lines) + "\n", encoding="utf-8")
REPORT_JSON.write_text(json.dumps(payload, ensure_ascii=True, indent=2) + "\n", encoding="utf-8")
def run(args: argparse.Namespace) -> dict[str, Any]:
holdout = json.loads(HOLDOUT.read_text(encoding="utf-8"))
cases = list(holdout["cases"])
if args.limit:
cases = cases[: args.limit]
radii = tuple(int(item) for item in args.radii)
variants = all_variants()
if args.quick:
variants = [item for item in variants if item.name in {"baseline", "R1@len", "R2", "R3@1.0", "R4@lse", "R5"}]
results: dict[str, dict[str, list[dict[str, Any]]]] = {
str(radius): {variant.name: [] for variant in variants} for radius in radii
}
kp_rows = []
errors = []
for case in cases:
case_id = case["case_id"]
true_time = str(case["birth"]["time"])[:5]
print(f"case {case_id}", flush=True)
try:
for radius in radii:
request = scoring_request_for(case, radius)
contexts = compute_candidate_static_contexts(request)
bundle = collect_samples(request, contexts)
changes = kp_window_changes(bundle)
eligible = changes["any"] > 0
if radius == 10:
kp_rows.append({
"case_id": case_id,
"changes": changes["any"],
"by_domain": changes,
"eligible": eligible,
})
print(f" radius ±{radius} candidates={len(bundle['times'])}", flush=True)
for variant in variants:
scored = score_variant(
request=request,
bundle=bundle,
variant=variant,
true_time=true_time,
)
scored["case_id"] = case_id
scored["kp_eligible"] = eligible
results[str(radius)][variant.name].append(scored)
except Exception as exc: # noqa: BLE001
errors.append({"case_id": case_id, "error": f"{type(exc).__name__}: {exc}", "trace": traceback.format_exc()})
print(f" FAIL {case_id}: {exc}", flush=True)
by_radius = {}
baseline = {}
verdicts: dict[str, dict[str, str]] = {}
for radius in radii:
variant_summaries = {}
for variant in variants:
rows = results[str(radius)][variant.name]
variant_summaries[variant.name] = summarize(rows, r3_only=variant.uses_r3)
by_radius[str(radius)] = variant_summaries
baseline[str(radius)] = variant_summaries["baseline"]
names = [variant.name for variant in variants if variant.name != "baseline"]
for name in names:
notes = []
labels = []
for radius in radii:
base = by_radius[str(radius)]["baseline"]
cand = by_radius[str(radius)][name]
label = verdict(base, cand)
labels.append(label)
notes.append(f{radius}:{label}")
if all(item == "benefit" for item in labels):
final = "benefit"
elif all(item == "no_benefit" for item in labels):
final = "no_benefit"
else:
final = "uncertain"
verdicts[name] = {"verdict": final, "note": "".join(notes)}
payload = {
"generated_at": TODAY.isoformat(),
"ayanamsa": AYANAMSA,
"node_mode": NODE_MODE,
"holdout": str(HOLDOUT.relative_to(ROOT)).replace("\\", "/"),
"case_count": len(cases),
"radii": list(radii),
"minute_step": MINUTE_STEP,
"baseline": baseline,
"by_radius": by_radius,
"verdicts": verdicts,
"kp": kp_rows,
"errors": errors,
"results": results,
}
write_report(payload)
return payload
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--limit", type=int, default=0)
parser.add_argument("--radii", nargs="+", default=[str(item) for item in RADII])
parser.add_argument("--quick", action="store_true")
args = parser.parse_args()
payload = run(args)
print(f"wrote {REPORT_MD} cases={payload['case_count']} errors={len(payload['errors'])}")
return 0 if not payload["errors"] else 1
if __name__ == "__main__":
sys.exit(main())