File size: 5,099 Bytes
6b3618b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """Training utilities: data loading, training loops, metric computation."""
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.datasets import make_moons, make_circles, load_iris, load_wine, load_breast_cancer
def load_synthetic_dataset(name="moons", n_samples=1000, noise=0.2, seed=42):
"""Load synthetic 2D dataset: 'moons', 'circles', or 'xor'."""
rng = np.random.RandomState(seed)
if name == "moons":
return make_moons(n_samples=n_samples, noise=noise, random_state=seed)
elif name == "circles":
return make_circles(n_samples=n_samples, noise=noise, factor=0.5, random_state=seed)
elif name == "xor":
X = rng.randn(n_samples, 2)
y = ((X[:, 0] > 0) ^ (X[:, 1] > 0)).astype(int)
X += noise * rng.randn(n_samples, 2)
return X, y
raise ValueError(f"Unknown dataset: {name}")
def load_uci_dataset(name="iris"):
"""Load UCI dataset. Returns (X, y, task_type)."""
loaders = {
"iris": (load_iris, "classification"),
"wine": (load_wine, "classification"),
"breast_cancer": (load_breast_cancer, "classification"),
}
if name == "california_housing":
from sklearn.datasets import fetch_california_housing
d = fetch_california_housing()
return d.data, d.target, "regression"
if name not in loaders:
raise ValueError(f"Unknown dataset: {name}")
d = loaders[name][0]()
return d.data, d.target, loaders[name][1]
def prepare_data(X, y, test_size=0.2, batch_size=64, seed=42, scale=True):
"""Split, scale, and create DataLoaders. Returns (train_loader, test_loader, d_in, n_classes)."""
stratify = y if len(np.unique(y)) < 20 else None
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=test_size, random_state=seed, stratify=stratify)
if scale:
sc = StandardScaler()
X_tr, X_te = sc.fit_transform(X_tr), sc.transform(X_te)
X_tr_t, X_te_t = torch.FloatTensor(X_tr), torch.FloatTensor(X_te)
n_classes = len(np.unique(y))
if n_classes > 20:
y_tr_t = torch.FloatTensor(y_tr).unsqueeze(-1)
y_te_t = torch.FloatTensor(y_te).unsqueeze(-1)
n_classes = 1
else:
y_tr_t, y_te_t = torch.LongTensor(y_tr), torch.LongTensor(y_te)
train_loader = DataLoader(TensorDataset(X_tr_t, y_tr_t), batch_size=batch_size, shuffle=True)
test_loader = DataLoader(TensorDataset(X_te_t, y_te_t), batch_size=batch_size)
return train_loader, test_loader, X_tr.shape[1], n_classes
def train_epoch(model, loader, criterion, optimizer, device):
"""Train one epoch. Returns average loss."""
model.train()
total, n = 0.0, 0
for xb, yb in loader:
xb, yb = xb.to(device), yb.to(device)
optimizer.zero_grad()
loss = criterion(model(xb), yb)
loss.backward()
optimizer.step()
total += loss.item()
n += 1
return total / max(n, 1)
@torch.no_grad()
def evaluate(model, loader, criterion, device, task="classification"):
"""Evaluate model. Returns dict with 'loss' and 'accuracy' or 'rmse'."""
model.eval()
total_loss, correct, total = 0.0, 0, 0
preds_all, targs_all = [], []
for xb, yb in loader:
xb, yb = xb.to(device), yb.to(device)
out = model(xb)
total_loss += criterion(out, yb).item()
if task == "classification":
correct += (out.argmax(-1) == yb).sum().item()
total += yb.shape[0]
else:
preds_all.append(out.cpu())
targs_all.append(yb.cpu())
res = {"loss": total_loss / max(len(loader), 1)}
if task == "classification":
res["accuracy"] = correct / max(total, 1)
else:
res["rmse"] = torch.sqrt(((torch.cat(preds_all) - torch.cat(targs_all)) ** 2).mean()).item()
return res
def train_model(model, train_loader, test_loader, n_epochs=100, lr=1e-3, device=None, task="classification", verbose=True, print_every=10):
"""Full training loop. Returns history dict."""
if device is None:
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)
criterion = nn.CrossEntropyLoss() if task == "classification" else nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=lr)
history = {"train_losses": [], "test_losses": [], "test_metrics": []}
for epoch in range(1, n_epochs + 1):
tl = train_epoch(model, train_loader, criterion, optimizer, device)
te = evaluate(model, test_loader, criterion, device, task)
history["train_losses"].append(tl)
history["test_losses"].append(te["loss"])
mk = "accuracy" if task == "classification" else "rmse"
history["test_metrics"].append(te[mk])
if verbose and epoch % print_every == 0:
print(f"Epoch {epoch:4d} | Train: {tl:.4f} | Test: {te['loss']:.4f} | {mk}: {te[mk]:.4f}")
return history
|