Skills › Scientific Agent Skills › scikit-survival › model_report.py
model_report.py
303 lines · skills/scikit-survival/scripts/model_report.py at 5f2bc03
1#!/usr/bin/env python3
2"""Render a bounded Markdown model card from local aggregate JSON summaries."""
4from __future__ import annotations
6import argparse
7import math
8from typing import Any
10from _common import CliError, emit_text, load_json
13COMPETING_RISK_CHOICES = ("not-assessed", "absent", "present")
16def _mapping(value: Any, label: str) -> dict[str, Any]:
17 if not isinstance(value, dict):
18 raise CliError(f"{label} must be a JSON object")
19 return value
22def _scalar(value: Any, default: str = "not supplied") -> str:
23 """Render a bounded scalar without accepting row-level structures."""
25 if value is None:
26 return default
27 if isinstance(value, bool):
28 return "yes" if value else "no"
29 if isinstance(value, int):
30 return str(value)
31 if isinstance(value, float):
32 if not math.isfinite(value):
33 raise CliError("report values must be finite")
34 return f"{value:.4g}"
35 if isinstance(value, str):
36 cleaned = " ".join(value.split())
37 if len(cleaned) > 160:
38 raise CliError("report strings must be at most 160 characters")
39 return cleaned.replace("`", "'")
40 raise CliError("model report accepts aggregate scalar values only")
43def synthetic_training_summary() -> dict[str, Any]:
44 """Return a deterministic aggregate example with no person-level data."""
46 return {
47 "data": {
48 "censored_test": 18,
49 "censored_train": 54,
50 "events_test": 42,
51 "events_train": 126,
52 "rows_test": 60,
53 "rows_train": 180,
54 },
55 "evaluation": {
56 "harrell_c": 0.71,
57 "uno_c": {
58 "available": True,
59 "score": 0.69,
60 "training_censoring_distribution": True,
61 "truncation_time": 8.0,
62 },
63 },
64 "leakage_controls": {
65 "holdout_used_for_tuning": False,
66 "preprocessing_fit_scope": "training folds only",
67 "split_before_imputation_encoding_scaling": True,
68 "tuning_evaluation": "not_run",
69 },
70 "model": {
71 "best_params": None,
72 "family": "coxph",
73 "transformed_feature_count": 4,
74 },
75 "package_versions": {
76 "numpy": "2.5.1",
77 "pandas": "3.0.6",
78 "scikit-learn": "1.9.1",
79 "scikit-survival": "0.28.0",
80 },
81 "schema": {
82 "categorical_features": ["segment"],
83 "event": "event",
84 "numeric_features": ["x_linear", "x_noise"],
85 "time": "time",
86 },
87 "seed": 20_260_723,
88 "source": {"kind": "synthetic"},
89 "tuning": {"performed": False},
90 }
93def synthetic_metric_summary() -> dict[str, Any]:
94 """Return deterministic aggregate metric examples."""
96 return {
97 "assumptions": {
98 "censoring_distribution_fit_on_training_only": True,
99 "independent_censoring_from_features": "required_by_IPCW_estimator",
100 "test_follow_up_within_training_support": True,
101 "time_grid_within_test_follow_up": True,
102 },
103 "metrics": {
104 "cumulative_dynamic_auc": {
105 "mean": 0.70,
106 "times": [2.0, 4.0, 6.0],
107 "values": [0.68, 0.70, 0.72],
108 },
109 "harrell_c": 0.71,
110 "uno_c": 0.69,
111 "integrated_brier_score": 0.18,
112 },
113 "source": {"kind": "synthetic"},
114 }
117def _feature_list(schema: dict[str, Any], key: str) -> str:
118 value = schema.get(key, [])
119 if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
120 raise CliError(f"schema.{key} must be a list of strings")
121 if len(value) > 256:
122 raise CliError("too many feature names in model summary")
123 return ", ".join(f"`{_scalar(item)}`" for item in value) or "none"
126def render_report(
127 training: dict[str, Any],
128 metrics: dict[str, Any],
129 *,
130 competing_risks: str,
131 title: str,
132) -> str:
133 """Render aggregate summaries while preserving metric distinctions."""
135 training = _mapping(training, "training summary")
136 metrics = _mapping(metrics, "metrics summary")
137 data = _mapping(training.get("data", {}), "training.data")
138 model = _mapping(training.get("model", {}), "training.model")
139 schema = _mapping(training.get("schema", {}), "training.schema")
140 leakage = _mapping(
141 training.get("leakage_controls", {}), "training.leakage_controls"
142 )
143 versions = _mapping(
144 training.get("package_versions", {}), "training.package_versions"
145 )
146 metric_values = _mapping(metrics.get("metrics", {}), "metrics.metrics")
147 assumptions = _mapping(metrics.get("assumptions", {}), "metrics.assumptions")
148 dynamic_auc = metric_values.get("cumulative_dynamic_auc")
149 dynamic_auc_mean = (
150 _mapping(dynamic_auc, "cumulative_dynamic_auc").get("mean")
151 if dynamic_auc is not None
152 else None
153 )
155 if competing_risks == "present":
156 competing_note = (
157 "Competing causes are present. Standard all-event metrics are not "
158 "cause-specific CIF validation; report each cause separately."
159 )
160 elif competing_risks == "absent":
161 competing_note = "No competing cause was declared for this analysis."
162 else:
163 competing_note = (
164 "Competing risks were not assessed; this must be resolved before "
165 "interpreting cause-specific event probability."
166 )
168 lines = [
169 f"# {_scalar(title)}",
170 "",
171 "## Scope",
172 "",
173 "- Aggregate, local report only; no person-level rows are embedded.",
174 "- Supplied summaries must refer to the same fitted model and held-out cohort; "
175 "this renderer does not verify that provenance.",
176 "- Predictive performance does not establish causal effects or clinical utility.",
177 "- This report is not clinical advice.",
178 "",
179 "## Data and split",
180 "",
181 f"- Training rows: {_scalar(data.get('rows_train'))}",
182 f"- Test rows: {_scalar(data.get('rows_test'))}",
183 f"- Training events / censored: {_scalar(data.get('events_train'))} / "
184 f"{_scalar(data.get('censored_train'))}",
185 f"- Test events / censored: {_scalar(data.get('events_test'))} / "
186 f"{_scalar(data.get('censored_test'))}",
187 f"- Deterministic seed: {_scalar(training.get('seed'))}",
188 "",
189 "## Explicit schema",
190 "",
191 f"- Event column: `{_scalar(schema.get('event'))}`",
192 f"- Time column: `{_scalar(schema.get('time'))}`",
193 f"- Numeric features: {_feature_list(schema, 'numeric_features')}",
194 f"- Categorical features: {_feature_list(schema, 'categorical_features')}",
195 "",
196 "## Model and leakage controls",
197 "",
198 f"- Model family: `{_scalar(model.get('family'))}`",
199 f"- Transformed features: {_scalar(model.get('transformed_feature_count'))}",
200 "- Split before fitting imputation/encoding/scaling: "
201 f"{_scalar(leakage.get('split_before_imputation_encoding_scaling'))}",
202 f"- Preprocessing fit scope: {_scalar(leakage.get('preprocessing_fit_scope'))}",
203 f"- Holdout used for tuning: {_scalar(leakage.get('holdout_used_for_tuning'))}",
204 f"- Tuning evaluation: {_scalar(leakage.get('tuning_evaluation'))}",
205 "",
206 "## Performance",
207 "",
208 f"- Harrell C (rank discrimination): {_scalar(metric_values.get('harrell_c'))}",
209 f"- Uno C (IPCW rank discrimination): {_scalar(metric_values.get('uno_c'))}",
210 "- Mean cumulative/dynamic AUC (time-specific discrimination): "
211 f"{_scalar(dynamic_auc_mean)}",
212 "- Integrated Brier score (probability prediction error; lower is better): "
213 f"{_scalar(metric_values.get('integrated_brier_score'))}",
214 "- A Brier score mixes discrimination and calibration; it is not a "
215 "standalone calibration curve.",
216 "",
217 "## Censoring and competing risks",
218 "",
219 "- IPCW censoring distribution fit on training data: "
220 f"{_scalar(assumptions.get('censoring_distribution_fit_on_training_only'))}",
221 "- Independent censoring from features: "
222 f"{_scalar(assumptions.get('independent_censoring_from_features'))}",
223 "- Evaluation grid within train/test support: "
224 f"{_scalar(assumptions.get('test_follow_up_within_training_support'))} / "
225 f"{_scalar(assumptions.get('time_grid_within_test_follow_up'))}",
226 f"- {competing_note}",
227 "",
228 "## Runtime snapshot",
229 "",
230 f"- scikit-survival: {_scalar(versions.get('scikit-survival'))}",
231 f"- scikit-learn: {_scalar(versions.get('scikit-learn'))}",
232 f"- NumPy: {_scalar(versions.get('numpy'))}",
233 f"- pandas: {_scalar(versions.get('pandas'))}",
234 "",
235 "## Required follow-up",
236 "",
237 "- Check proportional-hazards assumptions when interpreting Cox effects.",
238 "- Inspect horizon-specific calibration on independent validation data.",
239 "- Use nested CV for performance claims made during hyperparameter tuning.",
240 "- Evaluate transportability, subgroup behavior, and decision consequences "
241 "separately; these metrics alone do not establish utility.",
242 "",
243 ]
244 training_source = _mapping(training.get("source", {}), "training.source")
245 metrics_source = _mapping(metrics.get("source", {}), "metrics.source")
246 if training_source.get("kind") == "synthetic" and metrics_source.get("kind") == "synthetic":
247 lines[4:4] = ["- Synthetic example: metric values may be illustrative; this report alone is not fit evidence."]
248 return "\n".join(lines)
251def build_parser() -> argparse.ArgumentParser:
252 parser = argparse.ArgumentParser(
253 description=(
254 "Render a Markdown model report from aggregate local JSON summaries. "
255 "No inputs produces a deterministic synthetic example."
256 )
257 )
258 parser.add_argument("--training-summary", help="Optional local training .json")
259 parser.add_argument("--metrics-summary", help="Optional local metrics .json")
260 parser.add_argument(
261 "--competing-risks",
262 choices=COMPETING_RISK_CHOICES,
263 default="not-assessed",
264 )
265 parser.add_argument("--title", default="Survival model report")
266 parser.add_argument("--output", help="Optional local .md output")
267 parser.add_argument("--force", action="store_true")
268 return parser
271def main(argv: list[str] | None = None) -> int:
272 parser = build_parser()
273 args = parser.parse_args(argv)
274 try:
275 if bool(args.training_summary) != bool(args.metrics_summary):
276 raise CliError(
277 "provide both --training-summary and --metrics-summary, or neither "
278 "for the fully synthetic example"
279 )
280 training = (
281 load_json(args.training_summary)
282 if args.training_summary
283 else synthetic_training_summary()
284 )
285 metrics = (
286 load_json(args.metrics_summary)
287 if args.metrics_summary
288 else synthetic_metric_summary()
289 )
290 report = render_report(
291 training,
292 metrics,
293 competing_risks=args.competing_risks,
294 title=args.title,
295 )
296 emit_text(report, output=args.output, force=args.force)
297 except CliError as exc:
298 parser.error(str(exc))
299 return 0
302if __name__ == "__main__":
303 raise SystemExit(main())