┌─────────────────────────────────────────────────────────────────────────┐
│                        User Interface / API Gateway                    │
│                            (Nginx / Kong)                             │
└───────────────────────────────┬─────────────────────────────────────────┘
                                │
┌───────────────────────────────▼─────────────────────────────────────────┐
│                        Load Balancer                                   │
│                        (HAProxy / AWS ELB)                            │
└───────────────────────────────┬─────────────────────────────────────────┘
                                │
┌───────────────────────────────▼─────────────────────────────────────────┐
│                    Application Layer (Kubernetes)                      │
│  ┌───────────────┐  ┌───────────────┐  ┌───────────────┐              │
│  │   API Server  │  │  Model Server │  │  Batch Jobs   │              │
│  │   (FastAPI)   │  │  (TorchServe) │  │  (Spark/K8s)  │              │
│  └───────────────┘  └───────────────┘  └───────────────┘              │
└───────────────────────────────┬─────────────────────────────────────────┘
                                │
┌───────────────────────────────▼─────────────────────────────────────────┐
│                        Data Layer                                      │
│  ┌───────────────┐  ┌───────────────┐  ┌───────────────┐              │
│  │  Feature Store│  │   Database    │  │   Object Store│              │
│  │  (Feast/Redis)│  │  (PostgreSQL) │  │   (S3/GCS)    │              │
│  └───────────────┘  └───────────────┘  └───────────────┘              │
└─────────────────────────────────────────────────────────────────────────┘
                                │
┌───────────────────────────────▼─────────────────────────────────────────┐
│                    Monitoring & Logging                                │
│  ┌───────────────┐  ┌───────────────┐  ┌───────────────┐              │
│  │  Prometheus   │  │  Grafana      │  │  ELK Stack    │              │
│  │  (Metrics)    │  │  (Dashboard)  │  │  (Logs)       │              │
│  └───────────────┘  └───────────────┘  └───────────────┘              │
└─────────────────────────────────────────────────────────────────────────┘

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: financial-ai-model
  labels:
    app: financial-ai
spec:
  replicas: 3
  selector:
    matchLabels:
      app: financial-ai
  template:
    metadata:
      labels:
        app: financial-ai
    spec:
      containers:
      - name: model-server
        image: financial-ai:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "2Gi"
            cpu: "1"
          limits:
            memory: "4Gi"
            cpu: "2"
        env:
        - name: CUDA_VISIBLE_DEVICES
          value: "0"
        - name: MODEL_VERSION
          value: "1.2.3"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
        volumeMounts:
        - name: model-storage
          mountPath: /app/models
      volumes:
      - name: model-storage
        persistentVolumeClaim:
          claimName: model-pvc
---
apiVersion: v1
kind: Service
metadata:
  name: model-service
spec:
  selector:
    app: financial-ai
  ports:
  - port: 80
    targetPort: 8000
  type: LoadBalancer

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: model-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: financial-ai-model
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  - type: Pods
    pods:
      metric:
        name: requests_per_second
      target:
        type: AverageValue
        averageValue: "100"

# .github/workflows/deploy.yml
name: Deploy Financial AI Model

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]
  workflow_dispatch:

jobs:
  test:
    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
        pip install -r requirements-dev.txt

    - name: Run unit tests
      run: pytest tests/unit --cov=. --cov-report=xml

    - name: Run integration tests
      run: pytest tests/integration

    - name: Upload coverage
      uses: codecov/codecov-action@v3

  validate:
    runs-on: ubuntu-latest
    needs: test
    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: Validate model performance
      run: |
        python scripts/validate_model.py --data validation_data.parquet

    - name: Check model drift
      run: |
        python scripts/check_drift.py --reference train_data.parquet --current validation_data.parquet

  build:
    runs-on: ubuntu-latest
    needs: [test, validate]
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
    steps:
    - uses: actions/checkout@v3

    - name: Set up Docker Buildx
      uses: docker/setup-buildx-action@v2

    - name: Log in to Docker Hub
      uses: docker/login-action@v2
      with:
        username: ${{ secrets.DOCKER_USERNAME }}
        password: ${{ secrets.DOCKER_TOKEN }}

    - name: Build and push Docker image
      uses: docker/build-push-action@v4
      with:
        context: .
        push: true
        tags: |
          ${{ secrets.DOCKER_USERNAME }}/financial-ai:latest
          ${{ secrets.DOCKER_USERNAME }}/financial-ai:${{ github.sha }}
        cache-from: type=gha
        cache-to: type=gha,mode=max

  deploy:
    runs-on: ubuntu-latest
    needs: build
    environment: production
    steps:
    - uses: actions/checkout@v3

    - name: Set up kubectl
      uses: azure/setup-kubectl@v3

    - name: Configure kubectl
      run: |
        echo "${{ secrets.KUBE_CONFIG }}" | base64 -d > kubeconfig
        export KUBECONFIG=kubeconfig

    - name: Deploy to Kubernetes
      run: |
        kubectl set image deployment/financial-ai-model model-server=${{ secrets.DOCKER_USERNAME }}/financial-ai:${{ github.sha }}
        kubectl rollout status deployment/financial-ai-model

    - name: Verify deployment
      run: |
        kubectl run test-pod --rm -it --restart=Never --image=curlimages/curl -- curl -f http://model-service/health

