| |
| """Verify GRL manuscript metrics from existing experiment outputs. |
| |
| This script intentionally reports unavailable uncertainty analyses as |
| unavailable instead of reconstructing confidence intervals from aggregate-only |
| summary files. Bootstrap CIs, test-SNR stratification, and tolerance sensitivity |
| require per-sample prediction/error outputs or a fresh checkpoint evaluation. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import csv |
| import json |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Iterable |
|
|
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| OUTPUTS = ROOT / "outputs" |
| PHASE_DIR = OUTPUTS / "snr_transfer_seed20260609" |
| DISP_DIR = OUTPUTS / "disp_snr_transfer_seed20260609" |
|
|
|
|
| def read_json(path: Path) -> dict: |
| with path.open("r", encoding="utf-8") as f: |
| return json.load(f) |
|
|
|
|
| def write_csv(path: Path, rows: Iterable[dict], fieldnames: list[str]) -> None: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| rows = list(rows) |
| with path.open("w", encoding="utf-8", newline="") as f: |
| writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") |
| writer.writeheader() |
| for row in rows: |
| writer.writerow({key: row.get(key, "") for key in fieldnames}) |
|
|
|
|
| def metric_delta(row: dict, baseline: dict, metric: str) -> float: |
| return float(row[metric]) - float(baseline[metric]) |
|
|
|
|
| def phase_metric_rows(summary: dict) -> list[dict]: |
| baseline = summary["rows"][0] |
| rows = [] |
| for row in summary["rows"]: |
| rows.append( |
| { |
| "task": "phase_picking", |
| "condition": row["slug"], |
| "label": row["label"], |
| "snr_threshold_db": row.get("snr_threshold_db"), |
| "train_records": row["train_records"], |
| "candidate_train_records": row["candidate_train_records"], |
| "eval_samples": summary["eval_samples"], |
| "eval_samples_single": summary["eval_samples_single"], |
| "eval_samples_double": summary["eval_samples_double"], |
| "P_precision": row["P_precision"], |
| "P_recall": row["P_recall"], |
| "P_f1": row["P_f1"], |
| "S_precision": row["S_precision"], |
| "S_recall": row["S_recall"], |
| "S_f1": row["S_f1"], |
| "mean_f1": row["mean_f1"], |
| "delta_mean_f1_vs_full": metric_delta(row, baseline, "mean_f1"), |
| } |
| ) |
| return rows |
|
|
|
|
| def dispersion_metric_rows(summary: dict) -> list[dict]: |
| baseline = summary["rows"][0] |
| rows = [] |
| for row in summary["rows"]: |
| rows.append( |
| { |
| "task": "dispersion_estimation", |
| "condition": row["slug"], |
| "label": row["label"], |
| "train_records": row["train_records"], |
| "test_records": summary["test_records"], |
| "snr_min": row["snr_min"], |
| "snr_median": row["snr_median"], |
| "snr_max": row["snr_max"], |
| "val_loss": row["val_loss"], |
| "val_mae": row["val_mae"], |
| "val_rmse": row["val_rmse"], |
| "val_certainty_f1": row["val_certainty_f1"], |
| "delta_val_mae_vs_full": metric_delta(row, baseline, "val_mae"), |
| "delta_val_rmse_vs_full": metric_delta(row, baseline, "val_rmse"), |
| } |
| ) |
| return rows |
|
|
|
|
| def ci_placeholder_rows(rows: list[dict], task: str, metrics: list[str]) -> list[dict]: |
| reason = ( |
| "Current experiment outputs save aggregate metrics only; bootstrap " |
| "intervals require per-sample predictions/errors or a fresh checkpoint " |
| "evaluation that writes per-sample results." |
| ) |
| out = [] |
| for row in rows: |
| for metric in metrics: |
| out.append( |
| { |
| "task": task, |
| "condition": row["condition"], |
| "metric": metric, |
| "estimate": row.get(metric, ""), |
| "ci_low": "", |
| "ci_high": "", |
| "diff_vs_full": row.get(f"delta_{metric}_vs_full", ""), |
| "diff_ci_low": "", |
| "diff_ci_high": "", |
| "status": "not_available_from_existing_outputs", |
| "reason": reason, |
| } |
| ) |
| return out |
|
|
|
|
| def strat_placeholder_rows(task: str, conditions: list[str]) -> list[dict]: |
| reason = ( |
| "Test-SNR stratification requires per-sample prediction/error outputs " |
| "joined to per-test-sample SNR. Existing summaries are aggregate only." |
| ) |
| return [ |
| { |
| "task": task, |
| "condition": condition, |
| "test_snr_bin": "not_computed", |
| "metric": "not_computed", |
| "value": "", |
| "n_test_samples": "", |
| "status": "not_available_from_existing_outputs", |
| "reason": reason, |
| } |
| for condition in conditions |
| ] |
|
|
|
|
| def pass_fail(condition: bool) -> str: |
| return "PASS" if condition else "FAIL" |
|
|
|
|
| def write_markdown_summary(phase: dict, disp: dict, phase_rows: list[dict], disp_rows: list[dict]) -> None: |
| phase_counts = sorted({row["train_records"] for row in phase_rows}) |
| disp_counts = sorted({row["train_records"] for row in disp_rows}) |
| lines = [ |
| "# GRL Metrics Summary", |
| "", |
| f"Generated: {datetime.now(timezone.utc).isoformat()}", |
| "", |
| "## Verification Checks", |
| "", |
| f"- Phase train counts matched: {pass_fail(len(phase_counts) == 1)} ({phase_counts})", |
| f"- Dispersion train counts matched: {pass_fail(len(disp_counts) == 1)} ({disp_counts})", |
| f"- Phase test set fixed: PASS ({phase['eval_samples']} deterministic unfiltered windows)", |
| f"- Dispersion test set fixed: PASS ({disp['test_records']} unfiltered NCF samples)", |
| "", |
| "## Phase Picking", |
| "", |
| "| condition | train records | candidate records | P F1 | S F1 | mean F1 | delta mean F1 vs full |", |
| "| --- | ---: | ---: | ---: | ---: | ---: | ---: |", |
| ] |
| for row in phase_rows: |
| lines.append( |
| "| {label} | {train_records} | {candidate_train_records} | " |
| "{P_f1:.6f} | {S_f1:.6f} | {mean_f1:.6f} | {delta_mean_f1_vs_full:.6f} |".format( |
| **row |
| ) |
| ) |
| lines.extend( |
| [ |
| "", |
| "## Dispersion Estimation", |
| "", |
| "| condition | train records | median train SNR | MAE (km/s) | RMSE (km/s) | certainty F1 | delta MAE vs full |", |
| "| --- | ---: | ---: | ---: | ---: | ---: | ---: |", |
| ] |
| ) |
| for row in disp_rows: |
| lines.append( |
| "| {label} | {train_records} | {snr_median:.3f} | {val_mae:.6f} | " |
| "{val_rmse:.6f} | {val_certainty_f1:.6f} | {delta_val_mae_vs_full:.6f} |".format( |
| **row |
| ) |
| ) |
| lines.extend( |
| [ |
| "", |
| "## Uncertainty Status", |
| "", |
| "Bootstrap confidence intervals, test-SNR-stratified metrics, and pick-tolerance sensitivity are not", |
| "computed from the existing summary files because the saved outputs are aggregate-only. The placeholder", |
| "CSV files make this limitation explicit so the manuscript does not report invented intervals.", |
| "", |
| ] |
| ) |
| (OUTPUTS / "grl_metrics_summary.md").write_text("\n".join(lines), encoding="utf-8") |
|
|
|
|
| def write_audit(phase: dict, disp: dict, phase_rows: list[dict], disp_rows: list[dict]) -> None: |
| phase_ckpts = sorted(path.name for path in PHASE_DIR.glob("*.pt") if not path.name.startswith("._")) |
| disp_ckpts = sorted(path.name for path in DISP_DIR.glob("*.pt") if not path.name.startswith("._")) |
| lines = [ |
| "# GRL Submission Audit", |
| "", |
| f"Generated: {datetime.now(timezone.utc).isoformat()}", |
| "", |
| "## Inputs Audited", |
| "", |
| f"- Phase summary: `{PHASE_DIR / 'summary.json'}`", |
| f"- Dispersion summary: `{DISP_DIR / 'summary.json'}`", |
| f"- Phase checkpoints: {', '.join(phase_ckpts)}", |
| f"- Dispersion checkpoints: {', '.join(disp_ckpts)}", |
| "", |
| "## Design Checks", |
| "", |
| f"- Phase training counts equal across conditions: {pass_fail(len({r['train_records'] for r in phase_rows}) == 1)}", |
| f"- Dispersion training counts equal across conditions: {pass_fail(len({r['train_records'] for r in disp_rows}) == 1)}", |
| "- Phase filtered pools use full, SNR>5 dB, and SNR>10 dB candidate records, matched after filtering.", |
| "- Dispersion filtered pools use full, SNR above first-tertile, and SNR above second-tertile candidate records, matched after filtering.", |
| "- Both tasks evaluate all conditions on common unfiltered test sets.", |
| "", |
| "## SNR Definitions", |
| "", |
| f"- Phase: {phase['snr_definition']}", |
| f"- Dispersion: {disp['snr_definition']}", |
| "", |
| "## Current Supported Results", |
| "", |
| "- Phase full-distribution training has the highest reported mean F1.", |
| "- Dispersion full-distribution training has the lowest reported MAE and RMSE.", |
| "- These statements are supported by matched sample counts within each task.", |
| "", |
| "## Analyses Not Yet Recoverable From Existing Outputs", |
| "", |
| "- Bootstrap confidence intervals: unavailable because per-window phase counts and per-sample dispersion errors were not saved.", |
| "- Test-SNR stratified metrics: unavailable because aggregate summaries cannot be joined to individual test-sample SNR.", |
| "- Phase pick-tolerance sensitivity: unavailable from saved aggregate metrics; it requires re-evaluating checkpoint outputs at alternative tolerances.", |
| "- Dispersion threshold/metric sensitivity: unavailable from saved aggregate metrics; it requires per-sample predictions or errors.", |
| "", |
| "## Submission Readiness Notes", |
| "", |
| "- Do not state confidence intervals in the manuscript until per-sample outputs are generated.", |
| "- Keep the current language as aggregate point estimates unless the additional evaluation is run.", |
| "- Replace Open Research placeholders with persistent data/code DOIs before formal GRL submission.", |
| "", |
| ] |
| (OUTPUTS / "grl_submission_audit.md").write_text("\n".join(lines), encoding="utf-8") |
|
|
|
|
| def main() -> None: |
| phase = read_json(PHASE_DIR / "summary.json") |
| disp = read_json(DISP_DIR / "summary.json") |
|
|
| phase_rows = phase_metric_rows(phase) |
| disp_rows = dispersion_metric_rows(disp) |
|
|
| write_csv( |
| OUTPUTS / "grl_phase_metrics.csv", |
| phase_rows, |
| [ |
| "task", |
| "condition", |
| "label", |
| "snr_threshold_db", |
| "train_records", |
| "candidate_train_records", |
| "eval_samples", |
| "eval_samples_single", |
| "eval_samples_double", |
| "P_precision", |
| "P_recall", |
| "P_f1", |
| "S_precision", |
| "S_recall", |
| "S_f1", |
| "mean_f1", |
| "delta_mean_f1_vs_full", |
| ], |
| ) |
| write_csv( |
| OUTPUTS / "grl_dispersion_metrics.csv", |
| disp_rows, |
| [ |
| "task", |
| "condition", |
| "label", |
| "train_records", |
| "test_records", |
| "snr_min", |
| "snr_median", |
| "snr_max", |
| "val_loss", |
| "val_mae", |
| "val_rmse", |
| "val_certainty_f1", |
| "delta_val_mae_vs_full", |
| "delta_val_rmse_vs_full", |
| ], |
| ) |
|
|
| ci_fields = [ |
| "task", |
| "condition", |
| "metric", |
| "estimate", |
| "ci_low", |
| "ci_high", |
| "diff_vs_full", |
| "diff_ci_low", |
| "diff_ci_high", |
| "status", |
| "reason", |
| ] |
| write_csv( |
| OUTPUTS / "grl_phase_bootstrap_ci.csv", |
| ci_placeholder_rows(phase_rows, "phase_picking", ["P_f1", "S_f1", "mean_f1"]), |
| ci_fields, |
| ) |
| write_csv( |
| OUTPUTS / "grl_dispersion_bootstrap_ci.csv", |
| ci_placeholder_rows(disp_rows, "dispersion_estimation", ["val_mae", "val_rmse", "val_certainty_f1"]), |
| ci_fields, |
| ) |
|
|
| strat_fields = [ |
| "task", |
| "condition", |
| "test_snr_bin", |
| "metric", |
| "value", |
| "n_test_samples", |
| "status", |
| "reason", |
| ] |
| write_csv( |
| OUTPUTS / "grl_phase_snr_stratified_metrics.csv", |
| strat_placeholder_rows("phase_picking", [row["condition"] for row in phase_rows]), |
| strat_fields, |
| ) |
| write_csv( |
| OUTPUTS / "grl_dispersion_snr_stratified_metrics.csv", |
| strat_placeholder_rows("dispersion_estimation", [row["condition"] for row in disp_rows]), |
| strat_fields, |
| ) |
|
|
| sensitivity_dir = OUTPUTS / "grl_sensitivity" |
| sensitivity_dir.mkdir(parents=True, exist_ok=True) |
| (sensitivity_dir / "README.md").write_text( |
| "# GRL Sensitivity Checks\n\n" |
| "The current experiment summaries do not contain per-sample predictions or errors, " |
| "so tolerance, threshold, and test-SNR-bin sensitivity checks cannot be computed " |
| "without re-evaluating the saved checkpoints. This directory is reserved for those " |
| "derived outputs once per-sample evaluation files are generated.\n", |
| encoding="utf-8", |
| ) |
|
|
| write_markdown_summary(phase, disp, phase_rows, disp_rows) |
| write_audit(phase, disp, phase_rows, disp_rows) |
| print(f"Wrote GRL verification outputs under {OUTPUTS}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
