from pathlib import Path
import json, time, copy, hashlib, platform, gzip, struct, urllib.request
import numpy as np
import pandas as pd
import sklearn
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, confusion_matrix, f1_score
import matplotlib.pyplot as plt
from IPython.display import display, Markdown
from PIL import Image
import torch
from torch import nn
from torch.nn import functional as F
from torch.utils.data import DataLoader, TensorDataset, Dataset
SEED=42
np.random.seed(SEED); torch.manual_seed(SEED)
torch.set_num_threads(2)
DEVICE=torch.device('cuda' if torch.cuda.is_available() else 'cpu')
ARTEFACTOS=Path('artefactos_pytorch'); ARTEFACTOS.mkdir(exist_ok=True)
plt.style.use('seaborn-v0_8-whitegrid')
plt.rcParams.update({'figure.dpi':110,'font.size':10})
print({'python':platform.python_version(),'torch':str(torch.__version__),
'numpy':np.__version__,'sklearn':sklearn.__version__,'dispositivo':str(DEVICE),
'hilos_cpu':torch.get_num_threads(),'semilla':SEED})
def semilla(n=SEED):
np.random.seed(n); torch.manual_seed(n)
def parametros(modelo):
return sum(p.numel() for p in modelo.parameters())
def inicializar(modelo):
# Conv y capas ocultas ReLU: He; salida lineal: Xavier explícito.
for m in modelo.modules():
if isinstance(m,(nn.Conv2d,nn.Linear)):
nn.init.kaiming_normal_(m.weight,nonlinearity='relu')
if m.bias is not None: nn.init.zeros_(m.bias)
lineales=[m for m in modelo.modules() if isinstance(m,nn.Linear)]
if lineales: nn.init.xavier_uniform_(lineales[-1].weight)
return modelo.to(DEVICE)
def conjunto(x,y):
return TensorDataset(torch.as_tensor(np.ascontiguousarray(x),dtype=torch.float32),
torch.as_tensor(np.asarray(y),dtype=torch.long))
def cargador(ds,lote=128,mezclar=False):
return DataLoader(ds,batch_size=lote,shuffle=mezclar,num_workers=0,
generator=torch.Generator().manual_seed(SEED))
@torch.inference_mode()
def evaluar(modelo,ds,lote=128):
modelo.eval(); perdidas=0.; aciertos=0; logits=[]; etiquetas=[]
for x,y in cargador(ds,lote):
x=x.to(DEVICE);y=y.to(DEVICE);z=modelo(x)
perdidas+=F.cross_entropy(z,y,reduction='sum').item()
aciertos+=(z.argmax(1)==y).sum().item()
logits.append(z.cpu());etiquetas.append(y.cpu())
z=torch.cat(logits);y=torch.cat(etiquetas)
assert torch.isfinite(z).all(), 'Logits no finitos'
return {'ce':perdidas/len(ds),'exactitud':aciertos/len(ds),
'logits':z.numpy(),'y':y.numpy(),'probabilidades':z.softmax(1).numpy()}
def entrenar(modelo,train,val,epocas,lote=128,lr=1e-3,paciencia=None,nombre='modelo',
modo_train=None,incluir_inicial=False):
# El optimizador se construye DESPUÉS de fijar requires_grad.
optimizador=torch.optim.Adam([p for p in modelo.parameters() if p.requires_grad],lr=lr)
dl=cargador(train,lote,True); historial=[]; mejor=float('inf'); pesos=None; espera=0
inicio=time.perf_counter();epoca_mejor=None
if incluir_inicial:
inicial=evaluar(modelo,val,lote); mejor=inicial['ce']
pesos=copy.deepcopy(modelo.state_dict());epoca_mejor=0
for epoca in range(1,epocas+1):
modelo.train()
if modo_train is not None: modo_train(modelo)
suma=0.;aciertos=0
for x,y in dl:
x=x.to(DEVICE);y=y.to(DEVICE)
optimizador.zero_grad(set_to_none=True)
z=modelo(x);perdida=F.cross_entropy(z,y)
if not torch.isfinite(perdida): raise FloatingPointError(nombre+' pérdida no finita')
perdida.backward();optimizador.step()
suma+=perdida.item()*len(y);aciertos+=(z.argmax(1)==y).sum().item()
# Comparar curvas en el MISMO modo eval y al cierre de cada época.
tr=evaluar(modelo,train,lote);va=evaluar(modelo,val,lote)
historial.append({'epoca':epoca,'ce_online':suma/len(train),
'acc_online':aciertos/len(train),'ce_train':tr['ce'],'ce_val':va['ce'],
'acc_train':tr['exactitud'],'acc_val':va['exactitud']})
if va['ce']<mejor:
mejor=va['ce'];pesos=copy.deepcopy(modelo.state_dict());epoca_mejor=epoca;espera=0
else: espera+=1
print(f"{nombre} {epoca}/{epocas}: CE val={va['ce']:.4f}, acc val={va['exactitud']:.4f}",flush=True)
if paciencia is not None and espera>=paciencia: break
modelo.load_state_dict(pesos);modelo.eval()
h=pd.DataFrame(historial);h.to_csv(ARTEFACTOS/f'{nombre}-historial.csv',index=False)
resumen={'modelo':nombre,'mejor_epoca':epoca_mejor,'epocas':len(h),'ce_val':mejor,
'segundos':time.perf_counter()-inicio,'parametros':parametros(modelo),
'entrenables':sum(p.numel() for p in modelo.parameters() if p.requires_grad)}
(ARTEFACTOS/f'{nombre}-entrenamiento.json').write_text(json.dumps(resumen,indent=2),encoding='utf-8')
return h,resumen
def curvas(h,titulo,mejor=None):
fig,axes=plt.subplots(1,2,figsize=(11,3.8))
for ax,a,b,t in [(axes[0],'ce_train','ce_val','Entropía cruzada'),
(axes[1],'acc_train','acc_val','Exactitud')]:
ax.plot(h.epoca,h[a],'-o',label='Entrenamiento (eval)',ms=3)
ax.plot(h.epoca,h[b],'-o',label='Validación',ms=3)
if mejor is not None: ax.axvline(mejor,color='.4',ls='--',label=f'Checkpoint {mejor}')
ax.set(xlabel='Época',ylabel=t);ax.legend(fontsize=8)
fig.suptitle(titulo);fig.tight_layout();plt.show()
def lectura(h,res):
e=res['mejor_epoca'];ultima=h.iloc[-1]
texto=(f"**Lectura del experimento.** Se ejecutaron {len(h)} épocas; el checkpoint seleccionado "
f"por CE de validación corresponde a la época {e}, con CE {res['ce_val']:.4f}. "
f"Al cierre, la brecha de exactitud train−val es "
f"{100*(ultima.acc_train-ultima.acc_val):+.2f} puntos porcentuales. "
"Una brecha describe este conjunto y este presupuesto; no identifica por sí sola su causa. ")
if e==0: texto+='El ajuste fino no mejoró la CE inicial: se conserva el punto de partida de la fase congelada.'
elif e<len(h): texto+='La última época no es el modelo recuperado: prolongar este ensayo no redujo el mínimo de validación.'
else: texto+='El mínimo está en el límite del presupuesto; no demuestra que más épocas mejorarían el resultado.'
display(Markdown(texto))
def diagnostico(resultado,nombres,imagenes=None,titulo='Prueba'):
y=resultado['y'];p=resultado['probabilidades'];pred=p.argmax(1);k=len(nombres)
print(classification_report(y,pred,labels=np.arange(k),target_names=nombres,digits=4,zero_division=0))
cm=confusion_matrix(y,pred,labels=np.arange(k));normal=cm/np.maximum(cm.sum(1,keepdims=True),1)
fig,ax=plt.subplots(figsize=(7,5.5));im=ax.imshow(normal,vmin=0,vmax=1,cmap='Blues')
ax.set_xticks(range(k),nombres,rotation=55,ha='right',fontsize=8)
ax.set_yticks(range(k),nombres,fontsize=8);ax.set(xlabel='Predicho',ylabel='Real',title=titulo+' · proporción por clase')
for i in range(k):
for j in range(k):
if cm[i,j]: ax.text(j,i,str(cm[i,j]),ha='center',va='center',fontsize=7,color='white' if normal[i,j]>.5 else 'black')
fig.colorbar(im,ax=ax,label='Proporción de la clase real');fig.tight_layout();plt.show()
err=np.flatnonzero(y!=pred)
peor=int(np.argmin(np.diag(normal)))
fuera=cm.copy();np.fill_diagonal(fuera,0);i,j=np.unravel_index(fuera.argmax(),fuera.shape)
display(Markdown(f"**Decisión después de medir.** Exactitud {resultado['exactitud']:.2%}; "
f"macro-F1 {f1_score(y,pred,average='macro'):.4f}; {len(err)} errores de {len(y)}. "
f"El menor recall corresponde a **{nombres[peor]}** ({normal[peor,peor]:.2%}). "
+(f"La confusión más frecuente es {nombres[i]} → {nombres[j]} ({fuera[i,j]} casos). " if len(err) else '')+
"Esto orienta una revisión de etiquetas, cobertura y costes por clase; no autoriza volver a seleccionar con este test."))
if imagenes is not None and len(err):
confianza=p.max(1);orden=err[np.argsort(-confianza[err])][:8]
fig,axes=plt.subplots(2,4,figsize=(11,5))
for ax,idx in zip(axes.flat,orden):
img=np.asarray(imagenes[idx]);img=np.moveaxis(img,0,-1) if img.ndim==3 and img.shape[0] in [1,3] else img
ax.imshow(img.squeeze(),cmap='gray' if img.squeeze().ndim==2 else None)
ax.set_title(f'Real: {nombres[y[idx]]}\nPred: {nombres[pred[idx]]} · p={confianza[idx]:.2f}',fontsize=8);ax.axis('off')
for ax in list(axes.flat)[len(orden):]: ax.axis('off')
fig.suptitle('Errores de mayor probabilidad predicha · no es una medida de calibración')
fig.tight_layout();plt.show()
return cm
def guardar(modelo,fabrica,nombre,contrato,val):
ruta=ARTEFACTOS/f'{nombre}.pt'
estado={k:v.detach().cpu().clone() for k,v in modelo.state_dict().items()}
torch.save({'state_dict':estado,'contrato':contrato,'torch':str(torch.__version__)},ruta)
cargado=fabrica().to(DEVICE)
paquete=torch.load(ruta,map_location=DEVICE,weights_only=True)
cargado.load_state_dict(paquete['state_dict']);cargado.eval()
# Comprobar recarga sobre validación; nunca seleccionar con prueba.
muestra=next(iter(cargador(val,32)))[0].to(DEVICE);modelo.eval()
with torch.inference_mode(): torch.testing.assert_close(modelo(muestra),cargado(muestra),rtol=1e-5,atol=1e-6)
contrato=dict(contrato,sha256=hashlib.sha256(ruta.read_bytes()).hexdigest(),bytes=ruta.stat().st_size)
(ARTEFACTOS/f'{nombre}-contrato.json').write_text(json.dumps(contrato,indent=2,ensure_ascii=False),encoding='utf-8')
print('Recarga verificada:',ruta,'bytes:',ruta.stat().st_size)
return cargado
def mnist(nombre='mnist'):
cache=Path.home()/'.keras'/'datasets'
if nombre=='mnist' and (cache/'mnist.npz').exists():
with np.load(cache/'mnist.npz',allow_pickle=False) as d:
return d['x_train'],d['y_train'],d['x_test'],d['y_test']
from torchvision.datasets import MNIST,FashionMNIST
clase=MNIST if nombre=='mnist' else FashionMNIST
# Leer la caché histórica IDX sin importar TensorFlow.
if nombre=='fashion' and (cache/'fashion-mnist'/'train-images-idx3-ubyte.gz').exists():
def idx(f,imagenes):
with gzip.open(cache/'fashion-mnist'/f,'rb') as h: b=h.read()
magic,n=struct.unpack('>II',b[:8]);assert magic==(2051 if imagenes else 2049)
return np.frombuffer(b,dtype=np.uint8,offset=16 if imagenes else 8).reshape((n,28,28) if imagenes else (n,)).copy()
return (idx('train-images-idx3-ubyte.gz',True),idx('train-labels-idx1-ubyte.gz',False),
idx('t10k-images-idx3-ubyte.gz',True),idx('t10k-labels-idx1-ubyte.gz',False))
a=clase(str(Path.home()/'.cache'/'bit-datasets'),train=True,download=True)
b=clase(str(Path.home()/'.cache'/'bit-datasets'),train=False,download=True)
return a.data.numpy(),a.targets.numpy(),b.data.numpy(),b.targets.numpy()
def galeria(x,y,nombres,titulo):
fig,axes=plt.subplots(2,5,figsize=(11,4.5))
for i,ax in enumerate(axes.flat):
j=np.flatnonzero(y==i%len(nombres))[0];img=x[j]
if img.ndim==3 and img.shape[0] in [1,3]: img=np.moveaxis(img,0,-1)
ax.imshow(img.squeeze(),cmap='gray' if img.squeeze().ndim==2 else None)
ax.set_title(nombres[i%len(nombres)],fontsize=9);ax.axis('off')
fig.suptitle(titulo);fig.tight_layout();plt.show()
