Building a Predictive Maintenance Platform from Scratch

The problem with reactive maintenance

Most manufacturers still run on reactive maintenance: equipment breaks, production stops, someone calls the repair crew. The costs are brutal — unplanned downtime runs $260,000 per hour in automotive, and the global tab for reactive maintenance is somewhere north of $400 billion annually.

The alternative — preventive maintenance — is better but blunt. You schedule maintenance every 500 hours or every quarter whether the machine needs it or not. You're throwing labor and parts at equipment that might be fine, while the machines that are actually degrading slip through the cracks between schedules.

Predictive maintenance is the upgrade: monitor the machine continuously, detect anomalies in sensor data before they become failures, and schedule maintenance only when the data says to. That's what PredictiveIQ does.

Here's how I built it.

Architecture overview

The stack is deliberately simple. I wanted something a small manufacturer could actually run on their existing hardware, not a cloud-first system requiring three AWS services and a data engineering team.

  ┌──────────────────┐     ┌──────────────────┐     ┌──────────────────┐
  │  Next.js 14      │────▶│  FastAPI          │────▶│  SQLite /        │
  │  Dashboard       │     │  Backend          │     │  TimescaleDB     │
  │  (Tailwind +     │     │  :8000            │     │                  │
  │   Recharts)      │     │  ML Pipeline      │     │  ML Models       │
  │  :3000           │     │  (scikit-learn)   │     │  (joblib)        │
  └──────────────────┘     └──────────────────┘     └──────────────────┘
                                    │
                            ┌───────┴────────┐
                            │  Claude API    │
                            │  (LLM advisor) │
                            └────────────────┘
        

• FastAPI backend — async Python, handles sensor ingestion, ML training/scoring, and LLM calls
• SQLite in dev, TimescaleDB-ready in prod — TimescaleDB is PostgreSQL with time-series superpowers; the schema works on both
• scikit-learn Isolation Forest — per-asset anomaly detection, models stored with joblib
• Claude API — explains anomalies in plain English with a template fallback when the key isn't set
• Next.js 14 dashboard — fleet overview, equipment detail pages, alert feed
• Docker Compose — one command to run everything

Data model and sensor ingestion

There are two core entities: Equipment and SensorReading. Equipment records represent physical assets — pumps, motors, conveyors, compressors. SensorReadings are the time-series measurements coming off those assets.

from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey, Enum
from sqlalchemy.orm import relationship
from datetime import datetime
import enum

class SensorType(str, enum.Enum):
    vibration = "vibration"
    temperature = "temperature"
    current = "current"

class Equipment(Base):
    __tablename__ = "equipment"

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    equipment_type = Column(String)          # pump, motor, conveyor, etc.
    location = Column(String)
    created_at = Column(DateTime, default=datetime.utcnow)

    readings = relationship("SensorReading", back_populates="equipment")
    alerts = relationship("Alert", back_populates="equipment")

class SensorReading(Base):
    __tablename__ = "sensor_readings"

    id = Column(Integer, primary_key=True)
    equipment_id = Column(Integer, ForeignKey("equipment.id"), nullable=False)
    sensor_type = Column(Enum(SensorType), nullable=False)
    value = Column(Float, nullable=False)
    timestamp = Column(DateTime, default=datetime.utcnow, index=True)

    equipment = relationship("Equipment", back_populates="readings")

The ingestion endpoint accepts a batch of readings — you don't want to open an HTTP connection per sensor sample.

@router.post("/sensors", response_model=list[SensorReadingOut])
async def ingest_sensor_readings(
    readings: list[SensorReadingCreate],
    db: AsyncSession = Depends(get_db),
):
    db_readings = [SensorReading(**r.model_dump()) for r in readings]
    db.add_all(db_readings)
    await db.commit()

    # After ingestion, check for anomalies on affected equipment
    equipment_ids = {r.equipment_id for r in readings}
    for eq_id in equipment_ids:
        await check_and_alert(eq_id, db)

    return db_readings

That check_and_alert call is where it gets interesting.

Feature engineering

