Time Series Deployment Guide - Generative Time Series

1. Deployment Architecture Overview

Deploying generative time series models in production requires careful consideration of scalability, reliability, and monitoring to ensure optimal performance and availability.

                Key Deployment Considerations:
                Scalability: Handle varying loads and concurrent requests
Reliability: Ensure high availability and fault tolerance
Performance: Optimize inference speed and resource usage
Monitoring: Track model performance and system health
Security: Protect models and data in production

            

2. Model Optimization

Model Compression

# Model Quantization
import torch.quantization as quantization

def optimize_model_for_deployment(model, calibration_data):
    """Optimize model for production deployment"""
    
    # 1. Quantization
    model.eval()
    model.qconfig = quantization.get_default_qconfig('fbgemm')
    model_prepared = quantization.prepare(model)
    
    # Calibrate with sample data
    with torch.no_grad():
        for batch in calibration_data:
            _ = model_prepared(batch)
    
    quantized_model = quantization.convert(model_prepared)
    
    # 2. Pruning
    import torch.nn.utils.prune as prune
    
    # Prune 20% of connections
    for module in quantized_model.modules():
        if isinstance(module, torch.nn.Linear):
            prune.l1_unstructured(module, name='weight', amount=0.2)
            prune.remove(module, 'weight')
    
    return quantized_model
            

TensorRT Optimization

NVIDIA TensorRT Integration:

import torch_tensorrt

def optimize_with_tensorrt(model, input_shape, precision='fp16'):
    """Optimize model with TensorRT for NVIDIA GPUs"""
    
    # Compile model for TensorRT
    trt_model = torch_tensorrt.compile(
        model,
        inputs=[torch.randn(input_shape)],
        enabled_precisions={torch.float, torch.half},
        workspace_size=1 << 30,  # 1GB workspace
        min_block_size=7,
        truncate_long_and_double=True
    )
    
    return trt_model

# Usage example
optimized_model = optimize_with_tensorrt(
    model=timegpt_model,
    input_shape=(1, 512, 1),
    precision='fp16'
)
                

3. Containerization

Docker Configuration

# Dockerfile for Time Series Model Deployment
FROM nvidia/cuda:11.8-devel-ubuntu20.04

