Мы сталкивались с ситуацией: обученная модель показывает 70% accuracy на исторических данных, но в продакшене предсказания приходят с задержкой в несколько секунд — стратегия теряет прибыль. Система realtime ML predictions — это не просто «запустить модель», это инфраструктура с low-latency serving, мониторингом качества и автоматическим переключением моделей. Наш опыт — 10+ лет в high-load ML и блокчейн-трейдинге, 5 внедрённых систем под ключ. Сертифицированные инженеры гарантируют P95 latency ниже 50 мс и точность предсказаний не хуже 55% directional accuracy. Мы реализовали более 5 таких систем для криптофондов и проп-трейдинговых компаний.
Чтобы добиться стабильной задержки и точности, нужно решить несколько ключевых проблем: оптимизация пайплайна признаков, выбор способа serving, batching, версионирование моделей и мониторинг в реальном времени. Разберём каждую на примере реального проекта — торговой системы на криптовалютном рынке. По данным NVIDIA, batching улучшает утилизацию GPU до 5 раз.
Как построить low-latency ML inference?
Архитектура realtime serving строится вокруг конвейера: данные → фичи → инференс → потребление. Покажем на примере торговой системы.
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 для realtime
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 с 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()
)
Почему batching в 10 раз эффективнее одиночных запросов?
При большом количестве запросов batching снижает overhead. Вместо тысячи отдельных вызовов — один батч. Так throughput растёт линейно до 10x, а на GPU — до 15x. Снижение затрат на GPU-часы достигает 50%. Batching — ключевой приём для low-latency систем: он уменьшает число вызовов модели и амортизирует фиксированные издержки. Благодаря батчингу и оптимизации пайплайна вы снижаете затраты на GPU-часы на 30-50%, а средний проект окупается за 4-6 месяцев.
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])
Пошаговая настройка batch-инференса
- Оцените типичный RPS (requests per second) — от этого зависит размер батча.
- Выберите
batch_sizeтак, чтобы latency не превышала 50 мс для 95% запросов. - Настройте таймаут накопления батча (обычно 5-15 мс).
- Используйте асинхронные очереди (asyncio.Queue) для сбора запросов.
- Профилируйте с помощью cProfile или py-spy.
Model registry и версионирование
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
Model registry с MLflow позволяет автоматически продвигать модели в Production по порогам точности и Sharpe ratio.
Почему важен мониторинг качества предсказаний?
Реалтайм-мониторинг позволяет поймать деградацию до потерь. Метрики собираются в Prometheus, визуализируются в Grafana. При падении directional accuracy ниже 50% — автоматический rollback.
| Метрика | Описание | Порог срабатывания |
|---|---|---|
| directional_accuracy | Доля совпадения направления | <0.55 |
| high_confidence_accuracy | Точность при confidence >0.7 | <0.65 |
| P95 latency | Задержка инференса | >50 ms |
| P99 latency | Максимальная задержка | >100 ms |
Как работает автоматический rollback?
Мы настроили пайплайн так: при снижении accuracy или росте latency выше порога система откатывает модель до предыдущей Production-версии. Это занимает менее 10 секунд. Все метрики логируются в MLflow, что позволяет быстро анализировать причину деградации.
Этапы внедрения
| Этап | Длительность | Результат |
|---|---|---|
| Аналитика и замер текущих latency | 1 неделя | baseline метрик, узкие места |
| Проектирование и прототип | 2 недели | архитектура, выбор технологий |
| Реализация core-компонентов | 3-4 недели | feature pipeline, inference server |
| Интеграция и нагрузочное тестирование | 1 неделя | подтверждение SLA по latency |
| Запуск и мониторинг | 1 неделя | product-система с алертингом |
Общий срок — от 4 до 8 недель. Стоимость рассчитывается индивидуально. Средний проект окупается за 4-6 месяцев за счёт снижения затрат на GPU-часы и повышения точности торговли.
Что входит в работу
- Аудит текущей ML-инфраструктуры
- Проектирование архитектуры realtime serving
- Разработка feature pipeline и inference сервера
- Интеграция с MLflow и настройка A/B тестирования
- Мониторинг качества и алертинг (Prometheus + Grafana)
- Документация и обучение команды
Закажите разработку системы под ключ — получите консультацию по архитектуре и оценку latency в течение дня. Мы гарантируем SLA по latency и accuracy. Свяжитесь с нами для аудита вашей текущей ML-инфраструктуры. Экономия на GPU-часах за счёт batching достигает 30%.