cds2.guided_fit

Guided scientific fitting adds a user-controlled workflow on top of the low-level optimization API.

The workflow:

1. inspects one or more CSV datasets and reports missing values;
2. runs small pilot fits and repeated cross-validation before recommending one model;
3. explains the recommendation with speed, expected fit behaviour and complexity;
4. leaves final model selection to the user;
5. includes measurement uncertainty when a sigma column is supplied;
6. detects suspicious residual outliers, quantifies their estimated RMSE effect and asks before excluding them;
7. fits the selected model, calculates parameter uncertainty and 95% confidence intervals;
8. checks held-out performance with repeated 5-fold cross-validation;
9. cross-checks the fitted result with an independent numerical method;
10. produces Matplotlib fit and residual PNG/PDF figures, a reproducibility manifest and an optional PDF/HTML/Markdown report; long PDF reports paginate automatically instead of truncating results;
11. labels the overall result reliable, caution or unreliable;
12. recommends a different model when the selected fit is weak;
13. recommends dataset-specific models when a single common model is materially weaker;
14. compares manifest reruns with the saved analysis and warns when runtime versions, inputs, confidence intervals, fit parameters, RMSE or the reliability label change materially.

CLI

Interactive use:

cds2 guided-fit experiment.csv --x time --y response --sigma uncertainty

The CLI asks for decisions only when needed: missing-data handling, model selection, outlier exclusion and report format.

Non-interactive/reproducible use:

cds2 guided-fit experiment.csv \
  --x time \
  --y response \
  --sigma uncertainty \
  --model linear \
  --missing interpolate \
  --outliers keep \
  --report pdf \
  --seed 0 \
  --output-dir results

Multiple datasets can be supplied to search for a common model:

cds2 guided-fit run-a.csv run-b.csv --x time --y response

Repeat an analysis from its saved manifest:

cds2 guided-fit-rerun results/guided_fit_manifest.json

The manifest stores the selected model, policies, seed, data hashes, input source metadata and NumPy/SciPy/pandas/Matplotlib/Python versions. Reruns compare those saved results with the new calculation and surface material instability instead of silently replacing the prior result.

What a drift warning looks like

When the runtime environment or numerical result has moved, the CLI makes that visible instead of silently treating the rerun as equivalent:

warning   rerun differs materially from the saved analysis
detail    runtime version changed: scipy <saved-version> -> <current-version>
detail    confidence intervals changed materially for experiment: <relative-change>

The placeholders above are replaced by the versions and measured relative change from the actual rerun. A version drift is reported even when the numerical fit still happens to agree; confidence-interval drift is checked independently from RMSE and parameter drift.

API

Guided, user-controlled scientific model fitting.

The workflow recommends one model, but never silently chooses user-facing decisions such as model selection, missing-data treatment or outlier removal.

rerun_manifest(path, *, allow_outside_run_dir=False, trusted_roots=None)

Repeat a guided fit from a saved manifest.

By default every source_path recorded in the manifest must resolve inside the manifest's own directory (the run directory) or the current working directory, so a foreign manifest cannot make the tool read arbitrary files. Pass allow_outside_run_dir=True to opt out of the confinement, or trusted_roots to extend the whitelist explicitly (the manifest directory itself is always trusted).

Source code in src/cds2/guided_fit.py

