verify_external_exp2_multi.py

from pathlib import Path
# -*- coding: utf-8 -*-
"""Exp-2 repeat: surrogate R2 vs n with 3 subset draws for stability."""
import os, warnings
import numpy as np
import pandas as pd
warnings.filterwarnings("ignore")
os.environ.setdefault("LOKY_MAX_CPU_COUNT", "1")
from sklearn.model_selection import KFold, LeaveOneOut
from sklearn.metrics import r2_score
from smallmatprep.modeling.uwe import make_default_models, uwe_fit_predict

RAW = str(Path(__file__).resolve().parent / "data" / "conductivity_5035.csv")
RESULTS = str(Path(__file__).resolve().parent / "results")
COMP = ["PC", "EC", "EMC", "LiPF_6"]

def load_a():
    df = pd.read_csv(RAW, sep=";").iloc[2:].copy()
    for c in ["temperature", "PC", "EC", "EMC", "LiPF_6", "EIS_conductivity"]:
        df[c] = pd.to_numeric(df[c], errors="coerce")
    df = df.dropna(subset=["temperature", "PC", "EC", "EMC", "LiPF_6", "EIS_conductivity"]).reset_index(drop=True)
    d = df[df["temperature"] == 20.0].reset_index(drop=True)
    X = d[COMP].values.astype(float)
    y = np.log(d["EIS_conductivity"].values.astype(float))
    return X, y

def cv_uwe(X, y, cv, seed=0):
    models = make_default_models(seed)
    preds, truths = [], []
    for tr, va in cv.split(X):
        uw = (len(X[tr]) <= 500)
        mu, _, _, _ = uwe_fit_predict(models, X[tr], y[tr], X[va], update_weights=uw)
        preds.append(mu); truths.append(y[va])
    preds = np.concatenate(preds); truths = np.concatenate(truths)
    return r2_score(truths, preds), float(np.sqrt(np.mean((truths - preds) ** 2)))

X, y = load_a()
ns = [47, 100, 200, 504]
n_rep = 3 if False else 3
rows = []
for n in ns:
    r2s, rmses = [], []
    for rep in range(n_rep):
        if n < len(y):
            rng = np.random.default_rng(1000 + rep)
            sub = rng.choice(len(y), size=n, replace=False)
            Xs, ys = X[sub], y[sub]
        else:
            Xs, ys = X, y
        cv = LeaveOneOut() if n <= 47 else KFold(n_splits=10, shuffle=True, random_state=42)
        r2u, rmse_u = cv_uwe(Xs, ys, cv)
        r2s.append(r2u); rmses.append(rmse_u)
        print(f"  n={n} rep{rep}: r2={r2u:.3f} rmse={rmse_u:.4f}")
    rows.append({"n": n, "r2_mean": float(np.mean(r2s)), "r2_std": float(np.std(r2s)),
                 "rmse_mean": float(np.mean(rmses))})
    print(f"  -> n={n}: r2 = {np.mean(r2s):.3f} ± {np.std(r2s):.3f}")
res = pd.DataFrame(rows)
res.to_csv(os.path.join(RESULTS, "external_exp2_ncurve.csv"), index=False, encoding="utf-8-sig")
print("\nsaved external_exp2_ncurve.csv")
print(res.round(4).to_string(index=False))