Raw sensor values aren't very useful for anomaly detection. A vibration reading of 2.3 mm/s means nothing without context — is that high for this machine? Is it trending up? The signal that matters is the shape of the data over time.

For each sensor type, I compute four rolling features over a window of 20 readings:

• Rolling mean — the baseline signal level. Drifting mean often signals wear.
• Rolling standard deviation — signal variability. High std dev on vibration = imbalance, looseness, or bearing degradation.
• Rate of change — first difference between consecutive readings. Sudden spikes matter even when the rolling stats look normal.
• Latest value — the raw reading itself, as a feature.

import pandas as pd
import numpy as np
from typing import Sequence

WINDOW = 20

def compute_features(readings: Sequence[float]) -> dict[str, float] | None:
    """
    Compute rolling features from a sequence of raw sensor values.
    Returns None if there's not enough data to fill a window.
    """
    if len(readings) < WINDOW:
        return None

    series = pd.Series(readings)

    rolling_mean = series.rolling(WINDOW).mean().iloc[-1]
    rolling_std  = series.rolling(WINDOW).std().iloc[-1]
    rate_of_change = series.diff().iloc[-1]
    latest = series.iloc[-1]

    return {
        "rolling_mean": rolling_mean,
        "rolling_std": rolling_std if not np.isnan(rolling_std) else 0.0,
        "rate_of_change": rate_of_change if not np.isnan(rate_of_change) else 0.0,
        "latest_value": latest,
    }

These four features are computed per sensor type (vibration, temperature, current), giving the model a 12-dimensional feature vector per equipment check — enough signal to detect meaningful anomalies without overfitting to noise.

Anomaly detection with Isolation Forest

I chose Isolation Forest for a few reasons specific to manufacturing:

1. Unsupervised — you rarely have labeled failure data. Isolation Forest learns what "normal" looks like and flags deviations, no failure labels required.
2. Per-asset training — a pump and a conveyor have completely different normal signatures. Training one global model would be useless. Isolation Forest is cheap enough to train per-asset.
3. Interpretable contamination parameter — setting contamination=0.05 means "expect roughly 5% anomalies in training data." You can tune this per asset type based on domain knowledge.
4. Fast inference — scoring a new reading takes microseconds. No latency problem when checking after every batch ingestion.

import numpy as np
import joblib
from pathlib import Path
from sklearn.ensemble import IsolationForest
from sqlalchemy.ext.asyncio import AsyncSession

from app.services.features import compute_features

MODEL_DIR = Path("models")
MODEL_DIR.mkdir(exist_ok=True)

def model_path(equipment_id: int) -> Path:
    return MODEL_DIR / f"equipment_{equipment_id}.joblib"

async def train_model(equipment_id: int, db: AsyncSession) -> dict:
    """
    Fetch historical readings for this equipment, compute features,
    and train a fresh Isolation Forest. Saves the model to disk.
    """
    readings_by_type = await fetch_readings_grouped(equipment_id, db)

    feature_matrix = []
    for sensor_type, values in readings_by_type.items():
        feats = compute_features(values)
        if feats:
            feature_matrix.append(list(feats.values()))

    if not feature_matrix:
        return {"status": "insufficient_data"}

    X = np.array(feature_matrix)

    clf = IsolationForest(
        n_estimators=100,
        contamination=0.05,   # expect ~5% anomalies in historical data
        random_state=42,
    )
    clf.fit(X)

    joblib.dump(clf, model_path(equipment_id))
    return {"status": "trained", "samples": len(X)}

def score_reading(equipment_id: int, feature_vector: list[float]) -> float:
    """
    Returns the raw anomaly score from Isolation Forest.
    More negative = more anomalous. Typically in range [-0.5, 0.5].
    """
    path = model_path(equipment_id)
    if not path.exists():
        return 0.0   # no model yet, can't score

    clf = joblib.load(path)
    X = np.array([feature_vector])
    score = clf.score_samples(X)[0]   # lower = more anomalous
    return float(score)

Health scoring

The raw Isolation Forest anomaly score is a float in roughly [-0.5, 0.5]. That's not useful to show an operator. I map it to a 0-100 health score with three status zones:

