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
FastAPI Serving Code
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
- Estimate typical RPS (requests per second) — this determines batch size.
- Choose
batch_sizeso that latency does not exceed 50 ms for 95% of requests. - Set a batch accumulation timeout (usually 5-15 ms).
- Use asynchronous queues (asyncio.Queue) to collect requests.
- 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.
| Metric | Description | Alert Threshold |
|---|---|---|
| directional_accuracy | Fraction of correct direction | <0.55 |
| high_confidence_accuracy | Accuracy when confidence >0.7 | <0.65 |
| P95 latency | Inference latency | >50 ms |
| P99 latency | Maximum latency | >100 ms |
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
| Stage | Duration | Result |
|---|---|---|
| Analytics and current latency measurement | 1 week | baseline metrics, bottlenecks |
| Design and prototype | 2 weeks | architecture, technology selection |
| Core component implementation | 3-4 weeks | feature pipeline, inference server |
| Integration and load testing | 1 week | SLA latency confirmation |
| Launch and monitoring | 1 week | production system with alerting |
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.