The system transforms raw sensor data into feature vectors suitable for anomaly detection. Two parallel feature pipelines exist: legacy 1Hz features (6 dimensions) and batch 100Hz features (16 dimensions).
What it measures: Average voltage over the past hour (smooths out transient spikes).
# From calculator.py:23-58def calculate_voltage_rolling_mean( df: pd.DataFrame, evaluation_idx: int, window: str = WINDOW_DURATION # "1h") -> Optional[float]: # Get window data (past-only, including current point) # Approximate 1 hour = 60 points at 1 point/minute window_start = max(0, evaluation_idx - 59) # -59 to include 60 points total window_data = df['voltage_v'].iloc[window_start:evaluation_idx + 1] if len(window_data) < 2: # Need at least 2 points for meaningful mean return None # Calculate mean using vectorized Pandas mean_value = window_data.mean() return float(mean_value) if not pd.isna(mean_value) else None
What it measures: Normalized power factor (0-1 scale).
# From calculator.py:106-133def calculate_power_factor_efficiency_score( power_factor: float) -> Optional[float]: if power_factor is None or math.isnan(power_factor): return None # Clamp to valid range pf = max(0.0, min(1.0, power_factor)) # Direct mapping: PF is already 0-1, monotonic # Score = PF (linear, no transformation needed) # This preserves interpretability: score of 0.85 means PF was 0.85 return round(pf, 4)
What it measures: Interaction term capturing relationship between vibration and power factor.
# From detector.py:154-161if 'vibration_intensity_rms' in result.columns and 'power_factor_efficiency_score' in result.columns: result['power_vibration_ratio'] = ( result['vibration_intensity_rms'] / (result['power_factor_efficiency_score'] + 0.01) # Epsilon prevents division by zero )else: result['power_vibration_ratio'] = 0.0
Formula:power_vibration_ratio=PF_score+0.01vibration_rmsHealthy Range: 0.15 - 0.25 (depends on asset)
Why this derived feature?
High vibration combined with low power factor often indicates mechanical misalignment or bearing wear. This interaction term helps the model detect such patterns.
# From batch_features.py:51-95def extract_batch_features(raw_points: List[Dict[str, Any]]) -> Optional[Dict[str, float]]: """ Extract a 16-dimensional feature vector from a 1-second batch of raw points. Args: raw_points: List of 50-200 raw sensor dicts, each containing voltage_v, current_a, power_factor, vibration_g. Returns: Dict mapping feature name → float value, or None if batch too small. """ if not raw_points or len(raw_points) < 10: return None import numpy as np features: Dict[str, float] = {} for signal in SIGNAL_COLUMNS: # Extract signal values as NumPy array (vectorized) values = np.array( [p.get(signal, 0.0) for p in raw_points], dtype=np.float64, ) # Mean mean_val = float(np.mean(values)) features[f"{signal}_mean"] = mean_val # Standard Deviation (ddof=0 for population std, consistent with training) std_val = float(np.std(values, ddof=0)) features[f"{signal}_std"] = std_val # Peak-to-Peak (Max - Min) p2p_val = float(np.max(values) - np.min(values)) features[f"{signal}_peak_to_peak"] = p2p_val # RMS (Root Mean Square) rms_val = float(np.sqrt(np.mean(values ** 2))) features[f"{signal}_rms"] = rms_val return features
σ=n1i=1∑n(xi−μ)2What it captures: Variance/noise within the window. Critical for jitter detection.
Jitter Detection Key:A jitter fault has normal mean but high standard deviation. This is why the batch model (which includes std as a feature) achieves 99.6% F1, while the legacy model (which only sees the mean) achieves 78.1% F1.
Both pipelines use StandardScaler to normalize features before training:
# From detector.py:226self._scaler = StandardScaler()features_scaled = self._scaler.fit_transform(feature_matrix)
Why?
Features have different units (V vs A vs g)
StandardScaler ensures all features contribute equally to anomaly scoring
Formula: z=σx−μ
The scaler is fitted on healthy data only during calibration, then frozen. This ensures that “normal” is always defined by the original baseline, not drifting data.
Both pipelines return None (NaN) for incomplete windows:
# From calculator.py:53if len(window_data) < 2: # Need at least 2 points for meaningful mean return None# From batch_features.py:66if not raw_points or len(raw_points) < 10: return None
This prevents false zeros and allows the system to distinguish between “no data yet” and “zero value”.
Why is this important?
From ENGINEERING_LOG.md Phase 4:
NaN for Incomplete Windows: During the first hour of operation, there isn’t enough data for a 1-hour rolling mean. Instead of returning 0 (which would be falsely reassuring), we return None. This propagates as NaN in the feature vector and is explicitly handled downstream.
False zeros would cause the ML model to see abnormally low feature values during startup, triggering false alarms.
# From batch_features.py:113-139def extract_multi_window_features( raw_points: List[Dict[str, Any]], window_size: int = 100,) -> List[Dict[str, float]]: """ Slice a long stream of raw points into non-overlapping 1-second windows and extract batch features for each window. """ results: List[Dict[str, float]] = [] n = len(raw_points) for start in range(0, n - window_size + 1, window_size): window = raw_points[start : start + window_size] feat = extract_batch_features(window) if feat is not None: results.append(feat) return results
Used during calibration to convert historical 100Hz data into training feature rows.