Создание низколатентного ML-сервиса с мониторингом и автоматизацией

Мы сталкивались с ситуацией: обученная модель показывает 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-инференса

1. Оцените типичный RPS (requests per second) — от этого зависит размер батча.
2. Выберите batch_size так, чтобы latency не превышала 50 мс для 95% запросов.
3. Настройте таймаут накопления батча (обычно 5-15 мс).
4. Используйте асинхронные очереди (asyncio.Queue) для сбора запросов.
5. Профилируйте с помощью 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.

Как работает автоматический rollback?

Мы настроили пайплайн так: при снижении accuracy или росте latency выше порога система откатывает модель до предыдущей Production-версии. Это занимает менее 10 секунд. Все метрики логируются в MLflow, что позволяет быстро анализировать причину деградации.

Этапы внедрения

Общий срок — от 4 до 8 недель. Стоимость рассчитывается индивидуально. Средний проект окупается за 4-6 месяцев за счёт снижения затрат на GPU-часы и повышения точности торговли.

Что входит в работу

• Аудит текущей ML-инфраструктуры
• Проектирование архитектуры realtime serving
• Разработка feature pipeline и inference сервера
• Интеграция с MLflow и настройка A/B тестирования
• Мониторинг качества и алертинг (Prometheus + Grafana)
• Документация и обучение команды

Закажите разработку системы под ключ — получите консультацию по архитектуре и оценку latency в течение дня. Мы гарантируем SLA по latency и accuracy. Свяжитесь с нами для аудита вашей текущей ML-инфраструктуры. Экономия на GPU-часах за счёт batching достигает 30%.

MLflow documentation FastAPI