#!/usr/bin/env python3 """Stress-test rubric weights and report scholarly-work order instability.""" from __future__ import annotations import argparse import itertools from pathlib import Path from typing import Any import _common def _normalized(weights: dict[str, float]) -> dict[str, float]: total = sum(weights.values()) return {criterion_id: weight / total for criterion_id, weight in weights.items()} def _scenarios( base_weights: dict[str, float], delta: float ) -> list[tuple[str, str | None, float, dict[str, float]]]: scenarios = [("base", None, 1.0, dict(base_weights))] for criterion_id in sorted(base_weights): for label, multiplier in (("decrease", 1.0 - delta), ("increase", 1.0 + delta)): changed = dict(base_weights) changed[criterion_id] *= multiplier scenarios.append( ( f"{criterion_id}:{label}", criterion_id, multiplier, _normalized(changed), ) ) return scenarios def _direction(first: float | None, second: float | None) -> int | None: if first is None or second is None: return None difference = first - second if abs(difference) <= 1e-9: return 0 return 1 if difference > 0 else -1 def analyze( rubric: dict[str, Any], evaluations: list[dict[str, Any]], delta: float, ) -> dict[str, Any]: rubric_issues = _common.validate_rubric(rubric) _common.require_valid(rubric_issues) if not 2 <= len(evaluations) <= _common.MAX_EVALUATIONS: raise _common.ValidationError("EVALUATION_COUNT_INVALID") work_ids: set[str] = set() evaluation_ids: set[str] = set() for index, evaluation in enumerate(evaluations): issues = _common.validate_evaluation(evaluation, rubric) _common.require_valid(issues) if evaluation["work_id"] in work_ids: raise _common.ValidationError( "WORK_ID_DUPLICATE", f"$.evaluations[{index}].work_id" ) if evaluation["evaluation_id"] in evaluation_ids: raise _common.ValidationError( "EVALUATION_ID_DUPLICATE", f"$.evaluations[{index}].evaluation_id" ) work_ids.add(evaluation["work_id"]) evaluation_ids.add(evaluation["evaluation_id"]) if not 0 < delta <= 0.5: raise _common.ValidationError("DELTA_OUT_OF_RANGE") base_weights = _common.weights_by_criterion(rubric) scenario_output: list[dict[str, Any]] = [] score_ranges = { work_id: {"minimum": None, "maximum": None} for work_id in sorted(work_ids) } base_scores: dict[str, float | None] = {} pair_changes: set[tuple[str, str]] = set() pairs = list(itertools.combinations(sorted(work_ids), 2)) incomplete_work_ids: set[str] = set() for scenario_index, ( scenario_id, criterion_id, multiplier, weights, ) in enumerate(_scenarios(base_weights, delta)): item_scores: list[dict[str, Any]] = [] score_lookup: dict[str, float | None] = {} for evaluation in evaluations: score_report = _common.score_evaluation( rubric, evaluation, weights=weights ) score = score_report["aggregates"]["normalized_score"] coverage = score_report["aggregates"]["coverage_of_applicable_weight"] if score is None: raise _common.ValidationError( "SENSITIVITY_SCORE_UNAVAILABLE", evaluation["work_id"] ) if coverage is None or coverage < 1: incomplete_work_ids.add(evaluation["work_id"]) score_lookup[evaluation["work_id"]] = score current_range = score_ranges[evaluation["work_id"]] current_range["minimum"] = ( score if current_range["minimum"] is None else min(current_range["minimum"], score) ) current_range["maximum"] = ( score if current_range["maximum"] is None else max(current_range["maximum"], score) ) item_scores.append( { "work_id": evaluation["work_id"], "normalized_score": score, "coverage_of_applicable_weight": coverage, } ) if scenario_index == 0: base_scores = score_lookup else: for first, second in pairs: if _direction(base_scores[first], base_scores[second]) != _direction( score_lookup[first], score_lookup[second] ): pair_changes.add((first, second)) scenario_output.append( { "scenario_id": scenario_id, "perturbed_criterion_id": criterion_id, "weight_multiplier": _common.rounded(multiplier), "weights": { key: _common.rounded(value) for key, value in sorted(weights.items()) }, "item_scores": sorted(item_scores, key=lambda item: item["work_id"]), } ) base_order = [ work_id for work_id, _ in sorted( base_scores.items(), key=lambda item: ( -(item[1] if item[1] is not None else float("-inf")), item[0], ), ) ] range_output = [ { "work_id": work_id, "minimum_score": _common.rounded(bounds["minimum"]), "maximum_score": _common.rounded(bounds["maximum"]), "range_width": _common.rounded( bounds["maximum"] - bounds["minimum"] if bounds["minimum"] is not None and bounds["maximum"] is not None else None ), } for work_id, bounds in sorted(score_ranges.items()) ] warnings: list[str] = [] if incomplete_work_ids: warnings.append("INCOMPLETE_COVERAGE_LIMITS_COMPARABILITY") if pair_changes: warnings.append("ORDINAL_ORDER_CHANGES_UNDER_WEIGHT_PERTURBATION") return { "schema_version": _common.SCHEMA_VERSION, "report_type": "weight_sensitivity_and_rank_instability", "notice": _common.NOTICE, "rubric_id": rubric["rubric_id"], "purpose": _common.ALLOWED_PURPOSE, "unit_of_assessment": _common.ALLOWED_UNIT, "delta": delta, "perturbation_method": ( "multiply one criterion weight by 1-delta and 1+delta, then " "renormalize all weights to sum to one" ), "scenario_count": len(scenario_output), "base_ordinal_order": base_order, "base_order_is_a_decision_recommendation": False, "rank_instability_detected": bool(pair_changes), "changed_pair_count": len(pair_changes), "total_pair_count": len(pairs), "changed_work_pairs": [ {"first_work_id": first, "second_work_id": second} for first, second in sorted(pair_changes) ], "incomplete_coverage_work_ids": sorted(incomplete_work_ids), "score_ranges": range_output, "scenarios": scenario_output, "warnings": warnings, "limitations": [ "This is a deterministic local stress test, not a validity study.", "Ordinal order must not be used to rank people or make high-impact decisions.", "Results depend on the submitted rubric, ratings, missingness, and perturbation size.", ], "decision_recommendation_provided": False, } def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description=( "Perturb rubric weights and report score ranges and ordinal-order " "changes for scholarly works only. Never use this output to rank people." ) ) parser.add_argument("--rubric", required=True, type=Path, help="Local rubric JSON") parser.add_argument( "--evaluation", required=True, type=Path, action="append", help="Local evaluation JSON; repeat for 2 to 50 distinct scholarly works", ) parser.add_argument( "--delta", type=float, default=0.2, help="Relative one-at-a-time weight change in (0, 0.5] (default: 0.2)", ) parser.add_argument("--output", type=Path, help="Optional local JSON report") parser.add_argument( "--force", action="store_true", help="Replace an existing output JSON file" ) return parser def main() -> int: args = build_parser().parse_args() try: report = analyze( _common.read_json(args.rubric), [_common.read_json(path) for path in args.evaluation], args.delta, ) _common.write_json(report, args.output, force=args.force) return 0 except _common.ValidationError as error: _common.write_json(_common.failure_report(error), None) return 2 if __name__ == "__main__": raise SystemExit(main())