apiVersion: apps/v1
kind: Deployment
metadata:
  name: financial-ai-model-canary
  labels:
    app: financial-ai
    track: canary
spec:
  replicas: 1  # Small percentage of traffic
  selector:
    matchLabels:
      app: financial-ai
      track: canary
  template:
    metadata:
      labels:
        app: financial-ai
        track: canary
        version: canary
    spec:
      containers:
      - name: model-server
        image: financial-ai:canary

apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: model-routing
spec:
  hosts:
  - model-service
  http:
  - match:
    - headers:
        version:
          exact: canary
    route:
    - destination:
        host: model-service
        subset: canary
      weight: 100
  - route:
    - destination:
        host: model-service
        subset: stable
      weight: 95
    - destination:
        host: model-service
        subset: canary
      weight: 5  # 5% of traffic goes to canary
---
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: model-destination
spec:
  host: model-service
  subsets:
  - name: stable
    labels:
      track: stable
  - name: canary
    labels:
      track: canary

class ABTestManager:
    def __init__(self, model_a, model_b, test_name):
        self.model_a = model_a
        self.model_b = model_b
        self.test_name = test_name
        self.metrics = {'a': [], 'b': []}

    def get_model(self, user_id):
        """
        Route user to a model based on user_id hash.
        """
        if hash(user_id) % 2 == 0:
            return self.model_a, 'a'
        else:
            return self.model_b, 'b'

    def log_prediction(self, model_id, prediction, actual, metadata):
        """
        Log prediction for analysis.
        """
        self.metrics[model_id].append({
            'timestamp': datetime.now(),
            'prediction': prediction,
            'actual': actual,
            'metadata': metadata
        })

    def compute_metrics(self):
        """
        Compute performance metrics for A/B test.
        """
        results = {}
        for model_id, data in self.metrics.items():
            predictions = [d['prediction'] for d in data]
            actuals = [d['actual'] for d in data]

            results[model_id] = {
                'accuracy': np.mean(np.array(predictions) == np.array(actuals)),
                'sharpe': self.compute_sharpe(predictions, actuals),
                'sample_size': len(data)
            }

        # Statistical significance test
        from scipy.stats import ttest_ind
        p_value = ttest_ind(
            [d['prediction'] for d in self.metrics['a']],
            [d['prediction'] for d in self.metrics['b']]
        ).pvalue
        results['p_value'] = p_value

        return results

    def decide_winner(self, metric='accuracy', confidence_level=0.95):
        """
        Decide which model wins the A/B test.
        """
        results = self.compute_metrics()
        if results['p_value'] < (1 - confidence_level):
            if results['a'][metric] > results['b'][metric]:
                return 'a', results
            else:
                return 'b', results
        else:
            return None, results  # No statistically significant winner

from prometheus_client import Counter, Histogram, Gauge, Summary, Info

# Business metrics
predictions_total = Counter('predictions_total', 'Total number of predictions')
error_total = Counter('errors_total', 'Total number of errors')
prediction_latency = Histogram('prediction_latency_seconds', 'Prediction latency', buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5])

# Model performance metrics
model_accuracy = Gauge('model_accuracy', 'Current model accuracy')
model_sharpe = Gauge('model_sharpe', 'Current model Sharpe ratio')
model_latency = Gauge('model_latency_ms', 'Current model latency in ms')

# Data drift metrics
feature_drift_psi = Gauge('feature_drift_psi', 'PSI for each feature', ['feature'])
feature_drift_ks = Gauge('feature_drift_ks', 'KS statistic for each feature', ['feature'])

# Model info
model_info = Info('model_info', 'Model metadata')
model_info.info({
    'version': '1.2.3',
    'framework': 'pytorch',
    'training_date': '2024-01-15'
})

