708 KiB
Executable File
708 KiB
Executable File
In [3]:
ID = "12002840"
NOMBRE = "Alejandro Lembke Barrientos"
In [4]:
from pathlib import Path
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.datasets import make_classification, make_multilabel_classification
from sklearn.model_selection import train_test_split, KFold
from sklearn.preprocessing import MinMaxScaler, StandardScaler
from sklearn.metrics import (
accuracy_score,
precision_score,
recall_score,
f1_score,
confusion_matrix,
mean_squared_error,
mean_absolute_error,
r2_score,
)
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import TensorDataset, DataLoader
torch.manual_seed(42)
np.random.seed(42)
DATA_DIR = Path("data")
DATA_DIR.mkdir(exist_ok=True)
print("PyTorch:", torch.__version__)
print("Data directory:", DATA_DIR.resolve())
PyTorch: 2.10.0a0+b4e4ee81d3.nv25.12 Data directory: /workspace/data
In [5]:
%%writefile public_tests.py
"""
Public tests for the MLP lab.
Usage inside the notebook:
import public_tests as public_tests
public_tests.test_tarea1(globals())
public_tests.test_tarea2(globals())
public_tests.test_tarea3(globals())
These tests are intentionally lightweight. Passing public tests does not guarantee
full credit in the private grader.
"""
from pathlib import Path
import numpy as np
import pandas as pd
import torch
import torch.nn as nn
from torch.utils.data import TensorDataset, DataLoader
from sklearn.preprocessing import MinMaxScaler, StandardScaler
DATA_DIR = Path("data")
def _require(namespace, name):
assert name in namespace, f"Missing required object: {name}"
return namespace[name]
def _preprocess_features(train_df, dev_df, feature_cols=("x1", "x2")):
X_train = train_df[list(feature_cols)].to_numpy(dtype=np.float32)
X_dev = dev_df[list(feature_cols)].to_numpy(dtype=np.float32)
all_X = np.vstack([X_train, X_dev])
normalizer = MinMaxScaler()
normalizer.fit(all_X)
X_train_norm = normalizer.transform(X_train)
X_dev_norm = normalizer.transform(X_dev)
standardizer = StandardScaler()
standardizer.fit(X_train_norm)
X_train_std = standardizer.transform(X_train_norm).astype(np.float32)
X_dev_std = standardizer.transform(X_dev_norm).astype(np.float32)
return X_train_std, X_dev_std
def _loader_regression(batch_size=64):
train_df = pd.read_csv(DATA_DIR / "dataset_B_train.csv")
dev_df = pd.read_csv(DATA_DIR / "dataset_B_dev.csv")
X_train, X_dev = _preprocess_features(train_df, dev_df)
y_train = train_df["y"].to_numpy(dtype=np.float32).reshape(-1, 1)
y_dev = dev_df["y"].to_numpy(dtype=np.float32).reshape(-1, 1)
train_ds = TensorDataset(torch.tensor(X_train), torch.tensor(y_train))
dev_ds = TensorDataset(torch.tensor(X_dev), torch.tensor(y_dev))
return DataLoader(train_ds, batch_size=batch_size, shuffle=True), DataLoader(dev_ds, batch_size=256)
def _loader_multiclass(batch_size=64):
train_df = pd.read_csv(DATA_DIR / "dataset_C_train.csv")
dev_df = pd.read_csv(DATA_DIR / "dataset_C_dev.csv")
X_train, X_dev = _preprocess_features(train_df, dev_df)
y_train = train_df["y"].to_numpy(dtype=np.int64)
y_dev = dev_df["y"].to_numpy(dtype=np.int64)
train_ds = TensorDataset(torch.tensor(X_train), torch.tensor(y_train))
dev_ds = TensorDataset(torch.tensor(X_dev), torch.tensor(y_dev))
return DataLoader(train_ds, batch_size=batch_size, shuffle=True), DataLoader(dev_ds, batch_size=256)
def _loader_multilabel(batch_size=64):
train_df = pd.read_csv(DATA_DIR / "dataset_D_train.csv")
dev_df = pd.read_csv(DATA_DIR / "dataset_D_dev.csv")
X_train, X_dev = _preprocess_features(train_df, dev_df)
y_cols = ["y0", "y1", "y2", "y3"]
y_train = train_df[y_cols].to_numpy(dtype=np.float32)
y_dev = dev_df[y_cols].to_numpy(dtype=np.float32)
train_ds = TensorDataset(torch.tensor(X_train), torch.tensor(y_train))
dev_ds = TensorDataset(torch.tensor(X_dev), torch.tensor(y_dev))
return DataLoader(train_ds, batch_size=batch_size, shuffle=True), DataLoader(dev_ds, batch_size=256)
def _count_linears(model):
return [m for m in model.modules() if isinstance(m, nn.Linear)]
def _has_relu(model):
return any(isinstance(m, nn.ReLU) for m in model.modules())
def _assert_three_layer_mlp(model, output_dim):
linears = _count_linears(model)
assert len(linears) == 3, "The model must contain exactly 3 Linear layers."
assert linears[0].in_features == 2, "The first Linear layer must receive 2 input features."
assert linears[0].out_features == 8, "The first hidden layer must have 8 units."
assert linears[1].in_features == 8 and linears[1].out_features == 8, "The second hidden layer must be 8 -> 8."
assert linears[2].in_features == 8 and linears[2].out_features == output_dim, f"The output layer must have {output_dim} units."
assert _has_relu(model), "The model should include ReLU activations."
def test_tarea1(namespace):
torch.manual_seed(123)
RegressionMLP = _require(namespace, "RegressionMLP")
train_fn = _require(namespace, "train_regression_model")
eval_fn = _require(namespace, "evaluate_regression_model")
model = RegressionMLP()
assert isinstance(model, nn.Module), "RegressionMLP must be a torch.nn.Module."
_assert_three_layer_mlp(model, output_dim=1)
x = torch.randn(5, 2)
out = model(x)
assert tuple(out.shape) == (5, 1), "RegressionMLP forward output must have shape (batch, 1)."
train_loader, dev_loader = _loader_regression()
_ = train_fn(model, train_loader, epochs=5, lr=1e-2)
metrics = eval_fn(model, dev_loader)
assert isinstance(metrics, dict), "evaluate_regression_model must return a dictionary."
for key in ["mse", "rmse", "mae", "r2"]:
assert key in metrics, f"Missing regression metric: {key}"
assert np.isfinite(metrics[key]), f"Metric {key} must be finite."
print("Tarea 1 public tests passed.")
def test_tarea2(namespace):
torch.manual_seed(123)
MulticlassMLP = _require(namespace, "MulticlassMLP")
train_fn = _require(namespace, "train_multiclass_model")
eval_fn = _require(namespace, "evaluate_multiclass_model")
model = MulticlassMLP()
assert isinstance(model, nn.Module), "MulticlassMLP must be a torch.nn.Module."
_assert_three_layer_mlp(model, output_dim=4)
x = torch.randn(5, 2)
out = model(x)
assert tuple(out.shape) == (5, 4), "MulticlassMLP forward output must have shape (batch, 4)."
train_loader, dev_loader = _loader_multiclass()
_ = train_fn(model, train_loader, epochs=5, lr=1e-2)
metrics = eval_fn(model, dev_loader)
assert isinstance(metrics, dict), "evaluate_multiclass_model must return a dictionary."
for key in ["accuracy", "macro_f1", "confusion_matrix"]:
assert key in metrics, f"Missing multiclass metric: {key}"
assert np.isfinite(metrics["accuracy"]), "accuracy must be finite."
assert np.isfinite(metrics["macro_f1"]), "macro_f1 must be finite."
print("Tarea 2 public tests passed.")
def test_tarea3(namespace):
torch.manual_seed(123)
MultilabelMLP = _require(namespace, "MultilabelMLP")
train_fn = _require(namespace, "train_multilabel_model")
eval_fn = _require(namespace, "evaluate_multilabel_model")
model = MultilabelMLP()
assert isinstance(model, nn.Module), "MultilabelMLP must be a torch.nn.Module."
_assert_three_layer_mlp(model, output_dim=4)
x = torch.randn(5, 2)
out = model(x)
assert tuple(out.shape) == (5, 4), "MultilabelMLP forward output must have shape (batch, 4)."
train_loader, dev_loader = _loader_multilabel()
_ = train_fn(model, train_loader, epochs=5, lr=1e-2)
metrics = eval_fn(model, dev_loader)
assert isinstance(metrics, dict), "evaluate_multilabel_model must return a dictionary."
for key in ["subset_accuracy", "micro_f1", "macro_f1"]:
assert key in metrics, f"Missing multilabel metric: {key}"
assert np.isfinite(metrics[key]), f"Metric {key} must be finite."
print("Tarea 3 public tests passed.")
Writing public_tests.py
In [6]:
import importlib
import public_tests
importlib.reload(public_tests)
print("public_tests listo en:", Path("public_tests.py").resolve())
public_tests listo en: /workspace/public_tests.py
In [7]:
rng = np.random.default_rng(42)
# Dataset A: clasificación binaria, 2 features
XA, yA = make_classification(
n_samples=1200,
n_features=2,
n_informative=2,
n_redundant=0,
n_repeated=0,
n_classes=2,
n_clusters_per_class=1,
class_sep=1.35,
flip_y=0.03,
random_state=11,
)
XA_train, XA_dev, yA_train, yA_dev = train_test_split(
XA, yA,
train_size=1000,
test_size=200,
stratify=yA,
random_state=101,
)
# Dataset B: regresión no lineal de orden 5, 2 features
XB = rng.uniform(-1.6, 1.6, size=(1200, 2))
x1, x2 = XB[:, 0], XB[:, 1]
noise = rng.normal(0, 0.35, size=1200)
yB = (
0.55 * x1**5
- 0.45 * x2**5
+ 0.80 * x1**3
- 0.65 * x2**2
+ 0.50 * x1**2 * x2
- 0.35 * x1 * x2**3
+ 0.25 * x1
+ noise
)
XB_train, XB_dev, yB_train, yB_dev = train_test_split(
XB, yB,
train_size=1000,
test_size=200,
random_state=202,
)
# Dataset C: clasificación multiclase con 4 clases, 2 features
XC, yC = make_classification(
n_samples=1200,
n_features=2,
n_informative=2,
n_redundant=0,
n_repeated=0,
n_classes=4,
n_clusters_per_class=1,
class_sep=1.45,
flip_y=0.04,
random_state=33,
)
XC_train, XC_dev, yC_train, yC_dev = train_test_split(
XC, yC,
train_size=1000,
test_size=200,
stratify=yC,
random_state=303,
)
# Dataset D: clasificación multilabel con 4 clases, 2 features
XD, yD = make_multilabel_classification(
n_samples=1200,
n_features=2,
n_classes=4,
n_labels=2,
allow_unlabeled=False,
random_state=44,
)
XD = XD.astype(np.float64) + rng.normal(0, 0.05, size=XD.shape)
strat_D = yD.sum(axis=1)
XD_train, XD_dev, yD_train, yD_dev = train_test_split(
XD, yD,
train_size=1000,
test_size=200,
stratify=strat_D,
random_state=404,
)
def save_csv(prefix, X_train, X_dev, y_train, y_dev, multilabel=False):
if multilabel:
train_df = pd.DataFrame(X_train, columns=["x1", "x2"])
dev_df = pd.DataFrame(X_dev, columns=["x1", "x2"])
for k in range(y_train.shape[1]):
train_df[f"y{k}"] = y_train[:, k].astype(int)
dev_df[f"y{k}"] = y_dev[:, k].astype(int)
else:
train_df = pd.DataFrame(X_train, columns=["x1", "x2"])
dev_df = pd.DataFrame(X_dev, columns=["x1", "x2"])
train_df["y"] = y_train
dev_df["y"] = y_dev
train_df.to_csv(DATA_DIR / f"dataset_{prefix}_train.csv", index=False)
dev_df.to_csv(DATA_DIR / f"dataset_{prefix}_dev.csv", index=False)
save_csv("A", XA_train, XA_dev, yA_train, yA_dev)
save_csv("B", XB_train, XB_dev, yB_train, yB_dev)
save_csv("C", XC_train, XC_dev, yC_train, yC_dev)
save_csv("D", XD_train, XD_dev, yD_train, yD_dev, multilabel=True)
for path in sorted(DATA_DIR.glob("*.csv")):
print(path, pd.read_csv(path).shape)
data/dataset_A_dev.csv (200, 3) data/dataset_A_train.csv (1000, 3) data/dataset_B_dev.csv (200, 3) data/dataset_B_train.csv (1000, 3) data/dataset_C_dev.csv (200, 3) data/dataset_C_train.csv (1000, 3) data/dataset_D_dev.csv (200, 6) data/dataset_D_train.csv (1000, 6)
In [8]:
rng = np.random.default_rng(123)
p_true = 0.72
n = 1000
sample = rng.binomial(n=1, p=p_true, size=n)
p_empirical = sample.mean()
grid = np.linspace(0.01, 0.99, 200)
def bernoulli_log_likelihood(sample, p):
return np.sum(sample * np.log(p) + (1 - sample) * np.log(1 - p))
def kl_bernoulli(q, p):
return q * np.log(q / p) + (1 - q) * np.log((1 - q) / (1 - p))
log_likelihoods = np.array([bernoulli_log_likelihood(sample, p) for p in grid])
kl_values = np.array([kl_bernoulli(p_empirical, p) for p in grid])
p_mle_grid = grid[np.argmax(log_likelihoods)]
p_kl_min_grid = grid[np.argmin(kl_values)]
print("p verdadero:", p_true)
print("p empírico:", p_empirical)
print("p que maximiza log-verosimilitud en la grilla:", p_mle_grid)
print("p que minimiza KL en la grilla:", p_kl_min_grid)
plt.figure(figsize=(7, 4))
plt.plot(grid, log_likelihoods)
plt.axvline(p_true, linestyle="--", label="p verdadero")
plt.axvline(p_empirical, linestyle=":", label="p empírico")
plt.title("Log-verosimilitud Bernoulli")
plt.xlabel("p candidato")
plt.ylabel("log L(p)")
plt.legend()
plt.grid(True)
plt.show()
plt.figure(figsize=(7, 4))
plt.plot(grid, kl_values)
plt.axvline(p_true, linestyle="--", label="p verdadero")
plt.axvline(p_empirical, linestyle=":", label="p empírico")
plt.title("KL entre distribución empírica y Bernoulli candidata")
plt.xlabel("p candidato")
plt.ylabel("D_KL(q || p)")
plt.legend()
plt.grid(True)
plt.show()
p verdadero: 0.72 p empírico: 0.722 p que maximiza log-verosimilitud en la grilla: 0.724070351758794 p que minimiza KL en la grilla: 0.724070351758794
In [9]:
def normalize_all_then_standardize_train(train_df, dev_df, feature_cols=("x1", "x2")):
X_train = train_df[list(feature_cols)].to_numpy(dtype=np.float32)
X_dev = dev_df[list(feature_cols)].to_numpy(dtype=np.float32)
all_X = np.vstack([X_train, X_dev])
normalizer = MinMaxScaler()
normalizer.fit(all_X)
X_train_norm = normalizer.transform(X_train)
X_dev_norm = normalizer.transform(X_dev)
standardizer = StandardScaler()
standardizer.fit(X_train_norm)
X_train_std = standardizer.transform(X_train_norm).astype(np.float32)
X_dev_std = standardizer.transform(X_dev_norm).astype(np.float32)
return X_train_std, X_dev_std, normalizer, standardizer
def make_loader(X, y, batch_size=64, shuffle=True):
X_tensor = torch.tensor(X, dtype=torch.float32)
if y.dtype.kind in {"i", "u"}:
y_tensor = torch.tensor(y, dtype=torch.long)
else:
y_tensor = torch.tensor(y, dtype=torch.float32)
ds = TensorDataset(X_tensor, y_tensor)
return DataLoader(ds, batch_size=batch_size, shuffle=shuffle)
In [10]:
A_train = pd.read_csv(DATA_DIR / "dataset_A_train.csv")
A_dev = pd.read_csv(DATA_DIR / "dataset_A_dev.csv")
plt.figure(figsize=(6, 5))
plt.scatter(A_train["x1"], A_train["x2"], c=A_train["y"], s=16, alpha=0.75)
plt.title("Dataset A: clasificación binaria")
plt.xlabel("x1")
plt.ylabel("x2")
plt.grid(True)
plt.show()
X_A_train, X_A_dev, norm_A, std_A = normalize_all_then_standardize_train(A_train, A_dev)
y_A_train = A_train["y"].to_numpy(dtype=np.float32).reshape(-1, 1)
y_A_dev = A_dev["y"].to_numpy(dtype=np.float32).reshape(-1, 1)
train_A_loader = DataLoader(
TensorDataset(torch.tensor(X_A_train), torch.tensor(y_A_train)),
batch_size=64,
shuffle=True,
)
dev_A_loader = DataLoader(
TensorDataset(torch.tensor(X_A_dev), torch.tensor(y_A_dev)),
batch_size=256,
shuffle=False,
)
In [11]:
class BinaryMLP(nn.Module):
def __init__(self, input_dim=2, hidden_dim=8, output_dim=1):
super().__init__()
self.net = nn.Sequential(
nn.Linear(input_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, output_dim),
)
def forward(self, x):
return self.net(x)
def train_binary_model(model, train_loader, epochs=100, lr=1e-2):
criterion = nn.BCEWithLogitsLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=lr)
history = []
for epoch in range(epochs):
model.train()
total_loss = 0.0
total_n = 0
for xb, yb in train_loader:
logits = model(xb)
loss = criterion(logits, yb)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item() * xb.size(0)
total_n += xb.size(0)
history.append(total_loss / total_n)
return history
def evaluate_binary_model(model, dev_loader, threshold=0.5):
model.eval()
y_true = []
y_pred = []
with torch.no_grad():
for xb, yb in dev_loader:
logits = model(xb)
probs = torch.sigmoid(logits)
preds = (probs >= threshold).int()
y_true.append(yb.cpu().numpy())
y_pred.append(preds.cpu().numpy())
y_true = np.vstack(y_true).reshape(-1)
y_pred = np.vstack(y_pred).reshape(-1)
return {
"accuracy": accuracy_score(y_true, y_pred),
"precision": precision_score(y_true, y_pred, zero_division=0),
"recall": recall_score(y_true, y_pred, zero_division=0),
"f1": f1_score(y_true, y_pred, zero_division=0),
"confusion_matrix": confusion_matrix(y_true, y_pred),
}
binary_model = BinaryMLP()
history_A = train_binary_model(binary_model, train_A_loader, epochs=100, lr=1e-2)
metrics_A = evaluate_binary_model(binary_model, dev_A_loader)
print(metrics_A)
plt.figure(figsize=(6, 4))
plt.plot(history_A)
plt.title("Dataset A: pérdida de entrenamiento")
plt.xlabel("epoch")
plt.ylabel("BCEWithLogitsLoss")
plt.grid(True)
plt.show()
{'accuracy': 0.98, 'precision': 0.9702970297029703, 'recall': 0.98989898989899, 'f1': 0.98, 'confusion_matrix': array([[98, 3],
[ 1, 98]])}
In [12]:
B_train = pd.read_csv(DATA_DIR / "dataset_B_train.csv")
B_dev = pd.read_csv(DATA_DIR / "dataset_B_dev.csv")
# Non-graded: visualización
plt.figure(figsize=(6, 5))
plt.scatter(B_train["x1"], B_train["x2"], c=B_train["y"], s=16, alpha=0.75)
plt.title("Dataset B: regresión no lineal")
plt.xlabel("x1")
plt.ylabel("x2")
plt.colorbar(label="y")
plt.grid(True)
plt.show()
# Non-graded: normalización y estandarización
X_B_train, X_B_dev, norm_B, std_B = normalize_all_then_standardize_train(B_train, B_dev)
y_B_train = B_train["y"].to_numpy(dtype=np.float32).reshape(-1, 1)
y_B_dev = B_dev["y"].to_numpy(dtype=np.float32).reshape(-1, 1)
train_B_loader = DataLoader(
TensorDataset(torch.tensor(X_B_train), torch.tensor(y_B_train)),
batch_size=64,
shuffle=True,
)
dev_B_loader = DataLoader(
TensorDataset(torch.tensor(X_B_dev), torch.tensor(y_B_dev)),
batch_size=256,
shuffle=False,
)
print("X_B_train:", X_B_train.shape)
print("y_B_train:", y_B_train.shape)
X_B_train: (1000, 2) y_B_train: (1000, 1)
In [13]:
# GRADED: Tarea 1
class RegressionMLP(nn.Module):
def __init__(self, input_dim=2, hidden_dim=8, output_dim=1):
super().__init__()
# START CODE HERE
# Arquitectura 2 -> 8 -> 8 -> 1.
# Las activaciones se registran como módulos nn.ReLU (no F.relu) para que
# aparezcan en model.modules(). La capa de salida queda lineal: en regresión
# el modelo predice directamente la media condicional de y.
self.net = nn.Sequential(
nn.Linear(input_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, output_dim),
)
# END CODE HERE
def forward(self, x):
# START CODE HERE
return self.net(x)
# END CODE HERE
def train_regression_model(model, train_loader, epochs=100, lr=1e-2):
"""
Entrena un modelo de regresión usando MSELoss.
Debe retornar una lista con la pérdida promedio por epoch.
"""
# START CODE HERE
# MSE es la log-verosimilitud negativa bajo error gaussiano homocedástico,
# salvo constantes y escala.
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=lr)
history = []
for epoch in range(epochs):
model.train()
total_loss = 0.0
total_n = 0
for xb, yb in train_loader:
preds = model(xb)
loss = criterion(preds, yb)
optimizer.zero_grad()
loss.backward()
optimizer.step()
# Se pondera por el tamaño del batch para que el último batch,
# que puede ser más pequeño, no distorsione el promedio.
total_loss += loss.item() * xb.size(0)
total_n += xb.size(0)
history.append(total_loss / total_n)
return history
# END CODE HERE
def evaluate_regression_model(model, dev_loader):
"""
Evalúa el modelo de regresión.
Debe retornar un diccionario con:
mse, rmse, mae, r2
"""
# START CODE HERE
model.eval()
y_true = []
y_pred = []
with torch.no_grad():
for xb, yb in dev_loader:
preds = model(xb)
y_true.append(yb.cpu().numpy())
y_pred.append(preds.cpu().numpy())
y_true = np.vstack(y_true).reshape(-1)
y_pred = np.vstack(y_pred).reshape(-1)
# El RMSE se calcula con np.sqrt: el argumento squared=False de
# mean_squared_error fue eliminado en scikit-learn 1.6.
mse = mean_squared_error(y_true, y_pred)
return {
"mse": float(mse),
"rmse": float(np.sqrt(mse)),
"mae": float(mean_absolute_error(y_true, y_pred)),
"r2": float(r2_score(y_true, y_pred)),
}
# END CODE HERE
In [14]:
# Public test: ejecutar después de implementar Tarea 1.
public_tests.test_tarea1(globals())
Tarea 1 public tests passed.
In [15]:
C_train = pd.read_csv(DATA_DIR / "dataset_C_train.csv")
C_dev = pd.read_csv(DATA_DIR / "dataset_C_dev.csv")
# Non-graded: visualización
plt.figure(figsize=(6, 5))
plt.scatter(C_train["x1"], C_train["x2"], c=C_train["y"], s=16, alpha=0.75)
plt.title("Dataset C: clasificación multiclase")
plt.xlabel("x1")
plt.ylabel("x2")
plt.colorbar(label="clase")
plt.grid(True)
plt.show()
# Non-graded: normalización y estandarización
X_C_train, X_C_dev, norm_C, std_C = normalize_all_then_standardize_train(C_train, C_dev)
y_C_train = C_train["y"].to_numpy(dtype=np.int64)
y_C_dev = C_dev["y"].to_numpy(dtype=np.int64)
train_C_loader = DataLoader(
TensorDataset(torch.tensor(X_C_train), torch.tensor(y_C_train)),
batch_size=64,
shuffle=True,
)
dev_C_loader = DataLoader(
TensorDataset(torch.tensor(X_C_dev), torch.tensor(y_C_dev)),
batch_size=256,
shuffle=False,
)
print("X_C_train:", X_C_train.shape)
print("y_C_train:", y_C_train.shape)
X_C_train: (1000, 2) y_C_train: (1000,)
In [16]:
# GRADED: Tarea 2
class MulticlassMLP(nn.Module):
def __init__(self, input_dim=2, hidden_dim=8, output_dim=4):
super().__init__()
# START CODE HERE
# Arquitectura 2 -> 8 -> 8 -> 4. La salida son logits crudos, sin softmax:
# CrossEntropyLoss ya aplica log_softmax internamente, y aplicarlo dos veces
# aplanaría los gradientes.
self.net = nn.Sequential(
nn.Linear(input_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, output_dim),
)
# END CODE HERE
def forward(self, x):
# START CODE HERE
return self.net(x)
# END CODE HERE
def train_multiclass_model(model, train_loader, epochs=100, lr=1e-2):
"""
Entrena un modelo de clasificación multiclase usando CrossEntropyLoss.
Debe retornar una lista con la pérdida promedio por epoch.
"""
# START CODE HERE
# CrossEntropyLoss espera logits (N, C) y etiquetas enteras (N,) de tipo long.
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=lr)
history = []
for epoch in range(epochs):
model.train()
total_loss = 0.0
total_n = 0
for xb, yb in train_loader:
logits = model(xb)
loss = criterion(logits, yb)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item() * xb.size(0)
total_n += xb.size(0)
history.append(total_loss / total_n)
return history
# END CODE HERE
def evaluate_multiclass_model(model, dev_loader):
"""
Evalúa el modelo multiclase.
Debe retornar un diccionario con:
accuracy, macro_f1, confusion_matrix
"""
# START CODE HERE
model.eval()
y_true = []
y_pred = []
with torch.no_grad():
for xb, yb in dev_loader:
logits = model(xb)
# argmax sobre los logits equivale a argmax sobre las probabilidades,
# porque softmax es monótona: no hace falta calcularla.
preds = logits.argmax(dim=1)
y_true.append(yb.cpu().numpy())
y_pred.append(preds.cpu().numpy())
y_true = np.concatenate(y_true)
y_pred = np.concatenate(y_pred)
return {
"accuracy": float(accuracy_score(y_true, y_pred)),
# macro promedia el F1 por clase sin ponderar por soporte, así que
# penaliza el desempeño pobre en las clases minoritarias.
"macro_f1": float(f1_score(y_true, y_pred, average="macro", zero_division=0)),
"confusion_matrix": confusion_matrix(y_true, y_pred),
}
# END CODE HERE
In [17]:
# Public test: ejecutar después de implementar Tarea 2.
public_tests.test_tarea2(globals())
Tarea 2 public tests passed.
In [18]:
D_train = pd.read_csv(DATA_DIR / "dataset_D_train.csv")
D_dev = pd.read_csv(DATA_DIR / "dataset_D_dev.csv")
# Non-graded: visualización.
# Para visualizar multilabel en 2D, coloreamos por el número de etiquetas activas.
label_cols = ["y0", "y1", "y2", "y3"]
D_train["label_count"] = D_train[label_cols].sum(axis=1)
plt.figure(figsize=(6, 5))
plt.scatter(D_train["x1"], D_train["x2"], c=D_train["label_count"], s=16, alpha=0.75)
plt.title("Dataset D: clasificación multilabel")
plt.xlabel("x1")
plt.ylabel("x2")
plt.colorbar(label="número de etiquetas activas")
plt.grid(True)
plt.show()
# Non-graded: normalización y estandarización
X_D_train, X_D_dev, norm_D, std_D = normalize_all_then_standardize_train(D_train, D_dev)
y_D_train = D_train[label_cols].to_numpy(dtype=np.float32)
y_D_dev = D_dev[label_cols].to_numpy(dtype=np.float32)
train_D_loader = DataLoader(
TensorDataset(torch.tensor(X_D_train), torch.tensor(y_D_train)),
batch_size=64,
shuffle=True,
)
dev_D_loader = DataLoader(
TensorDataset(torch.tensor(X_D_dev), torch.tensor(y_D_dev)),
batch_size=256,
shuffle=False,
)
print("X_D_train:", X_D_train.shape)
print("y_D_train:", y_D_train.shape)
X_D_train: (1000, 2) y_D_train: (1000, 4)
In [19]:
# GRADED: Tarea 3
class MultilabelMLP(nn.Module):
def __init__(self, input_dim=2, hidden_dim=8, output_dim=4):
super().__init__()
# START CODE HERE
# Arquitectura 2 -> 8 -> 8 -> 4. Cada una de las 4 salidas es el logit de una
# Bernoulli independiente, así que la sigmoide se aplica después (dentro de
# BCEWithLogitsLoss al entrenar, y de forma explícita al evaluar).
self.net = nn.Sequential(
nn.Linear(input_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, output_dim),
)
# END CODE HERE
def forward(self, x):
# START CODE HERE
return self.net(x)
# END CODE HERE
def train_multilabel_model(model, train_loader, epochs=100, lr=1e-2):
"""
Entrena un modelo de clasificación multilabel usando BCEWithLogitsLoss.
Debe retornar una lista con la pérdida promedio por epoch.
"""
# START CODE HERE
# BCEWithLogitsLoss combina sigmoide y BCE en una sola operación estable
# numéricamente (log-sum-exp), y promedia sobre las 4 etiquetas.
criterion = nn.BCEWithLogitsLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=lr)
history = []
for epoch in range(epochs):
model.train()
total_loss = 0.0
total_n = 0
for xb, yb in train_loader:
logits = model(xb)
loss = criterion(logits, yb)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item() * xb.size(0)
total_n += xb.size(0)
history.append(total_loss / total_n)
return history
# END CODE HERE
def evaluate_multilabel_model(model, dev_loader, threshold=0.5):
"""
Evalúa el modelo multilabel.
Debe retornar un diccionario con:
subset_accuracy, micro_f1, macro_f1
"""
# START CODE HERE
model.eval()
y_true = []
y_pred = []
with torch.no_grad():
for xb, yb in dev_loader:
logits = model(xb)
probs = torch.sigmoid(logits)
# Cada etiqueta se decide por separado contra el umbral.
preds = (probs >= threshold).int()
y_true.append(yb.cpu().numpy())
y_pred.append(preds.cpu().numpy())
# Ambas matrices se castean a entero: las etiquetas llegan como float32 y
# sklearn interpretaría una mezcla float/bool como continuous-multioutput.
y_true = np.vstack(y_true).astype(int)
y_pred = np.vstack(y_pred).astype(int)
return {
# Con matrices 2D, accuracy_score exige que las 4 etiquetas de la fila
# coincidan: es exactamente el subset accuracy (exact match ratio).
"subset_accuracy": float(accuracy_score(y_true, y_pred)),
# micro agrega TP/FP/FN de todas las etiquetas antes de calcular el F1;
# macro promedia el F1 de cada etiqueta por separado.
"micro_f1": float(f1_score(y_true, y_pred, average="micro", zero_division=0)),
"macro_f1": float(f1_score(y_true, y_pred, average="macro", zero_division=0)),
}
# END CODE HERE
In [20]:
# Public test: ejecutar después de implementar Tarea 3.
public_tests.test_tarea3(globals())
Tarea 3 public tests passed.
In [21]:
# Cálculos rápidos para verificar los ejemplos a mano.
z = 0.5 * 2 + (-2) * (-1) + 1
relu_z = max(0, z)
print("Forward neurona:", z, relu_z)
y = np.array([3.0, 0.0, 1.0])
y_hat = np.array([2.5, 0.7, 1.2])
mse = np.mean((y - y_hat) ** 2)
print("MSE:", mse)
print("BCE y=1, p=0.8:", -np.log(0.8))
print("BCE y=0, p=0.8:", -np.log(1 - 0.8))
p = np.array([0.1, 0.7, 0.2])
correct_class = 2
print("CE:", -np.log(p[correct_class]))
Forward neurona: 4.0 4.0 MSE: 0.26 BCE y=1, p=0.8: 0.2231435513142097 BCE y=0, p=0.8: 1.6094379124341005 CE: 1.6094379124341003