Skills › Scientific Agent Skills › scientific-brainstorming › evaluate_matrix.py
evaluate_matrix.py
553 lines · skills/scientific-brainstorming/scripts/evaluate_matrix.py at 5f2bc03
1"""Calculate a transparent weighted idea matrix with sensitivity analysis."""
3from __future__ import annotations
5import argparse
6import csv
7import io
8import math
9import sys
10from collections.abc import Mapping, Sequence
11from dataclasses import dataclass
12from typing import Any
14from _common import (
15 CliError,
16 bounded_float,
17 emit_json,
18 finite_number,
19 read_bounded_text,
20 read_json,
21 require_identifier,
22 require_text,
23 rounded,
24)
26MAX_CRITERIA = 25
27MAX_ROWS = 1_000
28MAX_COLUMNS = 200
29MAX_CSV_FIELD_CHARS = 100_000
32@dataclass(frozen=True)
33class Criterion:
34 """One explicit criterion in the disclosed additive model."""
36 name: str
37 description: str
38 weight: float
39 direction: str
40 minimum: float
41 maximum: float
43 def normalize(self, value: float) -> float:
44 """Map a bounded raw score to zero through one."""
46 if self.direction == "higher":
47 return (value - self.minimum) / (self.maximum - self.minimum)
48 return (self.maximum - value) / (self.maximum - self.minimum)
51@dataclass(frozen=True)
52class Rating:
53 """One central rating and its raw plausible interval."""
55 score: float
56 low: float
57 high: float
58 bounds_supplied: bool = True
61@dataclass(frozen=True)
62class IdeaRow:
63 """Validated ratings and qualitative context for one idea."""
65 idea_id: str
66 ratings: Mapping[str, Rating]
67 qualitative: Mapping[str, str]
70def load_criteria(raw_path: str) -> list[Criterion]:
71 """Load and validate the explicit criteria configuration."""
73 _, document = read_json(raw_path, label="criteria configuration")
74 if not isinstance(document, dict):
75 raise CliError("criteria configuration must be a JSON object")
76 if document.get("schema_version") != "1.0":
77 raise CliError("criteria configuration schema_version must be '1.0'")
78 raw_criteria = document.get("criteria")
79 if not isinstance(raw_criteria, list) or not raw_criteria:
80 raise CliError("criteria configuration must contain a non-empty list")
81 if len(raw_criteria) > MAX_CRITERIA:
82 raise CliError(
83 f"criteria count is {len(raw_criteria)}; limit is {MAX_CRITERIA}"
84 )
86 result: list[Criterion] = []
87 seen: set[str] = set()
88 for index, raw in enumerate(raw_criteria):
89 label = f"criteria[{index}]"
90 if not isinstance(raw, dict):
91 raise CliError(f"{label} must be an object")
92 name = require_identifier(raw.get("name"), f"{label}.name")
93 if name in seen:
94 raise CliError(f"duplicate criterion name: {name}")
95 seen.add(name)
96 description = require_text(
97 raw.get("description"),
98 f"{label}.description",
99 maximum=500,
100 )
101 weight = finite_number(raw.get("weight"), f"{label}.weight")
102 if weight <= 0 or weight > 1_000_000:
103 raise CliError(f"{label}.weight must be greater than 0 and at most 1000000")
104 direction = raw.get("direction")
105 if not isinstance(direction, str) or direction not in {"higher", "lower"}:
106 raise CliError(f"{label}.direction must be 'higher' or 'lower'")
107 minimum = finite_number(raw.get("minimum"), f"{label}.minimum")
108 maximum = finite_number(raw.get("maximum"), f"{label}.maximum")
109 if minimum >= maximum:
110 raise CliError(f"{label}.minimum must be less than maximum")
111 if not math.isfinite(maximum - minimum):
112 raise CliError(f"{label} scale width must be finite; rescale the anchors")
113 result.append(
114 Criterion(
115 name=name,
116 description=description,
117 weight=weight,
118 direction=direction,
119 minimum=minimum,
120 maximum=maximum,
121 )
122 )
123 column_owners = {name: "reserved column" for name in (
124 "idea_id", "qualitative_review", "uncertainties"
125 )}
126 for criterion in result:
127 for name in (criterion.name, f"{criterion.name}_low", f"{criterion.name}_high"):
128 if name in column_owners:
129 raise CliError(
130 f"criterion {criterion.name!r} creates ambiguous CSV column "
131 f"{name!r}, already used by {column_owners[name]}"
132 )
133 column_owners[name] = f"criterion {criterion.name!r}"
134 return result
137def _parse_bounded_score(
138 raw: str | None,
139 *,
140 label: str,
141 criterion: Criterion,
142) -> float:
143 if raw is None or not raw.strip():
144 raise CliError(f"{label} must contain a score")
145 score = finite_number(raw, label)
146 if not criterion.minimum <= score <= criterion.maximum:
147 raise CliError(
148 f"{label}={score} is outside [{criterion.minimum}, {criterion.maximum}]"
149 )
150 return score
153def load_scores(raw_path: str, criteria: Sequence[Criterion]) -> list[IdeaRow]:
154 """Load bounded CSV ratings and preserve qualitative columns."""
156 _, text = read_bounded_text(
157 raw_path,
158 label="scores CSV",
159 suffixes={".csv"},
160 encoding="utf-8-sig",
161 )
162 csv.field_size_limit(MAX_CSV_FIELD_CHARS)
163 reader = csv.DictReader(io.StringIO(text, newline=""))
164 fieldnames = reader.fieldnames
165 if not fieldnames:
166 raise CliError("scores CSV has no header")
167 if len(fieldnames) > MAX_COLUMNS:
168 raise CliError(f"scores CSV has more than {MAX_COLUMNS} columns")
169 if any(name is None or not name.strip() for name in fieldnames):
170 raise CliError("scores CSV contains an empty column name")
171 if len(set(fieldnames)) != len(fieldnames):
172 raise CliError("scores CSV column names must be unique")
174 required = {"idea_id", "qualitative_review", "uncertainties"}
175 required.update(criterion.name for criterion in criteria)
176 missing = sorted(required - set(fieldnames))
177 if missing:
178 raise CliError(f"scores CSV is missing required columns: {', '.join(missing)}")
180 interval_columns: set[str] = set()
181 for criterion in criteria:
182 low_name = f"{criterion.name}_low"
183 high_name = f"{criterion.name}_high"
184 has_low = low_name in fieldnames
185 has_high = high_name in fieldnames
186 if has_low != has_high:
187 raise CliError(
188 f"uncertainty columns for {criterion.name} must include both "
189 f"{low_name} and {high_name}"
190 )
191 if has_low:
192 interval_columns.update({low_name, high_name})
194 numeric_columns = {criterion.name for criterion in criteria} | interval_columns
195 qualitative_columns = [
196 name for name in fieldnames if name not in numeric_columns and name != "idea_id"
197 ]
199 rows: list[IdeaRow] = []
200 seen_ids: set[str] = set()
201 for row_index, raw_row in enumerate(reader, start=2):
202 if row_index > MAX_ROWS + 1:
203 raise CliError(f"scores CSV exceeds the limit of {MAX_ROWS} data rows")
204 if None in raw_row:
205 raise CliError(f"row {row_index} has more values than header columns")
206 idea_id = require_identifier(raw_row.get("idea_id"), f"row {row_index}.idea_id")
207 if idea_id in seen_ids:
208 raise CliError(f"duplicate idea_id in scores CSV: {idea_id}")
209 seen_ids.add(idea_id)
211 ratings: dict[str, Rating] = {}
212 for criterion in criteria:
213 score = _parse_bounded_score(
214 raw_row.get(criterion.name),
215 label=f"row {row_index}.{criterion.name}",
216 criterion=criterion,
217 )
218 low_name = f"{criterion.name}_low"
219 high_name = f"{criterion.name}_high"
220 bounds_supplied = False
221 if low_name in interval_columns:
222 raw_low = raw_row.get(low_name)
223 raw_high = raw_row.get(high_name)
224 low_blank = raw_low is None or not raw_low.strip()
225 high_blank = raw_high is None or not raw_high.strip()
226 if low_blank != high_blank:
227 raise CliError(
228 f"row {row_index} must provide both {low_name} and "
229 f"{high_name}, or leave both blank"
230 )
231 if low_blank:
232 low = high = score
233 else:
234 bounds_supplied = True
235 low = _parse_bounded_score(
236 raw_low,
237 label=f"row {row_index}.{low_name}",
238 criterion=criterion,
239 )
240 high = _parse_bounded_score(
241 raw_high,
242 label=f"row {row_index}.{high_name}",
243 criterion=criterion,
244 )
245 else:
246 low = high = score
247 if not low <= score <= high:
248 raise CliError(
249 f"row {row_index}.{criterion.name} requires low <= score <= high"
250 )
251 ratings[criterion.name] = Rating(
252 score=score, low=low, high=high, bounds_supplied=bounds_supplied
253 )
255 qualitative: dict[str, str] = {}
256 for name in qualitative_columns:
257 allow_empty = name not in {"qualitative_review", "uncertainties"}
258 qualitative[name] = require_text(
259 raw_row.get(name, ""),
260 f"row {row_index}.{name}",
261 maximum=10_000,
262 allow_empty=allow_empty,
263 )
264 rows.append(
265 IdeaRow(
266 idea_id=idea_id,
267 ratings=ratings,
268 qualitative=qualitative,
269 )
270 )
272 if not rows:
273 raise CliError("scores CSV must contain at least one data row")
274 return rows
277def _normalized_weights(
278 criteria: Sequence[Criterion],
279 *,
280 override: Mapping[str, float] | None = None,
281) -> dict[str, float]:
282 raw = {
283 criterion.name: (
284 override[criterion.name] if override is not None else criterion.weight
285 )
286 for criterion in criteria
287 }
288 total = sum(raw.values())
289 if total <= 0:
290 raise CliError("sum of weights must be positive")
291 return {name: value / total for name, value in raw.items()}
294def _score(
295 row: IdeaRow,
296 criteria: Sequence[Criterion],
297 weights: Mapping[str, float],
298) -> float:
299 return 100.0 * sum(
300 weights[criterion.name] * criterion.normalize(row.ratings[criterion.name].score)
301 for criterion in criteria
302 )
305def _ranks(scores: Mapping[str, float]) -> dict[str, int]:
306 """Assign deterministic ordinal ranks; lexical idea ID breaks ties."""
308 order = sorted(scores, key=lambda idea_id: (-scores[idea_id], idea_id))
309 return {idea_id: index + 1 for index, idea_id in enumerate(order)}
312def calculate_matrix(
313 criteria: Sequence[Criterion],
314 rows: Sequence[IdeaRow],
315 *,
316 weight_delta: float,
317) -> dict[str, Any]:
318 """Calculate disclosed scores, intervals, and local weight sensitivity."""
320 if not 0 <= weight_delta <= 0.5:
321 raise CliError("weight_delta must be between 0 and 0.5")
323 base_weights = _normalized_weights(criteria)
324 base_scores = {row.idea_id: _score(row, criteria, base_weights) for row in rows}
325 base_ranks = _ranks(base_scores)
327 scenario_scores: list[dict[str, float]] = []
328 scenario_ranks: list[dict[str, int]] = []
329 scenario_names: list[str] = []
330 if weight_delta > 0:
331 for criterion in criteria:
332 for sign, factor in (
333 ("minus", 1.0 - weight_delta),
334 ("plus", 1.0 + weight_delta),
335 ):
336 raw_weights = {
337 item.name: (
338 item.weight * factor
339 if item.name == criterion.name
340 else item.weight
341 )
342 for item in criteria
343 }
344 weights = _normalized_weights(criteria, override=raw_weights)
345 scores = {row.idea_id: _score(row, criteria, weights) for row in rows}
346 scenario_scores.append(scores)
347 scenario_ranks.append(_ranks(scores))
348 scenario_names.append(f"{criterion.name}:{sign}")
350 results: list[dict[str, Any]] = []
351 warnings: list[str] = []
352 for row in sorted(rows, key=lambda item: item.idea_id):
353 interval_low = 0.0
354 interval_high = 0.0
355 criterion_details: dict[str, Any] = {}
356 missing_bounds: list[str] = []
357 for criterion in criteria:
358 rating = row.ratings[criterion.name]
359 if not rating.bounds_supplied:
360 missing_bounds.append(criterion.name)
361 normalized_score = criterion.normalize(rating.score)
362 if criterion.direction == "higher":
363 normalized_low = criterion.normalize(rating.low)
364 normalized_high = criterion.normalize(rating.high)
365 else:
366 normalized_low = criterion.normalize(rating.high)
367 normalized_high = criterion.normalize(rating.low)
368 weight = base_weights[criterion.name]
369 interval_low += 100.0 * weight * normalized_low
370 interval_high += 100.0 * weight * normalized_high
371 criterion_details[criterion.name] = {
372 "raw_score": rating.score,
373 "raw_interval": [rating.low, rating.high],
374 "bounds_supplied": rating.bounds_supplied,
375 "normalized_score": rounded(normalized_score),
376 "normalized_interval": [
377 rounded(normalized_low),
378 rounded(normalized_high),
379 ],
380 "normalized_weight": rounded(weight),
381 "contribution_to_score": rounded(100.0 * weight * normalized_score),
382 }
384 if missing_bounds:
385 warnings.append(
386 f"{row.idea_id} has no supplied uncertainty bounds for "
387 f"{', '.join(missing_bounds)}; those ratings are held at their "
388 "central values for calculation, not known to be certain"
389 )
391 sensitivity_scores = [scores[row.idea_id] for scores in scenario_scores] or [
392 base_scores[row.idea_id]
393 ]
394 sensitivity_ranks = [ranks[row.idea_id] for ranks in scenario_ranks] or [
395 base_ranks[row.idea_id]
396 ]
397 rank_min = min(sensitivity_ranks)
398 rank_max = max(sensitivity_ranks)
399 if rank_min != rank_max:
400 warnings.append(
401 f"{row.idea_id} changes rank from {rank_min} to {rank_max} "
402 "under one-at-a-time weight perturbations"
403 )
404 results.append(
405 {
406 "idea_id": row.idea_id,
407 "base_score": rounded(base_scores[row.idea_id]),
408 "base_presentation_rank": base_ranks[row.idea_id],
409 "input_uncertainty_score_interval": [
410 rounded(interval_low),
411 rounded(interval_high),
412 ],
413 "criteria_without_uncertainty_bounds": missing_bounds,
414 "weight_sensitivity_score_range": [
415 rounded(min(sensitivity_scores)),
416 rounded(max(sensitivity_scores)),
417 ],
418 "weight_sensitivity_rank_range": [rank_min, rank_max],
419 "criteria": criterion_details,
420 "qualitative": dict(row.qualitative),
421 }
422 )
424 return {
425 "schema_version": "1.0",
426 "method": {
427 "name": "weighted-additive-normalized-matrix",
428 "formula": (
429 "score_i = 100 * sum_j(normalized_weight_j * "
430 "direction_normalized_rating_ij)"
431 ),
432 "higher_normalization": "(x - minimum) / (maximum - minimum)",
433 "lower_normalization": "(maximum - x) / (maximum - minimum)",
434 "supplied_weights": {
435 criterion.name: criterion.weight for criterion in criteria
436 },
437 "normalized_weights": {
438 name: rounded(value) for name, value in base_weights.items()
439 },
440 "criteria": [
441 {
442 "name": criterion.name,
443 "description": criterion.description,
444 "direction": criterion.direction,
445 "minimum": criterion.minimum,
446 "maximum": criterion.maximum,
447 }
448 for criterion in criteria
449 ],
450 "weight_sensitivity": {
451 "type": "one-at-a-time proportional perturbation",
452 "delta": weight_delta,
453 "scenario_count": len(scenario_names),
454 "scenarios": scenario_names,
455 "weights_renormalized_in_each_scenario": True,
456 },
457 "tie_rule": (
458 "Scores are ordered descending for presentation; exact ties are "
459 "broken by lexical idea_id. Rank is not a decision."
460 ),
461 },
462 "results": results,
463 "warnings": warnings,
464 "decision": None,
465 "notice": (
466 "No idea was selected and no scientific conclusion was made. "
467 "Review qualitative reasons, uncertainty, dissent, evidence, and "
468 "noncompensatory feasibility/ethics/safety gates."
469 ),
470 "limitations": [
471 (
472 "The additive model permits compensation across criteria; keep "
473 "hard gates outside the score."
474 ),
475 (
476 "Input intervals are combined as simultaneous criterion-wise "
477 "bounds and are not calibrated probability intervals."
478 ),
479 (
480 "Omitted or blank bounds hold the corresponding central rating "
481 "fixed for calculation; they do not establish zero uncertainty."
482 ),
483 (
484 "Weight sensitivity changes one weight at a time and does not "
485 "cover all plausible weights, criterion dependence, scale "
486 "uncertainty, or model-form uncertainty."
487 ),
488 (
489 "Ratings and weights are judgments, not empirical measurements "
490 "of scientific truth."
491 ),
492 ],
493 }
496def build_parser() -> argparse.ArgumentParser:
497 """Create the command-line parser."""
499 parser = argparse.ArgumentParser(
500 description=(
501 "Calculate a disclosed weighted matrix from bounded CSV ratings, "
502 "including input intervals and local weight sensitivity. No network, "
503 "LLM, or automatic scientific decision is used."
504 )
505 )
506 parser.add_argument("scores", help="Input ratings .csv file")
507 parser.add_argument(
508 "--config",
509 required=True,
510 help="Explicit criteria and weights .json file",
511 )
512 parser.add_argument(
513 "--weight-delta",
514 type=bounded_float(0.0, 0.5),
515 default=0.10,
516 help=(
517 "Proportional one-at-a-time weight perturbation from 0 through 0.5 "
518 "(default: 0.10)"
519 ),
520 )
521 parser.add_argument(
522 "--output",
523 help="Optional result .json file in an existing directory",
524 )
525 parser.add_argument(
526 "--force",
527 action="store_true",
528 help="Explicitly replace an existing regular output file",
529 )
530 return parser
533def main(argv: Sequence[str] | None = None) -> int:
534 """Run matrix evaluation."""
536 args = build_parser().parse_args(argv)
537 try:
538 criteria = load_criteria(args.config)
539 rows = load_scores(args.scores, criteria)
540 result = calculate_matrix(
541 criteria,
542 rows,
543 weight_delta=args.weight_delta,
544 )
545 emit_json(result, args.output, force=args.force)
546 except (CliError, csv.Error) as exc:
547 print(f"error: {exc}", file=sys.stderr)
548 return 2
549 return 0
552if __name__ == "__main__":
553 raise SystemExit(main())