fix(rectification): keep proportional prior after holdout calibration (BUG-560)

Implement offset and softmax relative-support scales and a public holdout calibration script, but leave the default proportional because no scheme passed the coverage-and-width gate.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Jesse_Chen
2026-09-06 22:14:42 +08:00
parent 3a9ae736e1
commit 62afa521f4
9 changed files with 642 additions and 10 deletions
+57 -8
View File
@@ -313,20 +313,18 @@ def _quantized_score(row: CandidateScoreRow) -> Decimal:
return _decimal(row.get("score")).quantize(SCORE_QUANTUM, rounding=ROUND_HALF_UP)
def _relative_support(scores: Sequence[Decimal]) -> list[int]:
if not scores:
def _distribute_percent(weights: Sequence[Decimal]) -> list[int]:
if not weights:
return []
weights = [max(score, Decimal(0)) for score in scores]
total = sum(weights, Decimal(0))
if total == 0:
base, remainder = divmod(100, len(scores))
return [base + (1 if index < remainder else 0) for index in range(len(scores))]
base, remainder = divmod(100, len(weights))
return [base + (1 if index < remainder else 0) for index in range(len(weights))]
exact = [weight * Decimal(100) / total for weight in weights]
floors = [int(value.to_integral_value(rounding=ROUND_FLOOR)) for value in exact]
remaining = 100 - sum(floors)
order = sorted(
range(len(scores)),
range(len(weights)),
key=lambda index: (-(exact[index] - Decimal(floors[index])), index),
)
for index in order[:remaining]:
@@ -334,18 +332,69 @@ def _relative_support(scores: Sequence[Decimal]) -> list[int]:
return floors
def _relative_support_proportional(scores: Sequence[Decimal]) -> list[int]:
return _distribute_percent([max(score, Decimal(0)) for score in scores])
def _relative_support_offset(scores: Sequence[Decimal], floor: Decimal) -> list[int]:
return _distribute_percent([max(score - floor, Decimal(0)) for score in scores])
def _relative_support_softmax(scores: Sequence[Decimal], temperature: Decimal) -> list[int]:
from math import exp
if not scores:
return []
peak = max(scores)
temp = temperature if temperature > 0 else Decimal("0.5")
weights = [Decimal(str(exp(float((score - peak) / temp)))) for score in scores]
return _distribute_percent(weights)
RELATIVE_SUPPORT_MODE = "proportional"
RELATIVE_SUPPORT_TEMPERATURE = Decimal("0.5")
def _relative_support(
scores: Sequence[Decimal],
*,
floor: Decimal | None = None,
mode: str | None = None,
temperature: Decimal | None = None,
) -> list[int]:
if not scores:
return []
selected = mode or RELATIVE_SUPPORT_MODE
if selected == "softmax":
return _relative_support_softmax(
scores,
temperature if temperature is not None else RELATIVE_SUPPORT_TEMPERATURE,
)
if selected == "offset" and floor is not None:
return _relative_support_offset(scores, floor)
return _relative_support_proportional(scores)
def build_candidate_decisions(
rows: Sequence[CandidateScoreRow],
*,
result_id: str,
static_contexts: Sequence[dict[str, Any]] | None = None,
support_mode: str | None = None,
temperature: Decimal | None = None,
) -> list[dict[str, Any]]:
ranked = sorted(rows, key=lambda row: (-_quantized_score(row), row["time"]))
public_rows = select_signature_representatives(ranked, static_contexts)
if not public_rows:
return []
supports = _relative_support([_quantized_score(row) for row in public_rows])
public_scores = [_quantized_score(row) for row in public_rows]
all_scores = [_quantized_score(row) for row in ranked]
floor = min(all_scores) if all_scores else Decimal(0)
supports = _relative_support(
public_scores,
floor=floor,
mode=support_mode,
temperature=temperature,
)
decisions = []
for index, row in enumerate(public_rows):
score = _quantized_score(row)