def rerun_manifest(
    path: str | Path,
    *,
    allow_outside_run_dir: bool = False,
    trusted_roots: Iterable[str | Path] | None = None,
) -> GuidedFitResult:
    """Repeat a guided fit from a saved manifest.

    By default every ``source_path`` recorded in the manifest must resolve
    inside the manifest's own directory (the run directory) or the current
    working directory, so a foreign manifest cannot make the tool read
    arbitrary files. Pass ``allow_outside_run_dir=True`` to opt out of the
    confinement, or ``trusted_roots`` to extend the whitelist explicitly
    (the manifest directory itself is always trusted).
    """
    manifest_file = Path(path).resolve()
    run_dir = manifest_file.parent
    roots: tuple[Path, ...]
    if trusted_roots is None:
        roots = (run_dir, Path.cwd())
    else:
        roots = (run_dir, *(Path(root).resolve() for root in trusted_roots))
    payload = json.loads(manifest_file.read_text(encoding="utf-8"))
    cfg = payload["result"]
    input_items = cast(list[dict[str, object]], payload["inputs"])
    datasets: list[FitDataset] = []
    for input_item in input_items:
        source_path = cast(str | None, input_item.get("source_path"))
        x_column = cast(str | None, input_item.get("x_column"))
        y_column = cast(str | None, input_item.get("y_column"))
        sigma_column = cast(str | None, input_item.get("sigma_column"))
        if not source_path or not x_column or not y_column:
            raise ValueError("manifest does not contain reusable CSV source metadata")
        source = _resolve_rerun_source(
            source_path, run_dir, roots, allow_outside_run_dir=allow_outside_run_dir
        )
        loaded = load_csv_dataset(source, x_column, y_column, sigma_column)
        datasets.append(replace(loaded, name=str(input_item.get("name") or loaded.name)))

    rerun = run_guided_fit(
        tuple(datasets),
        cfg["model"],
        missing_policy=cfg["missing_policy"],
        outlier_policy=cfg["outlier_policy"],
        seed=int(cfg["seed"]),
    )

    details: list[str] = []
    saved_versions = cast(dict[str, str], cfg.get("package_versions", {}))
    for package, current_version in rerun.package_versions.items():
        saved_version = saved_versions.get(package)
        if saved_version is None:
            details.append(
                f"runtime version was not recorded previously: {package}={current_version}"
            )
        elif saved_version != current_version:
            details.append(
                f"runtime version changed: {package} {saved_version} -> {current_version}"
            )
    for package, saved_version in saved_versions.items():
        if package not in rerun.package_versions:
            details.append(f"runtime package no longer reported: {package} (saved {saved_version})")

    previous_hashes = cast(
        dict[str, str],
        cfg.get("rerun_data_hashes", cfg["data_hashes"]),
    )
    runtime_keys = _dataset_keys(tuple(datasets))
    saved_keys = tuple(
        str(input_item.get("dataset_key") or runtime_key)
        for input_item, runtime_key in zip(input_items, runtime_keys, strict=True)
    )
    for runtime_key, saved_key in zip(runtime_keys, saved_keys, strict=True):
        digest = rerun.data_hashes[runtime_key]
        previous_digest = previous_hashes.get(saved_key)
        if previous_digest is None:
            previous_digest = previous_hashes.get(runtime_key)
        if previous_digest != digest:
            details.append(f"input data changed: {saved_key}")

    saved_results = cast(list[dict[str, object]], cfg["datasets"])
    if len(saved_results) != len(rerun.datasets):
        details.append(f"dataset count changed: {len(saved_results)} -> {len(rerun.datasets)}")
    for index, dataset_result in enumerate(rerun.datasets):
        if index >= len(saved_results):
            break
        previous = saved_results[index]
        key = saved_keys[index] if index < len(saved_keys) else runtime_keys[index]
        old_rmse = float(cast(float, previous["rmse"]))
        rmse_change = abs(dataset_result.rmse - old_rmse) / max(abs(old_rmse), 1e-12)

        old_ci = np.asarray(
            cast(list[list[float]], previous["confidence_95"]),
            dtype=np.float64,
        )
        if old_ci.shape != dataset_result.confidence_95.shape:
            details.append(f"confidence interval shape changed for {key}")
        else:
            new_ci = dataset_result.confidence_95
            ci_delta = new_ci - old_ci
            ci_scale = max(float(np.linalg.norm(old_ci)), 1e-12)
            ci_change = float(np.linalg.norm(ci_delta)) / ci_scale
            if ci_change > 0.05:
                max_bound_shift = float(np.max(np.abs(ci_delta)))
                saved_text = np.array2string(
                    old_ci,
                    precision=8,
                    separator=", ",
                    suppress_small=False,
                )
                rerun_text = np.array2string(
                    new_ci,
                    precision=8,
                    separator=", ",
                    suppress_small=False,
                )
                details.append(
                    f"confidence intervals changed materially for {key}: "
                    f"relative={ci_change:.1%}; max_bound_shift={max_bound_shift:.8g}; "
                    f"saved={saved_text}; rerun={rerun_text}"
                )

        old_params = np.asarray(cast(list[float], previous["params"]), dtype=np.float64)
        if old_params.shape != dataset_result.params.shape:
            details.append(f"parameter shape changed for {key}")
            continue
        param_scale = max(float(np.linalg.norm(old_params)), 1e-12)
        param_change = float(np.linalg.norm(dataset_result.params - old_params)) / param_scale
        if max(rmse_change, param_change) > 0.05:
            details.append(
                f"fit changed materially for {key}: "
                f"rmse={rmse_change:.1%}, parameters={param_change:.1%}"
            )

    if cast(str, cfg["trust"]) != rerun.trust:
        details.append(f"reliability label changed: {cfg['trust']} -> {rerun.trust}")

    return replace(
        rerun,
        stability_warning=bool(details),
        stability_details=tuple(details),
    )