┌─────────────────────────────────────────────────────────────────┐
│                       Data Sources                              │
│  ┌──────┐  ┌──────┐  ┌──────┐  ┌──────┐  ┌──────┐            │
│  │Price │  │Fund. │  │Alt.  │  │Order │  │Macro │            │
│  │Data  │  │Data  │  │Data  │  │Book  │  │Data  │            │
│  └──┬───┘  └──┬───┘  └──┬───┘  └──┬───┘  └──┬───┘            │
│     │         │         │         │         │                   │
│     └─────────┴─────────┴─────────┴─────────┘                   │
│                          │                                      │
│                    Feature Store                                │
│                          │                                      │
│              ┌───────────┴───────────┐                         │
│              │   Model Training       │                         │
│              │   (Offline/Air-gapped) │                         │
│              └───────────┬───────────┘                         │
│                          │                                      │
│              ┌───────────┴───────────┐                         │
│              │   Model Registry      │                         │
│              │   (Versioned Models)  │                         │
│              └───────────┬───────────┘                         │
│                          │                                      │
│              ┌───────────┴───────────┐                         │
│              │   Model Serving       │                         │
│              │   (REST API / Batch)  │                         │
│              └───────────┬───────────┘                         │
│                          │                                      │
│              ┌───────────┴───────────┐                         │
│              │   Monitoring &        │                         │
│              │   Alerting            │                         │
│              └───────────────────────┘                         │
└─────────────────────────────────────────────────────────────────┘

import torch
import joblib
import pickle
import json
from datetime import datetime

# Save model state dict (recommended)
torch.save(model.state_dict(), 'model_weights.pth')

# Save entire model (includes architecture)
torch.save(model, 'model_full.pth')

# Save with metadata
checkpoint = {
    'model_state_dict': model.state_dict(),
    'optimizer_state_dict': optimizer.state_dict(),
    'epoch': epoch,
    'val_loss': val_loss,
    'model_architecture': str(model),
    'timestamp': datetime.now().isoformat()
}
torch.save(checkpoint, 'checkpoint.pth')

# Load model
model = FinancialNN(input_dim, hidden_dims, output_dim)
model.load_state_dict(torch.load('model_weights.pth'))
model.eval()  # Important: set to eval mode for inference

# Trace with example input
example_input = torch.randn(1, 60, 50)  # (batch, seq_len, features)
traced_model = torch.jit.trace(model, example_input)
traced_model.save('model_traced.pt')

# Script (for models with control flow)
scripted_model = torch.jit.script(model)
scripted_model.save('model_scripted.pt')

# Load and run
loaded_model = torch.jit.load('model_traced.pt')
prediction = loaded_model(example_input)

import onnx
import onnxruntime

# Export to ONNX
dummy_input = torch.randn(1, 60, 50)
torch.onnx.export(
    model,
    dummy_input,
    "model.onnx",
    export_params=True,
    opset_version=11,
    do_constant_folding=True,
    input_names=['input'],
    output_names=['output'],
    dynamic_axes={
        'input': {0: 'batch_size', 1: 'seq_length'},
        'output': {0: 'batch_size'}
    }
)

# Load and run with ONNX Runtime
ort_session = onnxruntime.InferenceSession("model.onnx")
inputs = {ort_session.get_inputs()[0].name: dummy_input.numpy()}
outputs = ort_session.run(None, inputs)
prediction = outputs[0]

import joblib

# Save scikit-learn model
joblib.dump(model, 'model.joblib')

# Save with metadata
model_info = {
    'model': model,
    'features': feature_names,
    'scaler': scaler,
    'timestamp': datetime.now().isoformat(),
    'version': '1.0.0'
}
joblib.dump(model_info, 'model_with_metadata.joblib')

# Load
loaded = joblib.load('model.joblib')

