Training Marine Engine Anomaly Detectors — A Physics-Informed ML Pipeline

The Problem with Synthetic Telemetry

Our first version generated synthetic telemetry using sinusoidal patterns plus Gaussian noise. It was a proof-of-concept — and it worked well enough to demonstrate that anomaly detection could flag faults. But it had a critical flaw: the ML models were trained on the same synthetic data they were tested against. This is the ML equivalent of studying the answer key. The models weren’t learning to detect real anomalies — they were learning the patterns of our simulator. We needed two things:

1. A real dataset — ground truth telemetry labeled with fault types and severity levels
2. Physics-based fault injection — simulators that model how engines actually degrade, not just how they wiggle

---

The Fault Dataset: Single Source of Truth

All ML training flows from a single CSV file: marine_engine_fault_dataset.csv. This is the single source of truth for our anomaly detection pipeline. Every model is trained on this dataset, validated against it, and its performance is measured against it.

Schema

{
  "columns": {
    "timestamp": "ISO8601 datetime",
    "engine_id": "String (e.g., 'DE-1234')",
    "rpm": "Float, 0-2000",
    "fuel_flow_lph": "Float, liters per hour",
    "exhaust_temp_c": "Float, Celsius",
    "cylinder_pressure_bar": "Float, bar",
    "coolant_temp_c": "Float, Celsius",
    "oil_pressure_bar": "Float, bar",
    "vibration_mm_s": "Float, mm/s",
    "boost_pressure_bar": "Float, bar",
    "fuel_consumption_kg_h": "Float, kg/hour",
    "nox_ppm": "Float, ppm"
  },
  "labels": {
    "fault_type": "One of: bearing_wear, injector_clog, piston_ring_wear, coolant_leak, turbo_degradation, combustion_anomaly, oil_degradation",
    "severity": "none | mild | moderate | severe | critical",
    "fault_age_hours": "Hours since fault onset"
  }

Generation

The dataset is generated by a physics-based simulator that models each fault type with its own degradation curve. We don’t just shift sensor values randomly — we model how each component degrades over time:

• Bearing wear: Linear wear + exponential acceleration (fatigue spalling)
• Injector clog: Sigmoidal flow decay (particulate accumulation)
• Coolant leak: Wiener process (random walk with drift)
• Turbo degradation: Exponential efficiency decay
• Combustion anomalies: Quadratic misfire rate increase

This produces realistic telemetry where fault signatures evolve naturally over thousands of operating hours. The final dataset contains 10,000 labeled readings across 7 fault types and 5 severity levels.

---

The Training Pipeline

Our ML pipeline is a deterministic, reproducible sequence of steps. Every stage includes validation checks — if anything fails, training is blocked.

Step 1: Schema Validation

Before any training happens, the dataset is validated against the schema:

• All required columns present with correct types
• Range validation (e.g., RPM between 0-2000)
• Label integrity (valid fault types, valid severity values)
• Distribution checks (mean and std within expected ranges)
• Temporal consistency (monotonically increasing timestamps per engine)
• Minimum sample count (≥ 10 samples per fault type)

def validate_dataset(self) -> List[str]:
    """Validate the fault dataset against the schema."""
    if self._df is None:
        self.load_dataset()
    if self._schema is None:
        self.load_schema()

    warnings = []

    # Check required columns
    required_cols = set(self._schema["columns"].keys()) | set(self._schema["labels"].keys())
    missing_cols = required_cols - set(self._df.columns)
    if missing_cols:
        raise DatasetValidationError(f"Missing required columns: {missing_cols}")

    # Range validation
    for col, col_spec in self._schema["columns"].items():
        if col_spec["type"] == "float":
            if "min" in col_spec:
                below_min = (self._df[col] < col_spec["min"]).sum()
                if below_min > 0:
                    warnings.append(f"Column {col}: {below_min} values below min")
            if "max" in col_spec:
                above_max = (self._df[col] > col_spec["max"]).sum()
                if above_max > 0:
                    warnings.append(f"Column {col}: {above_max} values above max")

    return warnings

Step 2: Deterministic Train/Test Split

We use a stratified split that preserves the fault type distribution in both sets. The split is deterministic — same seed, same split, every time.

def split_data(self, test_ratio: float = 0.2, random_seed: int = 42):
    train_dfs = []
    test_dfs = []

    for fault_type, group in self._df.groupby("fault_type"):
        n_test = max(1, int(len(group) * test_ratio))
        group_shuffled = group.sample(frac=1, random_state=random_seed)
        test_dfs.append(group_shuffled.iloc[:n_test])
        train_dfs.append(group_shuffled.iloc[n_test:])

    train_df = pd.concat(train_dfs, ignore_index=True)
    test_df = pd.concat(test_dfs, ignore_index=True)

The split manifest is saved to disk for reproducibility audits.

Step 3: Feature Engineering

We use two feature sets: Standard features (direct sensor readings):

rpm, fuel_flow_lph, exhaust_temp_c, cylinder_pressure_bar,
coolant_temp_c, oil_pressure_bar, vibration_mm_s,
boost_pressure_bar, fuel_consumption_kg_h, nox_ppm

Physics-informed features (derived from domain knowledge):

Feature	Formula	What It Captures
specific_fuel_consumption	fuel_consumption_kg_h / (rpm * fuel_flow_lph / 1000)	Fuel efficiency — higher when combustion is poor
combustion_efficiency_proxy	cylinder_pressure_bar / (exhaust_temp_c + 1) * 100	Quality of combustion — high pressure + moderate temp = good
thermal_stress	exhaust_temp_c - coolant_temp_c	Thermal gradient across engine
lubrication_quality_index	oil_pressure_bar / (coolant_temp_c / 100 + 1)	Oil pressure normalized by temperature
turbo_efficiency_proxy	boost_pressure_bar / (exhaust_temp_c / 100 + 1)	Turbo performance — high boost + moderate EGT = good

These features encode marine engineering domain knowledge that the raw sensors don’t directly expose. A clogged injector, for instance, reduces cylinder pressure and raises exhaust temperature — the combustion efficiency proxy captures this interaction directly.

Step 4: Standardization

All features are standardized using StandardScaler fitted on training data only:

scaler = StandardScaler()
X_train = scaler.fit_transform(train_features)
X_test = scaler.transform(test_features)  # transform, NOT fit

This ensures no data leakage from the test set.

Step 5: Isolation Forest Training

We use scikit-learn’s IsolationForest — an unsupervised anomaly detection algorithm that isolates anomalies by randomly selecting a feature and then randomly selecting a split value between the maximum and minimum values of the selected feature.

model = IsolationForest(
    contamination=0.15,    # expect ~15% anomalies in training data
    n_estimators=200,
    max_samples="auto",
    random_state=42,
    n_jobs=-1,
)
model.fit(X_train)

Why Isolation Forest? It works well for high-dimensional data, doesn’t assume any distribution, and is computationally efficient at inference time. For our use case — detecting novel failure modes we haven’t seen before — unsupervised learning is essential.

Step 6: OOD Statistics Computation

After training, we compute statistics from the training data to enable out-of-distribution (OOD) detection at inference time:

ood_stats = OODStats(
    feature_means=scaler.mean_.tolist(),
    feature_stds=scaler.scale_.tolist(),
    mahalanobis_threshold=6.0,
    training_feature_count=X_train.shape[1],
)

These statistics are saved alongside the model and used at inference to flag when telemetry falls outside the training distribution.

---

Out-of-Distribution Detection

Here’s a critical insight: an anomaly detector can only detect anomalies it has seen patterns for. If the telemetry at inference time is fundamentally different from what the model was trained on, the predictions are unreliable. We solve this with Mahalanobis distance — a measure of how many standard deviations a point is from the training distribution, accounting for correlations between features.

def detect_ood(self, telemetry: Dict[str, float]) -> tuple:
    scaled = self._scaler.transform(features.reshape(1, -1))
    diff = scaled - means
    inv_stds = 1.0 / (stds + 1e-8)
    mahalanobis = np.sqrt(np.sum((diff * inv_stds) ** 2))

    is_ood = mahalanobis > self._threshold
    confidence_reduction = min(0.5, excess / threshold * 0.5) if is_ood else 0.0

When OOD is detected:

1. The anomaly confidence score is reduced by up to 50%
2. The reading is flagged for operator review
3. Detection still runs — we don’t skip it

This prevents the system from confidently making wrong predictions on unfamiliar data.

---

Model Registry & Deployment

Every trained model is registered with full metadata:

{
  "model_id": "anomaly_engine",
  "version": "2.0",
  "trained_on_dataset": "v1.0",
  "training_date": "2025-07-01T00:00:00Z",
  "metrics": {
    "isolation_forest_auc": 0.94,
    "precision": 0.91,
    "recall": 0.89,
    "f1": 0.90
  },
  "artifacts": {
    "isolation_forest": "isolation_forest_v2.0.joblib",
    "scaler": "scaler_v2.0.joblib",
    "ood_stats": "ood_stats_v2.0.json"
  },
  "is_deployed": true
}

Only one version is deployed at a time. Deploying a new version automatically undeploys the previous one. The registry is persisted as JSON and loaded on system startup.

---

3-Layer Anomaly Detection

At inference time, every telemetry reading passes through three detection layers:

Layer 1: Threshold-Based (Instantaneous)

Engineering thresholds from ISO 10816 and OEM specifications. Fast, interpretable, and always reliable for known fault patterns.

THRESHOLD_RULES = {
    "vibration_mm_s": {"warning": 11.2, "critical": 25.0},
    "exhaust_temp_c": {"warning": 550, "critical": 650},
    "oil_pressure_bar": {"warning": 1.5, "critical": 1.0},  # inverted: lower is worse
    ...
}

Layer 2: Isolation Forest (Statistical)

The pre-trained model scores each reading. More negative decision function = more anomalous.

Layer 3: Trajectory Analysis (Temporal)

A sliding window over recent readings detects:

• Stuck sensors: No change over N readings
• Sudden spikes: Value changes > 3 standard deviations from recent trend
• Trend violations: Unexpected direction changes

Scoring

The final anomaly score combines all layers:

combined = 0.4 * threshold_score + 0.4 * isolation_score + 0.2 * trajectory_score
combined *= (1.0 - ood_confidence_reduction)

---

Retraining Pipeline

Models are retrained automatically when:

1. Scheduled: Every 90 days
2. Data threshold: New labeled data exceeds 20% of current dataset
3. Performance: F1 drops below 0.80
4. Drift: Feature distribution drifts beyond 2 standard deviations
5. Manual: Operator triggers retraining

Each retraining runs the full validation pipeline. If any check fails, the model is not deployed and the previous version remains active.

---

Results

Metric	Value
Isolation Forest AUC	0.94
Precision	0.91
Recall	0.89
F1 Score	0.90
OOD Calibration	0.95
Training samples	8,000
Validation samples	2,000
Test suite	51 tests, all passing

---

What’s Next

• Auto-calibration: Learn degradation parameters from operational data instead of literature
• Multi-engine correlation: Detect fleet-wide defects
• Maintenance event modeling: Reset degradation after oil changes, filter replacements, etc.
• Active learning: Prioritize labeling of uncertain predictions
• Online learning: Gradually adapt to new fault patterns without full retraining

---