| """Isolated pipeline for analyses explicitly requested by the reviewers."""
|
|
|
| from __future__ import annotations
|
|
|
| import hashlib
|
| import importlib.metadata
|
| import json
|
| import platform
|
| from pathlib import Path
|
| import sys
|
| from typing import Any, Dict, Mapping, MutableMapping, Sequence
|
|
|
| import numpy as np
|
| import pandas as pd
|
| from rdkit import Chem
|
| from sklearn.ensemble import ExtraTreesRegressor
|
| from sklearn.linear_model import Ridge
|
| from sklearn.pipeline import make_pipeline
|
| from sklearn.preprocessing import OneHotEncoder, StandardScaler
|
|
|
| from src.data import MolecularFeatureExtractor
|
|
|
| from revision.scripts.reanalysis_core import (
|
| annotate_structures,
|
| compute_regression_metrics,
|
| ensure_new_output_dir,
|
| make_grouped_folds,
|
| make_grouped_holdout,
|
| paired_group_bootstrap,
|
| per_lab_metrics,
|
| )
|
|
|
|
|
| PRIMARY_CLASSICAL_MODEL = "fingerprint_plus_descriptors_et"
|
| ALLOWED_GROUP_COLUMNS = {
|
| "structure_group",
|
| "scaffold_group",
|
| "scaffold_component_group",
|
| }
|
|
|
|
|
| def _json_default(value: Any) -> Any:
|
| if isinstance(value, (np.integer,)):
|
| return int(value)
|
| if isinstance(value, (np.floating,)):
|
| return float(value)
|
| if isinstance(value, np.ndarray):
|
| return value.tolist()
|
| if isinstance(value, Path):
|
| return str(value)
|
| raise TypeError(f"Cannot serialize {type(value).__name__}")
|
|
|
|
|
| def _write_json(path: Path, payload: Mapping[str, Any]) -> None:
|
| path.write_text(
|
| json.dumps(payload, indent=2, ensure_ascii=False, default=_json_default) + "\n",
|
| encoding="utf-8",
|
| )
|
|
|
|
|
| def _resolve_path(path_value: str | Path, project_root: Path) -> Path:
|
| path = Path(path_value)
|
| return (project_root / path).resolve() if not path.is_absolute() else path.resolve()
|
|
|
|
|
| def _portable_path(path_value: str | Path, project_root: Path) -> str:
|
| """Represent a path without recording a machine-specific absolute path."""
|
|
|
| path = Path(path_value).resolve()
|
| root = project_root.resolve()
|
| try:
|
| return path.relative_to(root).as_posix()
|
| except ValueError:
|
|
|
|
|
| return (Path("external") / path.name).as_posix()
|
|
|
|
|
| def _portable_config(config: Mapping[str, Any], project_root: Path) -> Dict[str, Any]:
|
| portable = dict(config)
|
| for key in ("input_csv", "output_root"):
|
| if key in portable:
|
| portable[key] = _portable_path(portable[key], project_root)
|
| return portable
|
|
|
|
|
| def _package_version(distribution_name: str) -> str:
|
| try:
|
| return importlib.metadata.version(distribution_name)
|
| except importlib.metadata.PackageNotFoundError:
|
| return "not installed"
|
|
|
|
|
| def _environment_record() -> Dict[str, Any]:
|
| """Return reproducibility-critical versions without host-specific paths."""
|
|
|
| return {
|
| "python": {
|
| "version": platform.python_version(),
|
| "implementation": platform.python_implementation(),
|
| },
|
| "platform": {
|
| "system": platform.system(),
|
| "release": platform.release(),
|
| "machine": platform.machine(),
|
| },
|
| "packages": {
|
| name: _package_version(name)
|
| for name in (
|
| "joblib",
|
| "numpy",
|
| "pandas",
|
| "rdkit",
|
| "scikit-learn",
|
| "scipy",
|
| "torch",
|
| "torch-geometric",
|
| )
|
| },
|
| "python_major_minor": f"{sys.version_info.major}.{sys.version_info.minor}",
|
| }
|
|
|
|
|
| def _write_sha256_manifest(output_root: Path) -> None:
|
| """Hash every completed artifact using output-root-relative paths."""
|
|
|
| manifest = output_root / "SHA256SUMS.txt"
|
| lines = []
|
| for artifact in sorted(path for path in output_root.rglob("*") if path.is_file() and path != manifest):
|
| digest = hashlib.sha256()
|
| with artifact.open("rb") as handle:
|
| for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
| digest.update(chunk)
|
| lines.append(f"{digest.hexdigest()} {artifact.relative_to(output_root).as_posix()}")
|
| manifest.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
| def validate_config(config: Mapping[str, Any], project_root: Path) -> Dict[str, Any]:
|
| """Validate predeclared design choices before any output is created."""
|
|
|
| required = ("input_csv", "output_root", "outer_seeds", "split_strategies")
|
| missing = [key for key in required if key not in config]
|
| if missing:
|
| raise ValueError(f"Configuration is missing required keys: {missing}")
|
|
|
| normalized = dict(config)
|
| input_csv = _resolve_path(config["input_csv"], project_root)
|
| output_root = _resolve_path(config["output_root"], project_root)
|
| original_model_dir = (project_root / "hybrid_oof_models").resolve()
|
| if output_root == original_model_dir or original_model_dir in output_root.parents:
|
| raise ValueError("output_root must not be hybrid_oof_models or one of its descendants.")
|
| if not input_csv.is_file():
|
| raise FileNotFoundError(f"Input CSV not found: {input_csv}")
|
|
|
| seeds = [int(seed) for seed in config["outer_seeds"]]
|
| if not seeds or len(seeds) != len(set(seeds)):
|
| raise ValueError("outer_seeds must be a nonempty list of unique predeclared seeds.")
|
| raw_strategies = dict(config["split_strategies"])
|
| if not raw_strategies:
|
| raise ValueError("At least one split strategy is required.")
|
| strategy_specs: Dict[str, Dict[str, Any]] = {}
|
| for strategy_name, raw_spec in raw_strategies.items():
|
| if isinstance(raw_spec, str):
|
| spec = {
|
| "group_column": raw_spec,
|
| "outer_folds": int(config.get("outer_folds", 10)),
|
| "balance_group_sizes": False,
|
| }
|
| elif isinstance(raw_spec, Mapping):
|
| spec = {
|
| "group_column": str(raw_spec["group_column"]),
|
| "outer_folds": int(raw_spec.get("outer_folds", config.get("outer_folds", 10))),
|
| "balance_group_sizes": bool(raw_spec.get("balance_group_sizes", False)),
|
| }
|
| else:
|
| raise TypeError(f"Invalid split strategy specification for {strategy_name!r}.")
|
| if spec["outer_folds"] < 2:
|
| raise ValueError(f"outer_folds must be at least 2 for {strategy_name!r}.")
|
| strategy_specs[str(strategy_name)] = spec
|
|
|
| unknown_groups = {
|
| spec["group_column"] for spec in strategy_specs.values()
|
| } - ALLOWED_GROUP_COLUMNS
|
| if unknown_groups:
|
| raise ValueError(f"Unsupported split group columns: {sorted(unknown_groups)}")
|
|
|
| normalized["input_csv"] = str(input_csv)
|
| normalized["output_root"] = str(output_root)
|
| normalized["outer_seeds"] = seeds
|
| normalized["split_strategies"] = strategy_specs
|
| normalized.setdefault("outer_folds", 10)
|
| normalized.setdefault("inner_folds", 6)
|
| normalized.setdefault("analysis_id", output_root.name)
|
| normalized.setdefault("descriptor_features", ["LogP", "BertzCT", "MolMR", "NumAromaticRings", "HeavyAtomCount"])
|
| normalized.setdefault("fingerprint", {"radius": 2, "n_bits": 2048, "use_chirality": True})
|
| normalized.setdefault(
|
| "classical_baselines",
|
| {"n_estimators": 500, "max_depth": None, "min_samples_leaf": 2, "max_features": 1.0},
|
| )
|
| normalized.setdefault("bootstrap", {"n_resamples": 2000, "confidence": 0.95})
|
| return normalized
|
|
|
|
|
| def _fit_lab_encoder(train_labels: np.ndarray, test_labels: np.ndarray) -> tuple[np.ndarray, np.ndarray, OneHotEncoder]:
|
| encoder = OneHotEncoder(handle_unknown="ignore", sparse_output=False, dtype=np.float32)
|
| train_one_hot = encoder.fit_transform(train_labels.reshape(-1, 1))
|
| test_one_hot = encoder.transform(test_labels.reshape(-1, 1))
|
| return train_one_hot, test_one_hot, encoder
|
|
|
|
|
| def run_classical_baselines(
|
| *,
|
| fingerprints: np.ndarray,
|
| descriptors: np.ndarray,
|
| lab_labels: Sequence[object],
|
| targets: Sequence[float],
|
| train_indices: Sequence[int],
|
| test_indices: Sequence[int],
|
| seed: int,
|
| estimator_config: Mapping[str, Any],
|
| ) -> Dict[str, Any]:
|
| """Fit fixed, untuned classical baselines and one descriptor ablation.
|
|
|
| The primary model is predeclared as Morgan fingerprint + fixed descriptors
|
| + one-hot laboratory ExtraTrees. Test performance is never used to select
|
| among these models.
|
| """
|
|
|
| fingerprints_array = np.asarray(fingerprints, dtype=np.float32)
|
| descriptors_array = np.asarray(descriptors, dtype=np.float32)
|
| labs = np.asarray(lab_labels).astype(str)
|
| y = np.asarray(targets, dtype=float)
|
| train = np.asarray(train_indices, dtype=int)
|
| test = np.asarray(test_indices, dtype=int)
|
|
|
| lab_train, lab_test, lab_encoder = _fit_lab_encoder(labs[train], labs[test])
|
| descriptor_train = np.column_stack([descriptors_array[train], lab_train])
|
| descriptor_test = np.column_stack([descriptors_array[test], lab_test])
|
| fingerprint_train = np.column_stack([fingerprints_array[train], lab_train])
|
| fingerprint_test = np.column_stack([fingerprints_array[test], lab_test])
|
| combined_train = np.column_stack([fingerprints_array[train], descriptors_array[train], lab_train])
|
| combined_test = np.column_stack([fingerprints_array[test], descriptors_array[test], lab_test])
|
|
|
| tree_parameters = {
|
| "n_estimators": int(estimator_config.get("n_estimators", 500)),
|
| "max_depth": estimator_config.get("max_depth"),
|
| "min_samples_leaf": int(estimator_config.get("min_samples_leaf", 2)),
|
| "max_features": estimator_config.get("max_features", 1.0),
|
| "random_state": int(seed),
|
| "n_jobs": int(estimator_config.get("n_jobs", -1)),
|
| }
|
| models: MutableMapping[str, Any] = {
|
| "descriptor_only_ridge": make_pipeline(StandardScaler(), Ridge(alpha=1.0)),
|
| "descriptor_only_et": ExtraTreesRegressor(**tree_parameters),
|
| "fingerprint_no_lab_et": ExtraTreesRegressor(**tree_parameters),
|
| "fingerprint_only_et": ExtraTreesRegressor(**tree_parameters),
|
| "fingerprint_plus_descriptors_et": ExtraTreesRegressor(**tree_parameters),
|
| }
|
| feature_pairs = {
|
| "descriptor_only_ridge": (descriptor_train, descriptor_test),
|
| "descriptor_only_et": (descriptor_train, descriptor_test),
|
| "fingerprint_no_lab_et": (fingerprints_array[train], fingerprints_array[test]),
|
| "fingerprint_only_et": (fingerprint_train, fingerprint_test),
|
| "fingerprint_plus_descriptors_et": (combined_train, combined_test),
|
| }
|
| predictions: Dict[str, np.ndarray] = {}
|
| for name, model in models.items():
|
| train_features, test_features = feature_pairs[name]
|
| model.fit(train_features, y[train])
|
| predictions[name] = np.asarray(model.predict(test_features), dtype=float)
|
|
|
| no_lab_model = models["fingerprint_no_lab_et"]
|
| no_lab_train_predictions = np.asarray(
|
| no_lab_model.predict(fingerprints_array[train]),
|
| dtype=float,
|
| )
|
| affine_by_lab: Dict[str, tuple[float, float]] = {}
|
| global_offset = float(np.mean(y[train] - no_lab_train_predictions))
|
| for lab in sorted(np.unique(labs[train])):
|
| lab_mask = labs[train] == lab
|
| lab_y = y[train][lab_mask]
|
| lab_base = no_lab_train_predictions[lab_mask]
|
| if lab_mask.sum() >= 3 and np.ptp(lab_base) > 1e-8:
|
| slope, intercept = np.polyfit(lab_base, lab_y, 1)
|
| affine_by_lab[lab] = (float(slope), float(intercept))
|
| else:
|
| affine_by_lab[lab] = (1.0, float(np.mean(lab_y - lab_base)))
|
| predictions["fingerprint_no_lab_plus_lab_affine"] = np.asarray(
|
| [
|
| affine_by_lab.get(lab, (1.0, global_offset))[0] * base_prediction
|
| + affine_by_lab.get(lab, (1.0, global_offset))[1]
|
| for lab, base_prediction in zip(
|
| labs[test],
|
| predictions["fingerprint_no_lab_et"],
|
| )
|
| ],
|
| dtype=float,
|
| )
|
|
|
| global_median = float(np.median(y[train]))
|
| laboratory_medians = pd.Series(y[train]).groupby(labs[train]).median().to_dict()
|
| predictions["lab_median"] = np.asarray(
|
| [laboratory_medians.get(lab, global_median) for lab in labs[test]],
|
| dtype=float,
|
| )
|
| metrics = {name: compute_regression_metrics(y[test], values) for name, values in predictions.items()}
|
| return {
|
| "primary_model": PRIMARY_CLASSICAL_MODEL,
|
| "predictions": predictions,
|
| "metrics": metrics,
|
| "models": dict(models),
|
| "lab_encoder": lab_encoder,
|
| "lab_affine_parameters": affine_by_lab,
|
| "feature_dimensions": {
|
| "descriptor_only": int(descriptor_train.shape[1]),
|
| "fingerprint_only": int(fingerprint_train.shape[1]),
|
| "fingerprint_plus_descriptors": int(combined_train.shape[1]),
|
| },
|
| }
|
|
|
|
|
| def _extract_features(
|
| annotated: pd.DataFrame,
|
| descriptor_features: Sequence[str],
|
| fingerprint_config: Mapping[str, Any],
|
| ) -> tuple[np.ndarray, np.ndarray]:
|
| extractor = MolecularFeatureExtractor()
|
| descriptor_rows = []
|
| fingerprints = []
|
| for smiles in annotated["SMILES"].astype(str):
|
| descriptor_record = extractor.get_molecular_descriptors(smiles)
|
| missing = [name for name in descriptor_features if name not in descriptor_record]
|
| if missing:
|
| raise ValueError(f"Descriptor extraction did not produce: {missing}")
|
| descriptor_rows.append([float(descriptor_record[name]) for name in descriptor_features])
|
| fingerprints.append(
|
| extractor.get_morgan_fingerprint(
|
| smiles,
|
| n_bits=int(fingerprint_config.get("n_bits", 2048)),
|
| radius=int(fingerprint_config.get("radius", 2)),
|
| use_chirality=bool(fingerprint_config.get("use_chirality", True)),
|
| )
|
| )
|
| return np.asarray(descriptor_rows, dtype=np.float32), np.asarray(fingerprints, dtype=np.float32)
|
|
|
|
|
| def _dataset_audit(annotated: pd.DataFrame) -> Dict[str, Any]:
|
| labs_per_structure = annotated.groupby("structure_group")["Lab"].nunique()
|
| canonical_lab_rt_counts = annotated.groupby(["structure_group", "Lab"])["RT"].nunique()
|
| return {
|
| "n_rows": int(len(annotated)),
|
| "columns": [str(column) for column in annotated.columns],
|
| "n_laboratories": int(annotated["Lab"].nunique()),
|
| "laboratory_counts": annotated["Lab"].value_counts().sort_index().astype(int).to_dict(),
|
| "unique_smiles_as_supplied": int(annotated["SMILES"].nunique()),
|
| "unique_canonical_isomeric_smiles": int(annotated["canonical_smiles_isomeric"].nunique()),
|
| "unique_canonical_nonisomeric_smiles": int(annotated["canonical_smiles_nonisomeric"].nunique()),
|
| "unique_full_inchi_keys": int(annotated["inchi_key_full"].nunique()),
|
| "unique_connectivity_blocks": int(annotated["inchi_key_connectivity"].nunique()),
|
| "unique_raw_murcko_scaffold_groups": int(annotated["scaffold_group"].nunique()),
|
| "unique_identity_safe_scaffold_components": int(
|
| annotated["scaffold_component_group"].nunique()
|
| ),
|
| "graph_conversion_failures_in_provided_csv": 0,
|
| "upstream_curation_removals": "not inferable without the pre-curation export",
|
| "identity_policy": {
|
| "outer_group_key": "InChIKey connectivity block",
|
| "fallback": "canonical non-isomeric SMILES when InChI generation fails",
|
| "salt_handling": "no fragment removal or parent selection; fragment_count is recorded",
|
| "stereochemistry_handling": "full identity is recorded; connectivity grouping conservatively co-groups stereoisomers",
|
| "tautomer_handling": "no explicit tautomer canonicalizer; InChI connectivity grouping supplies the conservative boundary",
|
| "protonation_handling": "no neutralization; supplied formal charge and full structure remain recorded",
|
| },
|
| "fragment_count_distribution": annotated["fragment_count"].value_counts().sort_index().astype(int).to_dict(),
|
| "formal_charge_distribution": annotated["formal_charge"].value_counts().sort_index().astype(int).to_dict(),
|
| "rows_with_explicit_stereo": int(annotated["has_explicit_stereo"].sum()),
|
| "labs_per_structure_distribution": labs_per_structure.value_counts().sort_index().astype(int).to_dict(),
|
| "structures_seen_in_multiple_labs": int((labs_per_structure > 1).sum()),
|
| "rows_in_multi_lab_structures": int(annotated["structure_group"].isin(labs_per_structure[labs_per_structure > 1].index).sum()),
|
| "exact_duplicate_extra_rows": int(annotated.duplicated(["SMILES", "Lab", "RT"]).sum()),
|
| "structure_lab_duplicate_extra_rows": int(annotated.duplicated(["structure_group", "Lab"]).sum()),
|
| "structure_lab_groups_with_conflicting_rt": int((canonical_lab_rt_counts > 1).sum()),
|
| }
|
|
|
|
|
| def _write_shared_lab_matrix(annotated: pd.DataFrame, output_path: Path) -> None:
|
| labs = sorted(annotated["Lab"].astype(str).unique())
|
| matrix = pd.DataFrame(0, index=labs, columns=labs, dtype=int)
|
| for _, group in annotated.groupby("structure_group"):
|
| represented = sorted(group["Lab"].astype(str).unique())
|
| for left in represented:
|
| for right in represented:
|
| matrix.loc[left, right] += 1
|
| matrix.index.name = "Lab"
|
| matrix.to_csv(output_path)
|
|
|
|
|
| def _split_summary(
|
| annotated: pd.DataFrame,
|
| train_indices: np.ndarray,
|
| test_indices: np.ndarray,
|
| group_column: str,
|
| ) -> Dict[str, Any]:
|
| train_groups = set(annotated.loc[train_indices, group_column])
|
| test_groups = set(annotated.loc[test_indices, group_column])
|
| train_structures = set(annotated.loc[train_indices, "structure_group"])
|
| test_structures = set(annotated.loc[test_indices, "structure_group"])
|
| return {
|
| "n_development_rows": int(len(train_indices)),
|
| "n_test_rows": int(len(test_indices)),
|
| "test_fraction": float(len(test_indices) / len(annotated)),
|
| "group_column": group_column,
|
| "hard_group_overlap": int(len(train_groups & test_groups)),
|
| "structure_identity_overlap": int(len(train_structures & test_structures)),
|
| "development_laboratories": int(annotated.loc[train_indices, "Lab"].nunique()),
|
| "test_laboratories": int(annotated.loc[test_indices, "Lab"].nunique()),
|
| }
|
|
|
|
|
| def _write_split_outputs(
|
| *,
|
| split_dir: Path,
|
| annotated: pd.DataFrame,
|
| train_indices: np.ndarray,
|
| test_indices: np.ndarray,
|
| inner_folds: Sequence[tuple[np.ndarray, np.ndarray]],
|
| strategy: str,
|
| group_column: str,
|
| seed: int,
|
| ) -> None:
|
| assignments = annotated.copy()
|
| assignments["outer_split"] = "development"
|
| assignments.loc[test_indices, "outer_split"] = "test"
|
| assignments["inner_validation_fold"] = -1
|
| for fold_index, (_, validation_indices) in enumerate(inner_folds):
|
| assignments.loc[validation_indices, "inner_validation_fold"] = fold_index
|
| assignments["split_strategy"] = strategy
|
| assignments["outer_seed"] = int(seed)
|
| assignments.to_csv(split_dir / "split_assignments.csv", index=False)
|
| np.savez_compressed(
|
| split_dir / "split_indices.npz",
|
| development_indices=np.asarray(train_indices, dtype=np.int32),
|
| test_indices=np.asarray(test_indices, dtype=np.int32),
|
| inner_train_indices=np.asarray([pair[0] for pair in inner_folds], dtype=object),
|
| inner_validation_indices=np.asarray([pair[1] for pair in inner_folds], dtype=object),
|
| )
|
| _write_json(
|
| split_dir / "split_summary.json",
|
| _split_summary(annotated, train_indices, test_indices, group_column),
|
| )
|
|
|
|
|
| def _write_classical_outputs(
|
| *,
|
| split_dir: Path,
|
| annotated: pd.DataFrame,
|
| test_indices: np.ndarray,
|
| result: Mapping[str, Any],
|
| bootstrap_config: Mapping[str, Any],
|
| seed: int,
|
| ) -> None:
|
| y_true = annotated.loc[test_indices, "RT"].to_numpy(dtype=float)
|
| labs = annotated.loc[test_indices, "Lab"].astype(str).to_numpy()
|
| groups = annotated.loc[test_indices, "structure_group"].astype(str).to_numpy()
|
| predictions = result["predictions"]
|
|
|
| prediction_frame = annotated.loc[
|
| test_indices,
|
| [
|
| "record_index",
|
| "SMILES",
|
| "Lab",
|
| "RT",
|
| "structure_group",
|
| "scaffold_group",
|
| "scaffold_component_group",
|
| ],
|
| ].copy()
|
| for model_name, values in predictions.items():
|
| prediction_frame[f"prediction_{model_name}"] = values
|
| prediction_frame.to_csv(split_dir / "classical_predictions.csv", index=False)
|
|
|
| metrics_payload = {
|
| "primary_model_predeclared": result["primary_model"],
|
| "metrics": result["metrics"],
|
| "feature_dimensions": result["feature_dimensions"],
|
| "selection_rule": "No outer-test model selection; all fixed models are reported.",
|
| }
|
| _write_json(split_dir / "classical_metrics.json", metrics_payload)
|
|
|
| development_ranges = (
|
| annotated.loc[annotated.index.difference(test_indices)]
|
| .groupby("Lab")["RT"]
|
| .agg(lambda values: float(values.max() - values.min()))
|
| .to_dict()
|
| )
|
| per_lab_tables = []
|
| for model_name, values in predictions.items():
|
| table = per_lab_metrics(
|
| y_true,
|
| values,
|
| labs,
|
| normalization_ranges=development_ranges,
|
| )
|
| table.insert(0, "model", model_name)
|
| per_lab_tables.append(table)
|
| pd.concat(per_lab_tables, ignore_index=True).to_csv(split_dir / "classical_per_lab_metrics.csv", index=False)
|
|
|
| primary_predictions = predictions[result["primary_model"]]
|
| paired = {}
|
| for reference_name, reference_predictions in predictions.items():
|
| if reference_name == result["primary_model"]:
|
| continue
|
| paired[reference_name] = paired_group_bootstrap(
|
| y_true=y_true,
|
| candidate=primary_predictions,
|
| reference=reference_predictions,
|
| groups=groups,
|
| n_resamples=int(bootstrap_config.get("n_resamples", 2000)),
|
| confidence=float(bootstrap_config.get("confidence", 0.95)),
|
| seed=int(seed),
|
| )
|
| _write_json(split_dir / "classical_paired_group_bootstrap.json", paired)
|
|
|
|
|
| def run_reanalysis(
|
| config: Mapping[str, Any],
|
| *,
|
| project_root: Path,
|
| include_neural: bool = False,
|
| smoke_only: bool = False,
|
| ) -> Path:
|
| """Run audit, fixed splits, classical baselines, and optional neural stack."""
|
|
|
| project_root = project_root.resolve()
|
| normalized = validate_config(config, project_root)
|
| output_root = ensure_new_output_dir(normalized["output_root"])
|
| data = pd.read_csv(normalized["input_csv"])
|
| if smoke_only and len(data) > 240:
|
| data = data.groupby("Lab", group_keys=False).head(8).reset_index(drop=True)
|
| annotated = annotate_structures(data)
|
|
|
| input_hash = hashlib.sha256(Path(normalized["input_csv"]).read_bytes()).hexdigest()
|
| run_metadata = {
|
| "analysis_id": normalized["analysis_id"],
|
| "input_csv": _portable_path(normalized["input_csv"], project_root),
|
| "input_sha256": input_hash,
|
| "output_root": _portable_path(output_root, project_root),
|
| "outer_seeds_predeclared": normalized["outer_seeds"] if not smoke_only else normalized["outer_seeds"][:1],
|
| "outer_fold_selection": (
|
| "Canonical grouping uses the first shuffled StratifiedGroupKFold fold. "
|
| "Size-balanced strategies choose the fold with maximal laboratory coverage "
|
| "and closest row count to 1/k; RT values and model results are never used."
|
| ),
|
| "smoke_only": bool(smoke_only),
|
| "include_neural": bool(include_neural),
|
| "original_outputs_modified": False,
|
| }
|
| _write_json(output_root / "RUN_METADATA.json", run_metadata)
|
| _write_json(output_root / "FROZEN_CONFIG.json", _portable_config(normalized, project_root))
|
| _write_json(output_root / "ENVIRONMENT.json", _environment_record())
|
| _write_json(output_root / "dataset_audit.json", _dataset_audit(annotated))
|
| _write_shared_lab_matrix(annotated, output_root / "shared_compounds_by_lab.csv")
|
| annotated.to_csv(output_root / "record_identity_manifest.csv", index=False)
|
|
|
| descriptor_matrix, fingerprint_matrix = _extract_features(
|
| annotated,
|
| normalized["descriptor_features"],
|
| normalized["fingerprint"],
|
| )
|
| seeds = normalized["outer_seeds"][:1] if smoke_only else normalized["outer_seeds"]
|
| inner_fold_count = 2 if smoke_only else int(normalized["inner_folds"])
|
| estimator_config = dict(normalized["classical_baselines"])
|
| if smoke_only:
|
| estimator_config["n_estimators"] = min(8, int(estimator_config.get("n_estimators", 8)))
|
| bootstrap_config = dict(normalized["bootstrap"])
|
| if smoke_only:
|
| bootstrap_config["n_resamples"] = min(50, int(bootstrap_config.get("n_resamples", 50)))
|
|
|
| for strategy, strategy_spec in normalized["split_strategies"].items():
|
| group_column = strategy_spec["group_column"]
|
| outer_folds = 2 if smoke_only else int(strategy_spec["outer_folds"])
|
| balance_group_sizes = bool(strategy_spec.get("balance_group_sizes", False))
|
| for seed in seeds:
|
| split_dir = output_root / strategy / f"seed_{seed}"
|
| split_dir.mkdir(parents=True, exist_ok=False)
|
| train_indices, test_indices = make_grouped_holdout(
|
| annotated,
|
| group_column=group_column,
|
| seed=seed,
|
| n_splits=outer_folds,
|
| balance_group_sizes=balance_group_sizes,
|
| )
|
| inner_folds = make_grouped_folds(
|
| annotated,
|
| train_indices,
|
| group_column=group_column,
|
| seed=seed,
|
| n_splits=inner_fold_count,
|
| )
|
| _write_split_outputs(
|
| split_dir=split_dir,
|
| annotated=annotated,
|
| train_indices=train_indices,
|
| test_indices=test_indices,
|
| inner_folds=inner_folds,
|
| strategy=strategy,
|
| group_column=group_column,
|
| seed=seed,
|
| )
|
| classical_result = run_classical_baselines(
|
| fingerprints=fingerprint_matrix,
|
| descriptors=descriptor_matrix,
|
| lab_labels=annotated["Lab"].to_numpy(),
|
| targets=annotated["RT"].to_numpy(),
|
| train_indices=train_indices,
|
| test_indices=test_indices,
|
| seed=seed,
|
| estimator_config=estimator_config,
|
| )
|
| _write_classical_outputs(
|
| split_dir=split_dir,
|
| annotated=annotated,
|
| test_indices=test_indices,
|
| result=classical_result,
|
| bootstrap_config=bootstrap_config,
|
| seed=seed,
|
| )
|
|
|
| if include_neural:
|
| from revision.scripts.reanalysis_neural import run_neural_stack
|
|
|
| run_neural_stack(
|
| normalized,
|
| annotated=annotated,
|
| descriptor_matrix=descriptor_matrix,
|
| fingerprint_matrix=fingerprint_matrix,
|
| train_indices=train_indices,
|
| test_indices=test_indices,
|
| inner_folds=inner_folds,
|
| output_dir=split_dir / "neural_stack",
|
| seed=seed,
|
| smoke_only=smoke_only,
|
| )
|
| _write_sha256_manifest(output_root)
|
| return output_root
|
|
|