# Batch metrics
batch_processed = Counter('batch_processed_total', 'Total batch processed')
batch_duration = Histogram('batch_duration_seconds', 'Batch processing duration')

# System metrics (built-in)
from prometheus_client import Gauge
import psutil

cpu_usage = Gauge('cpu_usage_percent', 'CPU usage percentage')
memory_usage = Gauge('memory_usage_bytes', 'Memory usage in bytes')
gpu_usage = Gauge('gpu_usage_percent', 'GPU usage percentage')

from fastapi import FastAPI, Request
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response
import time

app = FastAPI()

@app.middleware("http")
async def monitor_requests(request: Request, call_next):
    """
    Middleware to collect request metrics.
    """
    start_time = time.time()

    # Process request
    response = await call_next(request)

    # Record metrics
    latency = time.time() - start_time
    prediction_latency.observe(latency)

    if response.status_code >= 400:
        error_total.inc()

    # Update model performance periodically (in background)
    # This would be done in a separate thread or scheduled job

    return response

@app.get("/metrics")
async def get_metrics():
    """
    Export Prometheus metrics.
    """
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

groups:
- name: model_alerts
  rules:
  - alert: HighErrorRate
    expr: rate(errors_total[5m]) / rate(predictions_total[5m]) > 0.01
    for: 5m
    annotations:
      summary: "High error rate detected"
      description: "Error rate is {{ $value | humanizePercentage }}"

  - alert: HighLatency
    expr: histogram_quantile(0.95, rate(prediction_latency_bucket[5m])) > 0.5
    for: 5m
    annotations:
      summary: "High prediction latency"
      description: "95th percentile latency is {{ $value }}s"

  - alert: FeatureDrift
    expr: feature_drift_psi > 0.2
    for: 1h
    annotations:
      summary: "Feature drift detected"
      description: "PSI for {{ $labels.feature }} is {{ $value }}"

  - alert: ModelAccuracyDrop
    expr: model_accuracy < 0.5
    for: 1h
    annotations:
      summary: "Model accuracy dropped below threshold"
      description: "Current accuracy is {{ $value }}"

  - alert: ModelOutdated
    expr: time() - model_info_last_update > 2592000  # 30 days
    for: 1h
    annotations:
      summary: "Model is outdated"
      description: "Last retrained {{ $value | humanizeDuration }} ago"

# dashboard.json (simplified)
{
  "title": "Financial AI Model Monitoring",
  "panels": [
    {
      "title": "Prediction Latency (P95)",
      "targets": [
        {
          "expr": "histogram_quantile(0.95, rate(prediction_latency_bucket[5m]))",
          "legendFormat": "P95 latency"
        }
      ],
      "type": "graph",
      "yaxes": [{"format": "s"}]
    },
    {
      "title": "Error Rate",
      "targets": [
        {
          "expr": "rate(errors_total[5m]) / rate(predictions_total[5m])",
          "legendFormat": "Error rate"
        }
      ],
      "type": "graph",
      "yaxes": [{"format": "percentunit"}]
    },
    {
      "title": "Model Accuracy (30-day rolling)",
      "targets": [
        {
          "expr": "model_accuracy",
          "legendFormat": "Accuracy"
        },
        {
          "expr": "0.6",
          "legendFormat": "Threshold"
        }
      ],
      "type": "graph",
      "yaxes": [{"format": "percentunit", "min": 0, "max": 1}]
    },
    {
      "title": "Feature Drift (PSI)",
      "targets": [
        {
          "expr": "feature_drift_psi",
          "legendFormat": "{{feature}}"
        }
      ],
      "type": "graph",
      "yaxes": [{"format": "none", "min": 0}]
    },
    {
      "title": "Request Volume",
      "targets": [
        {
          "expr": "rate(predictions_total[1h])",
          "legendFormat": "Predictions per hour"
        }
      ],
      "type": "stat",
      "fieldConfig": {
        "defaults": {
          "unit": "rpm"
        }
      }
    }
  ]
}

