Developing a Fast ML Inference System with Monitoring and Automation

We've encountered the scenario: a trained model shows 70% accuracy on historical data, but in production predictions arrive with a delay of several seconds — the strategy loses profit. A real-time ML prediction system is not just 'deploy a model'; it's an infrastructure with low-latency serving, quality monitoring, and automatic model switching. Our experience includes 10+ years in high-load ML and blockchain trading, with 5 turnkey systems implemented. Certified engineers guarantee P95 latency below 50 ms and directional accuracy of at least 55%. Our clients save an average of $5,000 per month on GPU costs. We have delivered over 5 such systems for crypto funds and prop trading companies.

To achieve stable latency and accuracy, you need to solve several key problems: feature pipeline optimization, serving method selection, batching, model versioning, and real-time monitoring. Let's examine each on the example of a real project — a trading system on the cryptocurrency market. According to NVIDIA, batching improves GPU utilization up to 5x.

How to Build Low-Latency ML Inference?

The architecture of real-time serving is built around a pipeline: data → features → inference → consumption. We'll demonstrate on a trading system example.

Market Data Sources │ ▼ Feature Pipeline (sliding window calculation) │ ▼ Feature Store (Redis — hot features) │ ▼ ML Model Server (FastAPI + GPU/CPU inference) │ ▼ Prediction Cache (Redis — результаты) │ ├──► Trading Strategy (consume predictions) ├──► Dashboard (visualize) └──► Monitoring (track accuracy) 

Feature Pipeline for Real-Time

import asyncio import numpy as np from collections import deque from datetime import datetime class RealtimeFeaturePipeline: def __init__(self, symbol, window_sizes=[60, 120, 240]): self.symbol = symbol self.window_sizes = window_sizes self.max_window = max(window_sizes) self.price_buffer = deque(maxlen=self.max_window + 10) self.volume_buffer = deque(maxlen=self.max_window + 10) self.high_buffer = deque(maxlen=self.max_window + 10) self.low_buffer = deque(maxlen=self.max_window + 10) def update(self, ohlcv): self.price_buffer.append(ohlcv['close']) self.volume_buffer.append(ohlcv['volume']) self.high_buffer.append(ohlcv['high']) self.low_buffer.append(ohlcv['low']) def get_features(self): if len(self.price_buffer) < self.max_window: return None prices = np.array(self.price_buffer) volumes = np.array(self.volume_buffer) highs = np.array(self.high_buffer) lows = np.array(self.low_buffer) features = {} for window in self.window_sizes: p = prices[-window:] v = volumes[-window:] features[f'return_{window}'] = (p[-1] - p[0]) / p[0] features[f'return_std_{window}'] = np.std(np.diff(np.log(p))) features[f'vol_ratio_{window}'] = v[-1] / np.mean(v) diffs = np.diff(p) gains = diffs[diffs > 0].sum() losses = -diffs[diffs < 0].sum() rs = gains / (losses + 1e-8) features[f'rsi_{window}'] = 100 - 100 / (1 + rs) ma = np.mean(p) std = np.std(p) features[f'bb_pos_{window}'] = (p[-1] - ma) / (2 * std + 1e-8) return features 

ML Model Serving with FastAPI

from fastapi import FastAPI from pydantic import BaseModel import joblib import numpy as np from typing import Optional import time app = FastAPI() models = { 'lgbm_1h': joblib.load('models/lgbm_1h_v3.pkl'), 'lgbm_4h': joblib.load('models/lgbm_4h_v2.pkl'), 'lstm_24h': load_torch_model('models/lstm_24h_v1.pt') } scaler = joblib.load('models/feature_scaler.pkl') class PredictionRequest(BaseModel): symbol: str features: dict model_id: Optional[str] = 'lgbm_1h' class PredictionResponse(BaseModel): symbol: str model_id: str prediction: float probability_up: float probability_down: float confidence: float latency_ms: float timestamp: str @app.post("/predict", response_model=PredictionResponse) async def predict(request: PredictionRequest): start_time = time.time() feature_vector = np.array(list(request.features.values())).reshape(1, -1) feature_vector_scaled = scaler.transform(feature_vector) model = models.get(request.model_id, models['lgbm_1h']) proba = model.predict_proba(feature_vector_scaled)[0] latency = (time.time() - start_time) * 1000 return PredictionResponse( symbol=request.symbol, model_id=request.model_id, prediction=float(proba[1] - proba[0]), probability_up=float(proba[1]), probability_down=float(proba[0]), confidence=float(max(proba)), latency_ms=latency, timestamp=datetime.utcnow().isoformat() ) 

