20 public AA cases, raman/mean. Lowering MIN_BOUNDARY_DAYS narrows ±10 from 15 to 11 minutes but drops top-1 0.80→0.75; wider radii get wider ranges. Varga sensitivity weights (V1/V2) match production; D60 (V3) hurts ±10. Day-precision events offset ±7 never squeeze the true minute out. Production 45/30 gate and equal varga weights stay unchanged.
143 lines
5.2 KiB
Python
143 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from scripts.rectification.event_probes import MIN_BOUNDARY_DAYS, REFRESH_MIN_BOUNDARY_DAYS
|
|
import scripts.rectification.event_probes as event_probes
|
|
from scripts.research.precision_gate_lib import (
|
|
GATES,
|
|
close_same_year_windows,
|
|
count_precision,
|
|
finest_precision,
|
|
gate_verdict,
|
|
jitter_day_events,
|
|
patched_boundary_gate,
|
|
production_defaults_intact,
|
|
threshold_for,
|
|
treat_events,
|
|
varga_factor,
|
|
varga_minutes,
|
|
window_minutes_for_radius,
|
|
)
|
|
|
|
HOLDOUT = Path("references/real_case_calibration/minute_rectification_holdout_v4.json")
|
|
|
|
|
|
def _events() -> list[dict]:
|
|
return [
|
|
{"id": "day-1", "domain": "career", "date": "2011-08-24", "precision": "day"},
|
|
{"id": "year-1", "domain": "education", "date": "2003", "precision": "year"},
|
|
]
|
|
|
|
|
|
def test_holdout_v4_has_the_stated_precision_mix() -> None:
|
|
payload = json.loads(HOLDOUT.read_text(encoding="utf-8"))
|
|
tallies = count_precision([
|
|
event for case in payload["cases"] for event in case["events"]
|
|
])
|
|
assert tallies["day"] == 59
|
|
assert tallies["year"] == 84
|
|
assert tallies["month"] == 0
|
|
assert len(payload["cases"]) == 20
|
|
|
|
|
|
def test_treatment_b_downgrades_day_only() -> None:
|
|
treated = treat_events(_events(), "B")
|
|
assert treated[0]["precision"] == "month"
|
|
assert treated[1]["precision"] == "year"
|
|
|
|
|
|
def test_treatment_c_downgrades_everything_to_year() -> None:
|
|
treated = treat_events(_events(), "C")
|
|
assert {item["precision"] for item in treated} == {"year"}
|
|
|
|
|
|
def test_finest_precision_prefers_day() -> None:
|
|
assert finest_precision(_events()) == "day"
|
|
assert finest_precision(treat_events(_events(), "B")) == "month"
|
|
assert finest_precision(treat_events(_events(), "C")) == "year"
|
|
|
|
|
|
def test_g0_refresh_stays_at_production_30() -> None:
|
|
assert threshold_for("G0", "day", refresh=False) == 45
|
|
assert threshold_for("G0", "day", refresh=True) == 30
|
|
assert threshold_for("G1", "day", refresh=True) == 10
|
|
assert threshold_for("G3", "day", refresh=False) == 5
|
|
assert set(GATES) == {"G0", "G1", "G2", "G3", "G4"}
|
|
|
|
|
|
def test_jitter_stays_inside_the_span_and_is_reproducible() -> None:
|
|
first = jitter_day_events(_events(), 7, case_id="demo")
|
|
second = jitter_day_events(_events(), 7, case_id="demo")
|
|
assert first[0]["date"] == second[0]["date"]
|
|
original = date.fromisoformat("2011-08-24")
|
|
shifted = date.fromisoformat(first[0]["date"])
|
|
delta = abs((shifted - original).days)
|
|
assert 1 <= delta <= 7
|
|
assert first[1]["date"] == "2003"
|
|
|
|
|
|
def test_varga_factor_downweights_slow_charts_in_a_20_minute_window() -> None:
|
|
assert window_minutes_for_radius(10) == 20
|
|
assert varga_minutes("D30") == 4.0
|
|
assert varga_minutes("D2") == 60.0
|
|
assert varga_factor("D30", 20) == 1.0
|
|
assert abs(varga_factor("D2", 20) - 20 / 60) < 1e-9
|
|
assert varga_minutes("D4") == 7.5
|
|
|
|
|
|
def test_forty_day_same_year_window_needs_the_lower_gate() -> None:
|
|
left = date(2001, 3, 1)
|
|
right = date(2001, 4, 10)
|
|
assert close_same_year_windows(left, right, 45) == []
|
|
kept = close_same_year_windows(left, right, 30)
|
|
assert [item.month for item in kept] == [3, 4]
|
|
|
|
|
|
def test_patched_gate_restores_production_defaults() -> None:
|
|
assert event_probes.MIN_BOUNDARY_DAYS == 45
|
|
with patched_boundary_gate(initial=10, refresh=7):
|
|
assert event_probes.MIN_BOUNDARY_DAYS == 10
|
|
assert event_probes.REFRESH_MIN_BOUNDARY_DAYS == 7
|
|
assert event_probes.MIN_BOUNDARY_DAYS == 45
|
|
assert event_probes.REFRESH_MIN_BOUNDARY_DAYS == 30
|
|
assert production_defaults_intact()
|
|
assert MIN_BOUNDARY_DAYS == 45
|
|
assert REFRESH_MIN_BOUNDARY_DAYS == 30
|
|
|
|
|
|
def test_verdict_blocks_a_plus_minus_seven_squeeze() -> None:
|
|
baseline = {
|
|
"n": 20, "top1": 0.8, "coverage": 1.0, "width_median": 15,
|
|
"squeezed": 0, "probes_mean": 4, "refresh_mean": 1, "discriminative_mean": 1,
|
|
}
|
|
narrower = {**baseline, "width_median": 9, "refresh_mean": 3, "discriminative_mean": 2}
|
|
squeezed = {**narrower, "coverage": 0.95, "squeezed": 1}
|
|
assert gate_verdict(baseline, narrower, jitter7_squeezed=0) == "benefit"
|
|
assert gate_verdict(baseline, narrower, jitter7_squeezed=1) == "no_benefit"
|
|
assert gate_verdict(baseline, squeezed, jitter7_squeezed=0) == "no_benefit"
|
|
|
|
|
|
def test_eight_extra_questions_for_two_minutes_is_not_worth_it() -> None:
|
|
baseline = {
|
|
"n": 20, "top1": 0.8, "coverage": 1.0, "width_median": 15,
|
|
"squeezed": 0, "probes_mean": 4, "refresh_mean": 1, "discriminative_mean": 1,
|
|
}
|
|
costly = {
|
|
**baseline, "width_median": 13, "refresh_mean": 10, "discriminative_mean": 8,
|
|
}
|
|
assert gate_verdict(baseline, costly, jitter7_squeezed=0) == "no_benefit"
|
|
|
|
|
|
def test_more_probes_with_a_wider_range_is_not_a_benefit() -> None:
|
|
baseline = {
|
|
"n": 20, "top1": 0.55, "coverage": 1.0, "width_median": 33,
|
|
"squeezed": 0, "probes_mean": 7.1, "refresh_mean": 5.45, "discriminative_mean": 5.45,
|
|
}
|
|
wider = {
|
|
**baseline, "width_median": 37, "refresh_mean": 5.5, "discriminative_mean": 5.5,
|
|
}
|
|
assert gate_verdict(baseline, wider, jitter7_squeezed=0) == "no_benefit"
|