class ModelRegistry:
    def __init__(self, base_path='./model_registry'):
        self.base_path = base_path
        os.makedirs(base_path, exist_ok=True)

    def register_model(self, model, metadata):
        """
        Register a new model version.
        """
        version = metadata.get('version', datetime.now().strftime('%Y%m%d_%H%M%S'))
        model_path = f"{self.base_path}/{version}"

        os.makedirs(model_path, exist_ok=True)

        # Save model
        if isinstance(model, torch.nn.Module):
            torch.save(model.state_dict(), f"{model_path}/model.pth")
        else:
            joblib.dump(model, f"{model_path}/model.joblib")

        # Save metadata
        with open(f"{model_path}/metadata.json", 'w') as f:
            json.dump(metadata, f, indent=2)

        # Update latest symlink
        latest_path = f"{self.base_path}/latest"
        if os.path.exists(latest_path) or os.path.islink(latest_path):
            os.unlink(latest_path)
        os.symlink(version, latest_path)

        print(f"Model registered: {version}")
        return version

    def load_model(self, version='latest'):
        """
        Load a specific model version.
        """
        model_path = f"{self.base_path}/{version}"
        if not os.path.exists(model_path):
            raise ValueError(f"Model version {version} not found")

        with open(f"{model_path}/metadata.json", 'r') as f:
            metadata = json.load(f)

        # Load model
        if os.path.exists(f"{model_path}/model.pth"):
            # PyTorch model (need to instantiate first)
            # Assume architecture is stored in metadata
            model = FinancialNN(metadata['input_dim'], metadata['hidden_dims'], metadata['output_dim'])
            model.load_state_dict(torch.load(f"{model_path}/model.pth"))
        else:
            model = joblib.load(f"{model_path}/model.joblib")

        return model, metadata

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import numpy as np
import torch
from typing import List, Optional
import uvicorn

app = FastAPI(title="Financial AI Trading API", version="1.0.0")

# Define request/response models
class PredictionRequest(BaseModel):
    features: List[List[float]]  # List of sequences or feature vectors
    seq_length: Optional[int] = 60
    asset_ids: Optional[List[str]] = None

class PredictionResponse(BaseModel):
    predictions: List[float]
    confidence_scores: Optional[List[float]] = None
    timestamp: str

# Global model instance
model = None
device = None

@app.on_event("startup")
async def load_model():
    """
    Load model on startup.
    """
    global model, device
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

    # Load from registry
    registry = ModelRegistry()
    model, metadata = registry.load_model('latest')
    model.to(device)
    model.eval()
    print(f"Model loaded: {metadata}")

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    """
    Make predictions from input features.
    """
    try:
        # Convert to numpy array
        features = np.array(request.features, dtype=np.float32)

        # Check dimensions
        if len(features.shape) == 2:
            # Single sequence: (seq_len, features)
            features = features.reshape(1, features.shape[0], features.shape[1])
        elif len(features.shape) == 3:
            # Batch of sequences: (batch, seq_len, features)
            pass
        else:
            raise HTTPException(status_code=400, detail="Invalid feature dimensions")

        # Convert to tensor
        input_tensor = torch.from_numpy(features).to(device)

        # Run inference
        with torch.no_grad():
            predictions = model(input_tensor)
            predictions = predictions.cpu().numpy().flatten().tolist()

        return PredictionResponse(
            predictions=predictions,
            confidence_scores=None,
            timestamp=datetime.now().isoformat()
        )

    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", "model_loaded": model is not None}

@app.get("/metrics")
async def get_metrics():
    """
    Get model performance metrics.
    """
    # In production, fetch from monitoring system
    return {
        "latency_p50": "15ms",
        "latency_p95": "45ms",
        "requests_per_second": 120,
        "error_rate": "0.01%"
    }

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

# Dockerfile
FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  model-api:
    build: .
    ports:
      - "8000:8000"
    volumes:
      - ./model_registry:/app/model_registry
    environment:
      - CUDA_VISIBLE_DEVICES=0
    restart: unless-stopped

  redis:
    image: redis:alpine
    ports:
      - "6379:6379"

  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin

import requests
import numpy as np

class FinancialAPIClient:
    def __init__(self, base_url="http://localhost:8000"):
        self.base_url = base_url

    def predict(self, features):
        """
        Send prediction request to API.
        """
        response = requests.post(
            f"{self.base_url}/predict",
            json={"features": features.tolist()}
        )

        if response.status_code == 200:
            return response.json()['predictions']
        else:
            raise Exception(f"API Error: {response.text}")

    def health_check(self):
        response = requests.get(f"{self.base_url}/health")
        return response.json()

class BatchInferencePipeline:
    def __init__(self, model_path, device='cpu'):
        self.device = torch.device(device)
        self.model = self.load_model(model_path)

    def load_model(self, model_path):
        model = torch.jit.load(model_path)
        model.to(self.device)
        model.eval()
        return model

    def process_batch(self, data_loader, batch_size=256):
        """
        Run inference on a batch of data.
        """
        predictions = []

        with torch.no_grad():
            for batch in data_loader:
                batch = batch.to(self.device)
                pred = self.model(batch)
                predictions.append(pred.cpu().numpy())

        return np.concatenate(predictions, axis=0)

    def run_scheduled_pipeline(self, data_source, output_path, schedule='daily'):
        """
        Run the inference pipeline on a schedule.
        """
        import schedule
        import time

        def job():
            print(f"Running pipeline at {datetime.now()}")
            data = data_source.fetch()
            predictions = self.process_batch(data)
            self.save_results(predictions, output_path)

        if schedule == 'daily':
            schedule.every().day.at("17:00").do(job)  # After market close
        elif schedule == 'hourly':
            schedule.every().hour.do(job)

        while True:
            schedule.run_pending()
            time.sleep(60)