# Install Python and dependencies
RUN apt-get update && apt-get install -y \
    python3.8 python3-pip \
    && rm -rf /var/lib/apt/lists/*

# Set working directory
WORKDIR /app

# Copy requirements and install dependencies
COPY requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt

# Copy model and application code
COPY models/ ./models/
COPY app.py .

# Expose port
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# Run application
CMD ["python3", "app.py"]
            

Docker Compose

Multi-Service Deployment:

# docker-compose.yml
version: '3.8'

services:
  timeseries-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - MODEL_PATH=/app/models/timegpt_best.pth
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
      - postgres
    deploy:
      replicas: 3
      resources:
        limits:
          memory: 4G
          cpus: '2.0'
        reservations:
          memory: 2G
          cpus: '1.0'

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  postgres:
    image: postgres:15
    environment:
      POSTGRES_DB: timeseries
      POSTGRES_USER: admin
      POSTGRES_PASSWORD: password
    volumes:
      - postgres_data:/var/lib/postgresql/data

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - timeseries-api

volumes:
  redis_data:
  postgres_data:
                

4. API Design

FastAPI Implementation

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
import asyncio
import redis
import json
from typing import List, Optional

app = FastAPI(title="TimeGPT API", version="1.0.0")

# Redis for caching
redis_client = redis.Redis(host='redis', port=6379, decode_responses=True)

class ForecastRequest(BaseModel):
    data: List[float]
    horizon: int = 24
    include_uncertainty: bool = False
    confidence_level: float = 0.95

class ForecastResponse(BaseModel):
    forecast: List[float]
    uncertainty: Optional[List[float]] = None
    confidence_level: float
    request_id: str

@app.post("/forecast", response_model=ForecastResponse)
async def forecast(request: ForecastRequest, background_tasks: BackgroundTasks):
    """Generate time series forecast"""
    
    # Check cache first
    cache_key = f"forecast:{hash(tuple(request.data))}:{request.horizon}"
    cached_result = redis_client.get(cache_key)
    
    if cached_result:
        result = json.loads(cached_result)
        result['from_cache'] = True
        return ForecastResponse(**result)
    
    try:
        # Generate forecast
        forecast_result = await generate_forecast(request)
        
        # Cache result
        cache_data = {
            'forecast': forecast_result['forecast'],
            'uncertainty': forecast_result.get('uncertainty'),
            'confidence_level': forecast_result['confidence_level'],
            'request_id': forecast_result['request_id']
        }
        redis_client.setex(cache_key, 3600, json.dumps(cache_data))  # Cache for 1 hour
        
        # Log request
        background_tasks.add_task(log_request, request, forecast_result)
        
        return ForecastResponse(**forecast_result)
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {
        "status": "healthy",
        "timestamp": datetime.utcnow().isoformat(),
        "version": "1.0.0"
    }
            

Rate Limiting

Request Rate Limiting:

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/forecast")
@limiter.limit("100/minute")
async def forecast(request: Request, forecast_request: ForecastRequest):
    """Rate-limited forecast endpoint"""
    # Implementation here
    pass
                

5. Monitoring and Observability

Metrics Collection

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# Metrics
REQUEST_COUNT = Counter('requests_total', 'Total requests', ['method', 'endpoint'])
REQUEST_DURATION = Histogram('request_duration_seconds', 'Request duration')
ACTIVE_CONNECTIONS = Gauge('active_connections', 'Active connections')
MODEL_ACCURACY = Gauge('model_accuracy', 'Model prediction accuracy')

class MetricsMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope['type'] == 'http':
            start_time = time.time()
            
            # Track request
            REQUEST_COUNT.labels(
                method=scope['method'],
                endpoint=scope['path']
            ).inc()
            
            # Track active connections
            ACTIVE_CONNECTIONS.inc()
            
            try:
                await self.app(scope, receive, send)
            finally:
                # Record duration
                REQUEST_DURATION.observe(time.time() - start_time)
                ACTIVE_CONNECTIONS.dec()
        else:
            await self.app(scope, receive, send)
            

Logging Configuration

Structured Logging:

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'level': record.levelname,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno
        }
        
        if hasattr(record, 'request_id'):
            log_entry['request_id'] = record.request_id
        
        if hasattr(record, 'user_id'):
            log_entry['user_id'] = record.user_id
        
        return json.dumps(log_entry)

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)
logger.addHandler(logging.StreamHandler())
            

6. Scalability Strategies

Horizontal Scaling

Scaling Considerations:

Load Balancing: Distribute requests across multiple instances
Stateless Design: Ensure services can be scaled independently
Database Scaling: Use read replicas and connection pooling
Caching Strategy: Implement distributed caching

Kubernetes Deployment

# kubernetes-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: timeseries-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: timeseries-api
  template:
    metadata:
      labels:
        app: timeseries-api
    spec:
      containers:
      - name: timeseries-api
        image: timeseries-api:latest
        ports:
        - containerPort: 8000
        env:
        - name: MODEL_PATH
          value: "/app/models/timegpt_best.pth"
        resources:
          requests:
            memory: "2Gi"
            cpu: "1000m"
            nvidia.com/gpu: 1
          limits:
            memory: "4Gi"
            cpu: "2000m"
            nvidia.com/gpu: 1
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5

---
apiVersion: v1
kind: Service
metadata:
  name: timeseries-service
spec:
  selector:
    app: timeseries-api
  ports:
  - port: 80
    targetPort: 8000
  type: LoadBalancer
            

7. Security Considerations

Authentication and Authorization

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
import os

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """Verify JWT token"""
    try:
        payload = jwt.decode(
            credentials.credentials,
            os.getenv('JWT_SECRET'),
            algorithms=['HS256']
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token expired"
        )
    except jwt.JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )

@app.post("/forecast")
async def forecast(
    request: ForecastRequest,
    user: dict = Depends(verify_token)
):
    """Authenticated forecast endpoint"""
    # Check user permissions
    if not user.get('can_forecast'):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Insufficient permissions"
        )
    
    # Generate forecast
    return await generate_forecast(request)
            

Input Validation

Data Validation:

from pydantic import validator
import numpy as np

class ForecastRequest(BaseModel):
    data: List[float]
    horizon: int = 24
    include_uncertainty: bool = False
    confidence_level: float = 0.95
    
    @validator('data')
    def validate_data(cls, v):
        if len(v) < 10:
            raise ValueError('Input data must have at least 10 points')
        if len(v) > 10000:
            raise ValueError('Input data too large (max 10000 points)')
        
        # Check for NaN or infinite values
        if any(not np.isfinite(x) for x in v):
            raise ValueError('Input data contains invalid values')
        
        return v
    
    @validator('horizon')
    def validate_horizon(cls, v):
        if v < 1 or v > 365:
            raise ValueError('Horizon must be between 1 and 365')
        return v
    
    @validator('confidence_level')
    def validate_confidence(cls, v):
        if v < 0.5 or v > 0.99:
            raise ValueError('Confidence level must be between 0.5 and 0.99')
        return v
                

8. Testing and Quality Assurance

Load Testing

# locustfile.py - Load testing with Locust
from locust import HttpUser, task, between
import random

class TimeSeriesUser(HttpUser):
    wait_time = between(1, 3)
    
    @task(3)
    def forecast_request(self):
        """Generate forecast requests"""
        data = [random.uniform(50, 150) for _ in range(100)]
        
        payload = {
            "data": data,
            "horizon": random.randint(1, 48),
            "include_uncertainty": random.choice([True, False]),
            "confidence_level": 0.95
        }
        
        with self.client.post("/forecast", json=payload) as response:
            if response.status_code != 200:
                print(f"Request failed: {response.status_code}")
    
    @task(1)
    def health_check(self):
        """Health check requests"""
        self.client.get("/health")
            

Integration Testing

Test Suite:

import pytest
import asyncio
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_forecast_endpoint():
    """Test forecast endpoint functionality"""
    async with AsyncClient(app=app, base_url="http://test") as ac:
        # Test valid request
        response = await ac.post("/forecast", json={
            "data": [100, 105, 110, 108, 112, 115, 118, 120],
            "horizon": 5,
            "include_uncertainty": True
        })
        
        assert response.status_code == 200
        data = response.json()
        assert "forecast" in data
        assert len(data["forecast"]) == 5
        assert "uncertainty" in data

@pytest.mark.asyncio
async def test_rate_limiting():
    """Test rate limiting functionality"""
    async with AsyncClient(app=app, base_url="http://test") as ac:
        # Make multiple requests quickly
        tasks = []
        for _ in range(150):  # Exceed rate limit
            task = ac.post("/forecast", json={
                "data": [100, 105, 110],
                "horizon": 3
            })
            tasks.append(task)
        
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Some requests should be rate limited
        rate_limited = sum(1 for r in responses if isinstance(r, Exception) or r.status_code == 429)
        assert rate_limited > 0
                