Skills › Scientific Agent Skills › scholar-evaluation › weight_sensitivity.py
weight_sensitivity.py
270 lines · skills/scholar-evaluation/scripts/weight_sensitivity.py at 5f2bc03
1#!/usr/bin/env python3
2"""Stress-test rubric weights and report scholarly-work order instability."""
4from __future__ import annotations
6import argparse
7import itertools
8from pathlib import Path
9from typing import Any
11import _common
14def _normalized(weights: dict[str, float]) -> dict[str, float]:
15 total = sum(weights.values())
16 return {criterion_id: weight / total for criterion_id, weight in weights.items()}
19def _scenarios(
20 base_weights: dict[str, float], delta: float
21) -> list[tuple[str, str | None, float, dict[str, float]]]:
22 scenarios = [("base", None, 1.0, dict(base_weights))]
23 for criterion_id in sorted(base_weights):
24 for label, multiplier in (("decrease", 1.0 - delta), ("increase", 1.0 + delta)):
25 changed = dict(base_weights)
26 changed[criterion_id] *= multiplier
27 scenarios.append(
28 (
29 f"{criterion_id}:{label}",
30 criterion_id,
31 multiplier,
32 _normalized(changed),
33 )
34 )
35 return scenarios
38def _direction(first: float | None, second: float | None) -> int | None:
39 if first is None or second is None:
40 return None
41 difference = first - second
42 if abs(difference) <= 1e-9:
43 return 0
44 return 1 if difference > 0 else -1
47def analyze(
48 rubric: dict[str, Any],
49 evaluations: list[dict[str, Any]],
50 delta: float,
51) -> dict[str, Any]:
52 rubric_issues = _common.validate_rubric(rubric)
53 _common.require_valid(rubric_issues)
54 if not 2 <= len(evaluations) <= _common.MAX_EVALUATIONS:
55 raise _common.ValidationError("EVALUATION_COUNT_INVALID")
56 work_ids: set[str] = set()
57 evaluation_ids: set[str] = set()
58 for index, evaluation in enumerate(evaluations):
59 issues = _common.validate_evaluation(evaluation, rubric)
60 _common.require_valid(issues)
61 if evaluation["work_id"] in work_ids:
62 raise _common.ValidationError(
63 "WORK_ID_DUPLICATE", f"$.evaluations[{index}].work_id"
64 )
65 if evaluation["evaluation_id"] in evaluation_ids:
66 raise _common.ValidationError(
67 "EVALUATION_ID_DUPLICATE", f"$.evaluations[{index}].evaluation_id"
68 )
69 work_ids.add(evaluation["work_id"])
70 evaluation_ids.add(evaluation["evaluation_id"])
71 if not _common.is_number(delta) or not 0 < delta <= 0.5:
72 raise _common.ValidationError("DELTA_OUT_OF_RANGE")
74 base_weights = _common.weights_by_criterion(rubric)
75 scenario_output: list[dict[str, Any]] = []
76 score_ranges = {
77 work_id: {"minimum": None, "maximum": None} for work_id in sorted(work_ids)
78 }
79 base_scores: dict[str, float | None] = {}
80 pair_changes: set[tuple[str, str]] = set()
81 pairs = list(itertools.combinations(sorted(work_ids), 2))
82 incomplete_work_ids: set[str] = set()
84 for scenario_index, (
85 scenario_id,
86 criterion_id,
87 multiplier,
88 weights,
89 ) in enumerate(_scenarios(base_weights, delta)):
90 item_scores: list[dict[str, Any]] = []
91 score_lookup: dict[str, float | None] = {}
92 for evaluation in evaluations:
93 score_report = _common.score_evaluation(
94 rubric, evaluation, weights=weights, round_output=False
95 )
96 score = score_report["aggregates"]["normalized_score"]
97 coverage = score_report["aggregates"]["coverage_of_applicable_weight"]
98 if score is None:
99 raise _common.ValidationError(
100 "SENSITIVITY_SCORE_UNAVAILABLE", "$.evaluations"
101 )
102 if coverage is None or coverage < 1:
103 incomplete_work_ids.add(evaluation["work_id"])
104 score_lookup[evaluation["work_id"]] = score
105 current_range = score_ranges[evaluation["work_id"]]
106 current_range["minimum"] = (
107 score
108 if current_range["minimum"] is None
109 else min(current_range["minimum"], score)
110 )
111 current_range["maximum"] = (
112 score
113 if current_range["maximum"] is None
114 else max(current_range["maximum"], score)
115 )
116 item_scores.append(
117 {
118 "work_id": evaluation["work_id"],
119 "normalized_score": _common.rounded(score),
120 "coverage_of_applicable_weight": _common.rounded(coverage),
121 }
122 )
123 if scenario_index == 0:
124 base_scores = score_lookup
125 else:
126 for first, second in pairs:
127 if _direction(base_scores[first], base_scores[second]) != _direction(
128 score_lookup[first], score_lookup[second]
129 ):
130 pair_changes.add((first, second))
131 scenario_output.append(
132 {
133 "scenario_id": scenario_id,
134 "perturbed_criterion_id": criterion_id,
135 "weight_multiplier": multiplier,
136 "weights": {
137 key: value
138 for key, value in sorted(weights.items())
139 },
140 "item_scores": sorted(item_scores, key=lambda item: item["work_id"]),
141 }
142 )
144 base_order = [
145 work_id
146 for work_id, _ in sorted(
147 base_scores.items(),
148 key=lambda item: (
149 -(item[1] if item[1] is not None else float("-inf")),
150 item[0],
151 ),
152 )
153 ]
154 range_output = [
155 {
156 "work_id": work_id,
157 "minimum_score": _common.rounded(bounds["minimum"]),
158 "maximum_score": _common.rounded(bounds["maximum"]),
159 "range_width": _common.rounded(
160 bounds["maximum"] - bounds["minimum"]
161 if bounds["minimum"] is not None and bounds["maximum"] is not None
162 else None
163 ),
164 }
165 for work_id, bounds in sorted(score_ranges.items())
166 ]
167 warnings: list[str] = []
168 if incomplete_work_ids:
169 warnings.append("INCOMPLETE_COVERAGE_LIMITS_COMPARABILITY")
170 if pair_changes:
171 warnings.append("ORDINAL_ORDER_CHANGES_UNDER_WEIGHT_PERTURBATION")
172 status_patterns = {
173 tuple(sorted((rating["criterion_id"], rating["status"]) for rating in evaluation["ratings"]))
174 for evaluation in evaluations
175 }
176 if len(status_patterns) > 1:
177 warnings.append("DIFFERENT_CRITERION_STATUS_PATTERNS_LIMIT_COMPARABILITY")
178 return {
179 "schema_version": _common.SCHEMA_VERSION,
180 "report_type": "weight_sensitivity_and_rank_instability",
181 "notice": _common.NOTICE,
182 "rubric_id": rubric["rubric_id"],
183 "evaluation_records": [
184 {"evaluation_id": evaluation["evaluation_id"], "work_id": evaluation["work_id"]}
185 for evaluation in sorted(evaluations, key=lambda item: item["work_id"])
186 ],
187 "purpose": _common.ALLOWED_PURPOSE,
188 "unit_of_assessment": _common.ALLOWED_UNIT,
189 "delta": delta,
190 "perturbation_method": (
191 "multiply one criterion weight by 1-delta and 1+delta, then "
192 "renormalize all weights to sum to one"
193 ),
194 "scenario_count": len(scenario_output),
195 "base_ordinal_order": base_order,
196 "base_tied_work_pairs": [
197 {"first_work_id": first, "second_work_id": second}
198 for first, second in pairs
199 if _direction(base_scores[first], base_scores[second]) == 0
200 ],
201 "order_comparison_absolute_tolerance": 1e-9,
202 "order_comparison_uses_unrounded_scores": True,
203 "base_order_is_a_decision_recommendation": False,
204 "rank_instability_detected": bool(pair_changes),
205 "changed_pair_count": len(pair_changes),
206 "total_pair_count": len(pairs),
207 "changed_work_pairs": [
208 {"first_work_id": first, "second_work_id": second}
209 for first, second in sorted(pair_changes)
210 ],
211 "incomplete_coverage_work_ids": sorted(incomplete_work_ids),
212 "score_ranges": range_output,
213 "scenarios": scenario_output,
214 "warnings": warnings,
215 "limitations": [
216 "This is a deterministic local stress test, not a validity study.",
217 "Ordinal order must not be used to rank people or make high-impact decisions.",
218 "Results depend on the submitted rubric, ratings, missingness, and perturbation size.",
219 "Linear list positions do not order tied pairs.",
220 "Different rated/applicable criterion sets do not establish comparable constructs.",
221 ],
222 "decision_recommendation_provided": False,
223 }
226def build_parser() -> argparse.ArgumentParser:
227 parser = argparse.ArgumentParser(
228 description=(
229 "Perturb rubric weights and report score ranges and ordinal-order "
230 "changes for scholarly works only. Never use this output to rank people."
231 )
232 )
233 parser.add_argument("--rubric", required=True, type=Path, help="Local rubric JSON")
234 parser.add_argument(
235 "--evaluation",
236 required=True,
237 type=Path,
238 action="append",
239 help="Local evaluation JSON; repeat for 2 to 50 distinct scholarly works",
240 )
241 parser.add_argument(
242 "--delta",
243 type=float,
244 default=0.2,
245 help="Relative one-at-a-time weight change in (0, 0.5] (default: 0.2)",
246 )
247 parser.add_argument("--output", type=Path, help="Optional local JSON report")
248 parser.add_argument(
249 "--force", action="store_true", help="Replace an existing output JSON file"
250 )
251 return parser
254def main() -> int:
255 args = build_parser().parse_args()
256 try:
257 report = analyze(
258 _common.read_json(args.rubric),
259 [_common.read_json(path) for path in args.evaluation],
260 args.delta,
261 )
262 _common.write_json(report, args.output, force=args.force)
263 return 0
264 except _common.ValidationError as error:
265 _common.write_json(_common.failure_report(error), None)
266 return 2
269if __name__ == "__main__":
270 raise SystemExit(main())