def compute_psi(train_dist, prod_dist, buckets=10):
    """
    Population Stability Index.
    PSI = Σ (prod_pct - train_pct) * ln(prod_pct / train_pct)
    """
    # Create bins for both distributions
    combined = np.concatenate([train_dist, prod_dist])
    bins = np.percentile(combined, np.linspace(0, 100, buckets+1))

    train_hist, _ = np.histogram(train_dist, bins=bins)
    prod_hist, _ = np.histogram(prod_dist, bins=bins)

    train_pct = train_hist / len(train_dist) + 1e-8
    prod_pct = prod_hist / len(prod_dist) + 1e-8

    psi = np.sum((prod_pct - train_pct) * np.log(prod_pct / train_pct))
    return psi

# Interpretation
# PSI < 0.1: No significant shift
# 0.1 ≤ PSI < 0.2: Moderate shift (investigate)
# PSI ≥ 0.2: Significant shift (retrain required)

from scipy.stats import ks_2samp

def detect_feature_drift(train_feature, prod_feature, threshold=0.05):
    """
    Detect drift using Kolmogorov-Smirnov test.
    """
    stat, p_value = ks_2samp(train_feature, prod_feature)

    drift_detected = p_value < threshold
    return {
        'statistic': stat,
        'p_value': p_value,
        'drift_detected': drift_detected
    }

class ConceptDriftDetector:
    def __init__(self, window_size=100, threshold=0.1):
        self.window_size = window_size
        self.threshold = threshold
        self.predictions = []
        self.actuals = []
        self.metrics = []

    def update(self, y_pred, y_true):
        """
        Update with new prediction and actual value.
        """
        self.predictions.append(y_pred)
        self.actuals.append(y_true)

        # Keep only recent window
        if len(self.predictions) > self.window_size:
            self.predictions = self.predictions[-self.window_size:]
            self.actuals = self.actuals[-self.window_size:]

        # Compute current accuracy
        accuracy = np.mean(np.array(self.predictions) == np.array(self.actuals))
        self.metrics.append(accuracy)

        # Check for drift
        if len(self.metrics) > self.window_size:
            recent_accuracy = np.mean(self.metrics[-10:])
            historical_accuracy = np.mean(self.metrics[:-10])

            drift_score = (historical_accuracy - recent_accuracy) / historical_accuracy
            return drift_score > self.threshold

        return False

from prometheus_client import Counter, Histogram, Gauge, start_http_server

# Define metrics
prediction_counter = Counter('predictions_total', 'Total predictions made')
error_counter = Counter('prediction_errors_total', 'Total prediction errors')
latency_histogram = Histogram('prediction_latency_seconds', 'Prediction latency')
drift_gauge = Gauge('feature_drift', 'Feature drift score')

# Example usage
@app.post("/predict")
async def predict(request: PredictionRequest):
    start_time = time.time()
    try:
        # ... prediction logic ...
        prediction_counter.inc()
        latency_histogram.observe(time.time() - start_time)
        return PredictionResponse(...)
    except Exception as e:
        error_counter.inc()
        raise

# Start Prometheus metrics server
start_http_server(8001)  # Metrics on port 8001

class RetrainingScheduler:
    def __init__(self, model_registry, data_source):
        self.registry = model_registry
        self.data_source = data_source
        self.last_retrain = datetime.now()

    def should_retrain(self):
        """
        Check if retraining is needed based on multiple conditions.
        """
        conditions = [
            self.time_based_check(),     # Every 30 days
            self.performance_check(),    # Accuracy drop > 10%
            self.drift_check(),          # PSI > 0.2
            self.volume_check()          # New data > threshold
        ]
        return any(conditions)

    def time_based_check(self):
        days_since_retrain = (datetime.now() - self.last_retrain).days
        return days_since_retrain >= 30

    def performance_check(self):
        # Compare current model performance with historical
        pass

    def drift_check(self):
        # Check PSI for all features
        pass

    def volume_check(self):
        # Check if new data volume exceeds threshold
        new_data_count = self.data_source.get_new_count(since=self.last_retrain)
        return new_data_count > 10000