class AuditLogger:
    def __init__(self, log_path='./audit_logs'):
        self.log_path = log_path
        os.makedirs(log_path, exist_ok=True)

    def log_prediction(self, request_id, user_id, features, prediction, timestamp, model_version):
        """
        Log prediction for audit purposes.
        """
        entry = {
            'request_id': request_id,
            'user_id': user_id,
            'timestamp': timestamp.isoformat(),
            'model_version': model_version,
            'features': features,
            'prediction': prediction
        }

        with open(f"{self.log_path}/{timestamp.date()}_predictions.jsonl", 'a') as f:
            f.write(json.dumps(entry) + '\n')

    def log_retraining(self, old_version, new_version, data_range, performance_metrics):
        """
        Log retraining events.
        """
        entry = {
            'timestamp': datetime.now().isoformat(),
            'old_version': old_version,
            'new_version': new_version,
            'data_range': data_range,
            'performance_metrics': performance_metrics
        }

        with open(f"{self.log_path}/retraining_log.jsonl", 'a') as f:
            f.write(json.dumps(entry) + '\n')

    def log_model_change(self, model_version, change_type, changed_by, reason):
        """
        Log model changes.
        """
        entry = {
            'timestamp': datetime.now().isoformat(),
            'model_version': model_version,
            'change_type': change_type,
            'changed_by': changed_by,
            'reason': reason
        }

        with open(f"{self.log_path}/model_changes.jsonl", 'a') as f:
            f.write(json.dumps(entry) + '\n')

    def query_audit(self, start_date, end_date, request_id=None):
        """
        Query audit logs for a date range.
        """
        results = []
        for log_file in os.listdir(self.log_path):
            if not log_file.endswith('.jsonl'):
                continue

            # Parse date from filename
            try:
                file_date = datetime.strptime(log_file[:10], '%Y-%m-%d').date()
            except:
                continue

            if start_date <= file_date <= end_date:
                with open(f"{self.log_path}/{log_file}", 'r') as f:
                    for line in f:
                        entry = json.loads(line)
                        if request_id is None or entry.get('request_id') == request_id:
                            results.append(entry)

        return results

class BackupManager:
    def __init__(self, connection_string, backup_path='./backups'):
        self.connection_string = connection_string
        self.backup_path = backup_path
        os.makedirs(backup_path, exist_ok=True)

    def create_backup(self):
        """
        Create a database backup.
        """
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        backup_file = f"{self.backup_path}/backup_{timestamp}.sql"

        import subprocess
        subprocess.run([
            'pg_dump',
            self.connection_string,
            '-f', backup_file,
            '-F', 'c'  # Custom format (compressed)
        ], check=True)

        # Also backup model artifacts
        subprocess.run([
            'tar', '-czf',
            f"{self.backup_path}/models_{timestamp}.tar.gz",
            '/app/models'
        ], check=True)

        return backup_file

    def restore_backup(self, backup_file):
        """
        Restore a database backup.
        """
        import subprocess
        subprocess.run([
            'pg_restore',
            self.connection_string,
            '-c',  # Clean (drop) before restoring
            backup_file
        ], check=True)

# Multi-region configuration
apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: financial-ai-model
spec:
  replicas: 6
  strategy:
    blueGreen:
      activeService: model-service
      previewService: model-service-preview
      autoPromotionEnabled: true
  template:
    spec:
      containers:
      - name: model-server
        image: financial-ai:latest
        env:
        - name: REGION
          value: "us-east-1"
---
# Service for each region
apiVersion: v1
kind: Service
metadata:
  name: model-service-us-east
  labels:
    region: us-east
spec:
  selector:
    region: us-east
  ports:
  - port: 80
    targetPort: 8000
---
apiVersion: v1
kind: Service
metadata:
  name: model-service-us-west
  labels:
    region: us-west
spec:
  selector:
    region: us-west
  ports:
  - port: 80
    targetPort: 8000

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = 'CLOSED'  # CLOSED, OPEN, HALF_OPEN

    def call(self, func, *args, **kwargs):
        """
        Execute a function with circuit breaker protection.
        """
        if self.state == 'OPEN':
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = 'HALF_OPEN'
                print("Circuit breaker: Entering HALF_OPEN state")
            else:
                raise Exception("Circuit breaker is OPEN")

        try:
            result = func(*args, **kwargs)
            if self.state == 'HALF_OPEN':
                self.state = 'CLOSED'
                self.failure_count = 0
                print("Circuit breaker: Recovered to CLOSED state")
            return result

        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()

            if self.failure_count >= self.failure_threshold:
                self.state = 'OPEN'
                print(f"Circuit breaker: OPEN (failures: {self.failure_count})")

            raise e

# Usage
cb = CircuitBreaker(failure_threshold=3, recovery_timeout=30)

try:
    result = cb.call(model.predict, input_data)
except Exception as e:
    # Fallback: use cached prediction or simpler model
    result = fallback_model.predict(input_data)