def anomaly_score_to_health(raw_score: float) -> int:
    """
    Map Isolation Forest raw score to a 0-100 health score.
    IF score: higher = more normal, lower = more anomalous.
    Typical range: [-0.5, 0.5]. We clamp and invert to get health.
    """
    # Clamp to expected range
    clamped = max(-0.5, min(0.5, raw_score))

    # Normalize to [0, 1]: 0.5 (normal) → 1.0, -0.5 (anomalous) → 0.0
    normalized = (clamped + 0.5) / 1.0

    # Scale to 0-100
    return round(normalized * 100)

def health_to_status(score: int) -> str:
    if score >= 75:
        return "healthy"
    elif score >= 50:
        return "warning"
    else:
        return "critical"

When a score drops below the warning threshold, the system automatically creates an Alert record and the dashboard surfaces it. The alert lifecycle is active → acknowledged → resolved, matching how maintenance teams actually work.

The LLM advisor

Here's where this goes from "ML dashboard" to something actually useful. A health score of 42 tells a maintenance manager something is wrong. It doesn't tell them what is wrong or what to do about it.

The LLM advisor takes the current sensor readings, computed features, health score, and equipment type — and asks Claude to explain what's happening in plain English and suggest a next action.

import anthropic
from app.config import settings

SYSTEM_PROMPT = """You are an industrial maintenance advisor. You analyze sensor data
from manufacturing equipment and explain anomalies in plain English.

When given sensor readings and an anomaly score, you:
1. Identify the most likely cause based on the sensor signatures
2. Explain what it means for the machine in non-technical language
3. Recommend a specific next action

Be concise. Maintenance technicians need actionable information, not lectures.
Respond in 3-4 sentences max."""

async def explain_anomaly(
    equipment_name: str,
    equipment_type: str,
    sensor_summary: dict,
    health_score: int,
) -> str:
    if not settings.ANTHROPIC_API_KEY:
        return _template_fallback(health_score, sensor_summary)

    client = anthropic.AsyncAnthropic(api_key=settings.ANTHROPIC_API_KEY)

    user_message = f"""
Equipment: {equipment_name} ({equipment_type})
Health score: {health_score}/100

Sensor readings:
- Vibration: mean={sensor_summary['vibration_mean']:.2f}, std={sensor_summary['vibration_std']:.2f}
- Temperature: mean={sensor_summary['temp_mean']:.1f}°F, std={sensor_summary['temp_std']:.2f}
- Current: mean={sensor_summary['current_mean']:.1f}A, std={sensor_summary['current_std']:.2f}

What's likely happening and what should the technician do?
"""

    message = await client.messages.create(
        model=settings.LLM_MODEL,
        max_tokens=256,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_message}],
    )

    return message.content[0].text

def _template_fallback(health_score: int, sensor_summary: dict) -> str:
    """Used when no API key is configured — keeps the UI functional in demos."""
    status = "critical" if health_score < 50 else "warning"
    return (
        f"Equipment health is {status} at {health_score}/100. "
        f"Elevated vibration readings (mean: {sensor_summary['vibration_mean']:.2f}) "
        f"suggest possible bearing wear or imbalance. "
        f"Recommend scheduling an inspection and checking lubrication levels."
    )

Wiring it up with FastAPI

The ML endpoints follow a clear pattern: train, score, explain. They're all under /api/ml.

@router.post("/ml/train")
async def train(equipment_id: int, db: AsyncSession = Depends(get_db)):
    """Train or retrain the anomaly model for a specific asset."""
    result = await train_model(equipment_id, db)
    return result

@router.get("/ml/health/{equipment_id}")
async def get_health(equipment_id: int, db: AsyncSession = Depends(get_db)):
    """Compute current health score for an asset."""
    readings = await fetch_recent_readings(equipment_id, db)
    features = build_feature_vector(readings)
    raw_score = score_reading(equipment_id, features)
    health = anomaly_score_to_health(raw_score)
    return {
        "equipment_id": equipment_id,
        "health_score": health,
        "status": health_to_status(health),
        "anomaly_score": raw_score,
    }