class RetrainingPipeline:
    def __init__(self, config):
        self.config = config
        self.registry = ModelRegistry()
        self.trainer = ModelTrainer()

    def run_retraining(self):
        """
        Execute the full retraining pipeline.
        """
        print("Starting retraining pipeline...")

        # 1. Fetch new data
        print("Fetching data...")
        data = self.data_source.fetch_all()

        # 2. Feature engineering
        print("Engineering features...")
        features = self.feature_engineer(data)

        # 3. Train new model
        print("Training model...")
        new_model = self.trainer.train(features)

        # 4. Validate new model
        print("Validating model...")
        validation_results = self.trainer.validate(new_model)

        # 5. Compare with current best
        current_model, _ = self.registry.load_model('latest')
        current_performance = self.trainer.evaluate(current_model)

        # 6. Register if better
        if validation_results['sharpe'] > current_performance['sharpe']:
            print("New model outperforms. Registering...")
            metadata = {
                'version': datetime.now().strftime('%Y%m%d_%H%M%S'),
                'sharpe': validation_results['sharpe'],
                'features_used': list(features.columns),
                'data_range': (data.index.min(), data.index.max())
            }
            self.registry.register_model(new_model, metadata)
            self.last_retrain = datetime.now()
        else:
            print("New model does not outperform. Skipping registration.")

        return validation_results

# .github/workflows/retrain.yml
name: Retrain Model

on:
  schedule:
    - cron: '0 17 * * *'  # Daily at 5 PM
  workflow_dispatch:  # Manual trigger

jobs:
  retrain:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Run retraining
        run: python scripts/retrain.py
      - name: Deploy if improved
        run: python scripts/deploy.py

class ModelValidation:
    def __init__(self, model, training_data, validation_data):
        self.model = model
        self.train_data = training_data
        self.val_data = validation_data

    def validate(self):
        """
        Perform comprehensive model validation.
        """
        results = {
            'statistical_validity': self.test_statistical_validity(),
            'conceptual_soundness': self.test_conceptual_soundness(),
            'out_of_sample_performance': self.test_oot_performance(),
            'stability': self.test_stability(),
            'sensitivity': self.test_sensitivity(),
            'explainability': self.test_explainability()
        }
        return results

    def test_statistical_validity(self):
        """
        Test statistical assumptions.
        """
        # Normality of residuals
        residuals = self.train_data['y'] - self.model.predict(self.train_data)
        _, p_value = stats.normaltest(residuals)
        return {'normality_p_value': p_value}

    def test_conceptual_soundness(self):
        """
        Test if the model makes conceptual sense.
        """
        # Feature importance should align with financial theory
        # e.g., volatility should have positive coefficient for option pricing
        pass

    def test_oot_performance(self):
        """
        Out-of-time and out-of-sample performance.
        """
        predictions = self.model.predict(self.val_data)
        sharpe = compute_sharpe_ratio(predictions)
        return {'oot_sharpe': sharpe}

    def test_stability(self):
        """
        Test model stability across different time periods.
        """
        # Use Chow test for structural breaks
        pass

    def test_sensitivity(self):
        """
        Sensitivity analysis.
        """
        # Perturb inputs and check output changes
        pass

    def test_explainability(self):
        """
        Ensure model is explainable.
        """
        # SHAP or LIME analysis
        pass

import shap

def explain_model_shap(model, X_background, X_explain):
    """
    Use SHAP to explain model predictions.
    """
    # Create explainer
    explainer = shap.KernelExplainer(model.predict, X_background)

    # Compute SHAP values
    shap_values = explainer.shap_values(X_explain)

    # Summary plot
    shap.summary_plot(shap_values, X_explain, feature_names=feature_names)

    # Force plot for individual prediction
    shap.force_plot(explainer.expected_value, shap_values[0], X_explain.iloc[0])

    return shap_values

import lime
import lime.lime_tabular

def explain_model_lime(model, X_train, X_explain):
    """
    Use LIME to explain model predictions.
    """
    explainer = lime.lime_tabular.LimeTabularExplainer(
        X_train.values,
        feature_names=feature_names,
        class_names=['Down', 'Up'],
        mode='classification'
    )

    explanation = explainer.explain_instance(
        X_explain.iloc[0].values,
        model.predict_proba,
        num_features=10
    )

    explanation.show_in_notebook()
    return explanation