""" train_model.py Entrena el modelo PCA + Mahalanobis usando FEATURES DE VENTANA, no muestras individuales. Por qué features de ventana: - Una muestra individual de una señal sinusoidal oscila entre -1 y +1. Su distancia de Mahalanobis sube y baja con la fase de la onda, produciendo falsos positivos aunque la máquina esté sana. - En cambio, las estadísticas de una ventana de 200 muestras (RMS, std, peak, etc.) son estables durante operación normal y cambian claramente cuando hay una falla. Features extraídas por cada ventana (4 sensores × 6 features = 24 features): - RMS : energía de la señal (detecta amplitud anormal) - Std : variabilidad (detecta ruido anormal) - Peak : valor absoluto máximo (detecta impactos) - Kurtosis : colas de distribución (detecta impactos súbitos) - Skewness : asimetría (detecta deriva) - Mean : nivel DC (detecta deriva de temperatura/corriente) Ejecutar una vez antes de correr main.py: python train_model.py Genera: model.pkl """ import numpy as np from sklearn.preprocessing import StandardScaler from sklearn.decomposition import PCA from scipy.stats import kurtosis, skew import joblib np.random.seed(42) # ── Parámetros ───────────────────────────────────────────────────────────── FREQ = 50.0 # Hz — frecuencia fundamental DT = 1 / 1000 # paso de tiempo (1 ms) N_WIN = 200 # muestras por ventana N_WINDOWS = 1000 # ventanas de entrenamiento (= 200,000 muestras de señal normal) N_PCA = 6 # componentes PCA a retener # ── Generador de señal ───────────────────────────────────────────────────── def generate_window(step_idx, fault_intensity=0.0, fault_type="bearing"): t = np.arange(N_WIN) * DT + step_idx * N_WIN * DT vib_x = 1.0 * np.sin(2*np.pi*FREQ*t) + np.random.normal(0, 0.05, N_WIN) vib_y = 0.8 * np.sin(2*np.pi*FREQ*t + 0.3) + np.random.normal(0, 0.05, N_WIN) temp = 0.3 * np.sin(2*np.pi*0.1 *t) + np.random.normal(0, 0.02, N_WIN) current = 0.6 * np.sin(2*np.pi*FREQ*t + 0.1) + np.random.normal(0, 0.04, N_WIN) if fault_intensity > 0: if fault_type == "bearing": ff = FREQ * 3.5 vib_x += fault_intensity*1.5*np.sin(2*np.pi*ff*t) \ + fault_intensity*np.random.normal(0, 0.3, N_WIN) vib_y += fault_intensity*1.2*np.sin(2*np.pi*ff*t+0.5) \ + fault_intensity*np.random.normal(0, 0.25, N_WIN) temp += fault_intensity * 0.8 current += fault_intensity*0.4*np.sin(2*np.pi*ff*t) elif fault_type == "imbalance": vib_x += fault_intensity*2.0*np.sin(2*np.pi*FREQ*t) vib_y += fault_intensity*2.0*np.sin(2*np.pi*FREQ*t + np.pi/2) temp += fault_intensity * 0.3 current += fault_intensity * 0.6 elif fault_type == "looseness": vib_x += fault_intensity*1.0*np.sin(2*np.pi*FREQ*0.5*t) \ + fault_intensity*np.random.choice([-1,1],N_WIN) \ *np.random.exponential(0.3, N_WIN) vib_y += fault_intensity*0.8*np.sin(2*np.pi*FREQ*0.5*t) current += fault_intensity * 0.3 return np.column_stack([vib_x, vib_y, temp, current]) # ── Extracción de features por ventana ──────────────────────────────────── def extract_features(window): """ Entrada: window shape (N_WIN, 4) Salida: features shape (24,) — 6 features × 4 sensores """ feats = [] for col in range(window.shape[1]): s = window[:, col] feats.append(np.sqrt(np.mean(s**2))) # RMS feats.append(np.std(s)) # Std feats.append(np.max(np.abs(s))) # Peak feats.append(float(kurtosis(s))) # Kurtosis feats.append(float(skew(s))) # Skewness feats.append(np.mean(s)) # Mean return np.array(feats) # ── Generar features de entrenamiento (operación normal) ─────────────────── print(f"Generando {N_WINDOWS} ventanas de operación normal...") X_train = np.array([ extract_features(generate_window(i, fault_intensity=0.0)) for i in range(N_WINDOWS) ]) print(f"Matriz de features: {X_train.shape} ({N_WINDOWS} ventanas × 24 features)") # ── Pipeline: Scaler → PCA ───────────────────────────────────────────────── scaler = StandardScaler() X_scaled = scaler.fit_transform(X_train) pca = PCA(n_components=N_PCA) X_pca = pca.fit_transform(X_scaled) ev = pca.explained_variance_ratio_ print(f"\nVarianza explicada por componente: {ev.round(3)}") print(f"Varianza acumulada ({N_PCA} PCs): {ev.sum():.3f}") # ── Parámetros de Mahalanobis en espacio PCA ─────────────────────────────── mu = X_pca.mean(axis=0) cov = np.cov(X_pca, rowvar=False) cov_inv = np.linalg.inv(cov) dists = np.array([ np.sqrt((x - mu) @ cov_inv @ (x - mu)) for x in X_pca ]) threshold_2sigma = dists.mean() + 2 * dists.std() threshold_3sigma = dists.mean() + 3 * dists.std() print(f"\nDistancias de Mahalanobis en entrenamiento:") print(f" Media: {dists.mean():.4f} | Std: {dists.std():.4f}") print(f" Umbral 2σ: {threshold_2sigma:.4f}") print(f" Umbral 3σ: {threshold_3sigma:.4f}") print(f" Falsos positivos esperados (>2σ): {(dists > threshold_2sigma).mean()*100:.1f}%") print(f" Falsos positivos esperados (>3σ): {(dists > threshold_3sigma).mean()*100:.1f}%") # ── Guardar modelo ───────────────────────────────────────────────────────── model = { "scaler": scaler, "pca": pca, "mu": mu, "cov_inv": cov_inv, "threshold_2sigma": threshold_2sigma, "threshold_3sigma": threshold_3sigma, "dist_mean": float(dists.mean()), "dist_std": float(dists.std()), "n_features": 24, # features por ventana "n_raw_sensors": 4, "n_components": N_PCA, "n_window": N_WIN, "freq": FREQ, "dt": DT, "explained_variance": ev, "feature_names": [ f"{s}_{f}" for s in ["vib_x","vib_y","temp","curr"] for f in ["rms","std","peak","kurtosis","skewness","mean"] ], } joblib.dump(model, "model.pkl") print("\n✓ model.pkl guardado") print(" → Ahora el modelo opera sobre features de ventana, no muestras individuales.") print(" → El score será estable en operación normal y subirá claramente con fallas.")