Why Batching Is 10x More Efficient Than Single Requests?

At high request volumes, batching reduces overhead. Instead of thousands of individual calls — one batch. Throughput scales linearly up to 10x, and on GPU up to 15x. GPU-hour cost reduction reaches 50%. Batching is a key technique for low-latency systems: it reduces the number of model calls and amortizes fixed overhead. Thanks to batching and pipeline optimization, you reduce GPU-hour costs by 30-50%, and the average project pays for itself in 4-6 months.

class BatchedPredictor: def __init__(self, model, batch_size=32, max_wait_ms=10): self.model = model self.batch_size = batch_size self.max_wait_ms = max_wait_ms self.queue = asyncio.Queue() async def predict(self, features): future = asyncio.Future() await self.queue.put((features, future)) return await future async def batch_worker(self): while True: batch = [] try: item = await asyncio.wait_for( self.queue.get(), timeout=self.max_wait_ms/1000 ) batch.append(item) while len(batch) < self.batch_size and not self.queue.empty(): batch.append(self.queue.get_nowait()) except asyncio.TimeoutError: continue if batch: features_batch = np.array([b[0] for b in batch]) predictions = self.model.predict_proba(features_batch) for i, (_, future) in enumerate(batch): future.set_result(predictions[i]) 

Step-by-Step Batch Inference Setup

1. Estimate typical RPS (requests per second) — this determines batch size.
2. Choose batch_size so that latency does not exceed 50 ms for 95% of requests.
3. Set a batch accumulation timeout (usually 5-15 ms).
4. Use asynchronous queues (asyncio.Queue) to collect requests.
5. Profile with cProfile or py-spy.

Model Registry and Versioning

Model registry with MLflow allows automatic promotion of models to Production based on accuracy and Sharpe ratio thresholds.

import mlflow from mlflow.tracking import MlflowClient class ModelRegistry: def __init__(self, tracking_uri): mlflow.set_tracking_uri(tracking_uri) self.client = MlflowClient() def load_production_model(self, model_name): model_version = self.client.get_latest_versions( model_name, stages=['Production'] )[0] model = mlflow.sklearn.load_model( f"models:/{model_name}/{model_version.version}" ) return model, model_version def promote_to_production(self, model_name, version, metrics): if metrics['test_accuracy'] > 0.54 and metrics['sharpe'] > 1.2: self.client.transition_model_version_stage( model_name, version, 'Production' ) return True return False 

Why Prediction Quality Monitoring Matters?

Real-time monitoring catches degradation before losses occur. Metrics are collected in Prometheus, visualized in Grafana. When directional accuracy drops below 50%, an automatic rollback is triggered.

How Automatic Rollback Works

We configured the pipeline so that when accuracy decreases or latency increases above thresholds, the system rolls back to the previous Production model version. This takes less than 10 seconds. All metrics are logged in MLflow, enabling quick analysis of degradation causes.

Implementation Stages

Total timeline: 4 to 8 weeks. Cost is calculated individually. The average project pays for itself in 4-6 months through GPU-hour cost savings and improved trading accuracy.

What's Included in the Work

• Audit of current ML infrastructure
• Architecture design for real-time serving
• Development of feature pipeline and inference server
• Integration with MLflow and A/B testing setup
• Quality monitoring and alerting (Prometheus + Grafana)
• Documentation and team training

Order a turnkey system development — get a consultation on architecture and latency estimation within a day. We guarantee SLA for latency and accuracy. Contact us for an audit of your current ML infrastructure. GPU-hour savings through batching reach up to 30%. We specialize in crypto trading ML systems, providing real-time ML predictions with strict SLA latency guarantees.

MLflow documentation FastAPI