Низьколатентний 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