@router.post("/ml/explain")
async def explain(req: ExplainRequest, db: AsyncSession = Depends(get_db)):
    """Get an LLM-powered explanation of current equipment status."""
    equipment = await get_equipment(req.equipment_id, db)
    sensor_summary = await build_sensor_summary(req.equipment_id, db)
    explanation = await explain_anomaly(
        equipment.name,
        equipment.equipment_type,
        sensor_summary,
        req.health_score,
    )
    return {"explanation": explanation}

The whole backend starts with a standard FastAPI app entry point. One thing worth noting: I use lifespan rather than the deprecated on_event handlers for database initialization.

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.database import init_db
from app.routers import equipment, sensors, alerts, ml
from app.config import settings

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()   # create tables on startup
    yield

app = FastAPI(title="PredictiveIQ API", lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(equipment.router, prefix="/api")
app.include_router(sensors.router, prefix="/api")
app.include_router(alerts.router, prefix="/api")
app.include_router(ml.router, prefix="/api")

The Next.js dashboard

The frontend has three views:

• Fleet overview (/) — summary cards with fleet health stats, equipment table with color-coded health bars
• Equipment detail (/equipment/[id]) — live sensor charts via Recharts, health trend, alert feed, and the AI advisor panel
• Alert feed (/alerts) — filterable list of all alerts with acknowledge/resolve actions

The AI advisor component is the most interesting UI piece. It sits on the equipment detail page and calls /api/ml/explain on demand:

'use client'
import { useState } from 'react'

export function AiAdvisor({ equipmentId, healthScore }: Props) {
  const [explanation, setExplanation] = useState<string | null>(null)
  const [loading, setLoading] = useState(false)

  async function analyze() {
    setLoading(true)
    try {
      const res = await fetch(`/api/ml/explain`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ equipment_id: equipmentId, health_score: healthScore }),
      })
      const data = await res.json()
      setExplanation(data.explanation)
    } finally {
      setLoading(false)
    }
  }

  return (
    <div className="rounded-xl border border-zinc-800 p-5">
      <div className="text-xs font-mono text-purple-400 uppercase tracking-widest mb-3">
        AI Advisor
      </div>
      {explanation ? (
        <p className="text-sm text-zinc-300 leading-relaxed">{explanation}</p>
      ) : (
        <button onClick={analyze} disabled={loading}
          className="text-sm px-4 py-2 bg-purple-500/10 text-purple-400 rounded-lg">
          {loading ? 'Analyzing...' : 'Analyze current status'}
        </button>
      )}
    </div>
  )
}

What I'd do differently

A few things I'd change building this again:

Start with TimescaleDB, not SQLite. The schema migration from SQLite to TimescaleDB is trivial, but the operational story is cleaner if you just start with Postgres from day one. TimescaleDB's automatic partitioning on the timestamp column matters once you're ingesting readings every few seconds across dozens of machines.

Add a model versioning layer. Right now the model is just a joblib file. That works, but when you retrain (which you should do periodically as the machine ages), you lose the ability to compare model versions or roll back if a new model starts producing bad scores. Something like MLflow is overkill for an MVP, but even a simple timestamp-tagged directory would help.

Per-sensor-type models, not per-asset. I train one model per asset using all sensor types together. In practice, vibration anomalies and temperature anomalies have different signatures and different failure modes. Training separate models per sensor type per asset gives you sharper detection and better explainability — you know which sensor triggered the anomaly, not just that something's wrong.

The LLM call should be async and cached. Right now the explain endpoint is blocking and called on demand. For a production system, I'd precompute explanations whenever the health score drops below a threshold, cache the result, and serve it instantly from the UI. LLM latency is noticeable on the shop floor.

The full source is private (it's a commercial project), but the architecture here is complete enough to build from. If you have questions about any of the implementation details, reach out or find me on LinkedIn.

Next in this series: RAG for Industrial Documentation — how I built The Guide, including chunking strategies for technical manuals and why BYOM architecture matters for manufacturing customers.