Real-Time Options Chain Analysis: Ollama CBOE Data Processing Guide

# Install Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# Start Ollama service
ollama serve

# Pull required model for data processing
ollama pull llama2:7b

# requirements.txt
import requests
import pandas as pd
import numpy as np
import asyncio
import websockets
import json
from datetime import datetime, timedelta
import ollama

# Install dependencies
pip install -r requirements.txt

import asyncio
import websockets
import json

class CBOEDataStream:
    def __init__(self, api_key, symbols):
        self.api_key = api_key
        self.symbols = symbols
        self.connection = None
        
    async def connect(self):
        """Establish WebSocket connection to CBOE data feed"""
        uri = f"wss://api.cboe.com/v1/stream?key={self.api_key}"
        self.connection = await websockets.connect(uri)
        
        # Subscribe to options chain data
        subscription = {
            "action": "subscribe",
            "symbols": self.symbols,
            "data_type": "options_chain"
        }
        await self.connection.send(json.dumps(subscription))
        
    async def stream_data(self):
        """Process incoming options data"""
        async for message in self.connection:
            data = json.loads(message)
            yield self.parse_options_data(data)
            
    def parse_options_data(self, raw_data):
        """Convert raw CBOE data to structured format"""
        return {
            'symbol': raw_data['symbol'],
            'strike': raw_data['strike'],
            'expiration': raw_data['expiration'],
            'option_type': raw_data['type'],
            'bid': raw_data['bid'],
            'ask': raw_data['ask'],
            'volume': raw_data['volume'],
            'open_interest': raw_data['open_interest'],
            'timestamp': datetime.now()
        }

import ollama
import pandas as pd

class OptionsAnalyzer:
    def __init__(self, model_name="llama2:7b"):
        self.model = model_name
        self.options_data = []
        
    def analyze_options_flow(self, data_batch):
        """Analyze options flow patterns using Ollama"""
        # Prepare data for analysis
        df = pd.DataFrame(data_batch)
        
        # Calculate key metrics
        metrics = self.calculate_flow_metrics(df)
        
        # Generate analysis prompt
        prompt = self.create_analysis_prompt(metrics)
        
        # Get Ollama analysis
        response = ollama.generate(
            model=self.model,
            prompt=prompt
        )
        
        return self.parse_analysis_response(response['response'])
        
    def calculate_flow_metrics(self, df):
        """Calculate options flow metrics"""
        return {
            'total_volume': df['volume'].sum(),
            'call_put_ratio': self.get_call_put_ratio(df),
            'avg_implied_volatility': df['implied_vol'].mean(),
            'unusual_activity': self.detect_unusual_volume(df),
            'price_momentum': self.calculate_momentum(df)
        }
        
    def create_analysis_prompt(self, metrics):
        """Create structured prompt for Ollama analysis"""
        return f"""
        Analyze these options flow metrics:
        - Total Volume: {metrics['total_volume']}
        - Call/Put Ratio: {metrics['call_put_ratio']}
        - Average IV: {metrics['avg_implied_volatility']}
        - Unusual Activity: {metrics['unusual_activity']}
        - Price Momentum: {metrics['price_momentum']}
        
        Provide:
        1. Market sentiment (bullish/bearish/neutral)
        2. Key support/resistance levels
        3. Recommended trading actions
        4. Risk assessment
        """
        
    def parse_analysis_response(self, response):
        """Parse Ollama response into structured data"""
        # Extract key insights from response
        lines = response.split('\n')
        
        analysis = {
            'sentiment': self.extract_sentiment(lines),
            'support_resistance': self.extract_levels(lines),
            'recommendations': self.extract_recommendations(lines),
            'risk_level': self.extract_risk_level(lines),
            'timestamp': datetime.now()
        }
        
        return analysis

class PatternDetector:
    def __init__(self):
        self.patterns = {
            'gamma_squeeze': self.detect_gamma_squeeze,
            'vol_spike': self.detect_volatility_spike,
            'flow_divergence': self.detect_flow_divergence,
            'pin_risk': self.detect_pin_risk
        }
        
    def detect_gamma_squeeze(self, options_data):
        """Identify potential gamma squeeze conditions"""
        # Calculate gamma exposure by strike
        gamma_exposure = {}
        
        for option in options_data:
            strike = option['strike']
            gamma = option['gamma']
            open_interest = option['open_interest']
            
            # Calculate dealer gamma exposure
            exposure = gamma * open_interest * 100
            gamma_exposure[strike] = gamma_exposure.get(strike, 0) + exposure
            
        # Find maximum gamma concentration
        max_gamma_strike = max(gamma_exposure, key=gamma_exposure.get)
        max_gamma_value = gamma_exposure[max_gamma_strike]
        
        # Determine squeeze probability
        squeeze_threshold = 1000000  # Adjust based on underlying
        
        return {
            'detected': max_gamma_value > squeeze_threshold,
            'key_strike': max_gamma_strike,
            'gamma_exposure': max_gamma_value,
            'probability': min(max_gamma_value / squeeze_threshold, 1.0)
        }
        
    def detect_volatility_spike(self, options_data):
        """Identify unusual volatility patterns"""
        current_iv = [opt['implied_vol'] for opt in options_data]
        avg_iv = np.mean(current_iv)
        
        # Compare to historical volatility
        historical_avg = self.get_historical_iv_avg()
        
        spike_ratio = avg_iv / historical_avg
        
        return {
            'detected': spike_ratio > 1.5,
            'current_iv': avg_iv,
            'historical_iv': historical_avg,
            'spike_ratio': spike_ratio
        }

from flask import Flask, render_template, jsonify
import threading
import queue

app = Flask(__name__)
data_queue = queue.Queue()

class DashboardServer:
    def __init__(self, analyzer, detector):
        self.analyzer = analyzer
        self.detector = detector
        self.latest_data = {}
        
    def start_processing(self):
        """Start background data processing"""
        def process_loop():
            while True:
                try:
                    data_batch = data_queue.get(timeout=1)
                    
                    # Run analysis
                    analysis = self.analyzer.analyze_options_flow(data_batch)
                    patterns = self.detector.detect_all_patterns(data_batch)
                    
                    # Update dashboard data
                    self.latest_data = {
                        'analysis': analysis,
                        'patterns': patterns,
                        'timestamp': datetime.now().isoformat()
                    }
                    
                except queue.Empty:
                    continue
                    
        thread = threading.Thread(target=process_loop)
        thread.daemon = True
        thread.start()
        
    @app.route('/')
    def dashboard():
        return render_template('dashboard.html')
        
    @app.route('/api/data')
    def get_data():
        return jsonify(self.latest_data)
        
    def run(self):
        app.run(debug=True, port=5000)

<!-- dashboard.html -->
<!DOCTYPE html>
<html>
<head>
    <title>Real-Time Options Analysis Dashboard</title>
    <script src="https://cdn.plot.ly/plotly-latest.min.js"></script>
    <style>
        .metric-card {
            background: #f8f9fa;
            border: 1px solid #dee2e6;
            border-radius: 8px;
            padding: 20px;
            margin: 10px;
            display: inline-block;
            width: 300px;
        }
        
        .alert {
            padding: 10px;
            margin: 10px 0;
            border-radius: 4px;
            font-weight: bold;
        }
        
        .alert-danger { background-color: #f8d7da; color: #721c24; }
        .alert-warning { background-color: #fff3cd; color: #856404; }
        .alert-success { background-color: #d4edda; color: #155724; }
    </style>
</head>
<body>
    <h1>Real-Time Options Chain Analysis</h1>
    
    <div id="metrics-container">
        <div class="metric-card">
            <h3>Market Sentiment</h3>
            <div id="sentiment-value">Loading...</div>
        </div>
        
        <div class="metric-card">
            <h3>Call/Put Ratio</h3>
            <div id="ratio-value">Loading...</div>
        </div>
        
        <div class="metric-card">
            <h3>Implied Volatility</h3>
            <div id="iv-value">Loading...</div>
        </div>
    </div>
    
    <div id="alerts-container">
        <h2>Pattern Alerts</h2>
        <div id="alerts-list"></div>
    </div>
    
    <div id="chart-container">
        <div id="options-flow-chart"></div>
    </div>
    
    <script>
        // Update dashboard every second
        setInterval(updateDashboard, 1000);
        
        function updateDashboard() {
            fetch('/api/data')
                .then(response => response.json())
                .then(data => {
                    updateMetrics(data.analysis);
                    updateAlerts(data.patterns);
                    updateChart(data);
                });
        }
        
        function updateMetrics(analysis) {
            document.getElementById('sentiment-value').textContent = analysis.sentiment;
            document.getElementById('ratio-value').textContent = analysis.call_put_ratio;
            document.getElementById('iv-value').textContent = analysis.avg_implied_volatility;
        }
        
        function updateAlerts(patterns) {
            const alertsContainer = document.getElementById('alerts-list');
            alertsContainer.innerHTML = '';
            
            for (const [pattern, data] of Object.entries(patterns)) {
                if (data.detected) {
                    const alert = document.createElement('div');
                    alert.className = 'alert alert-warning';
                    alert.textContent = `${pattern.toUpperCase()} detected - Probability: ${data.probability}`;
                    alertsContainer.appendChild(alert);
                }
            }
        }
    </script>
</body>
</html>

import gc
from collections import deque

class OptimizedDataProcessor:
    def __init__(self, max_buffer_size=10000):
        self.data_buffer = deque(maxlen=max_buffer_size)
        self.processing_batch_size = 100
        
    def process_efficiently(self, data_stream):
        """Process data in optimized batches"""
        batch = []
        
        for data_point in data_stream:
            batch.append(data_point)
            
            if len(batch) >= self.processing_batch_size:
                # Process batch
                results = self.analyze_batch(batch)
                
                # Clear processed data
                batch.clear()
                
                # Force garbage collection periodically
                if len(self.data_buffer) % 1000 == 0:
                    gc.collect()
                    
                yield results
                
    def analyze_batch(self, batch):
        """Analyze data batch with memory optimization"""
        # Convert to numpy for faster processing
        data_array = np.array(batch)
        
        # Vectorized calculations
        results = {
            'volume_weighted_price': np.average(
                data_array[:, 'price'], 
                weights=data_array[:, 'volume']
            ),
            'volatility_estimate': np.std(data_array[:, 'price']),
            'momentum_score': self.calculate_momentum_vectorized(data_array)
        }
        
        return results

import sqlite3
import threading
from contextlib import contextmanager

class OptimizedDatabase:
    def __init__(self, db_path):
        self.db_path = db_path
        self.local_storage = threading.local()
        self.setup_database()
        
    @contextmanager
    def get_connection(self):
        """Thread-safe database connection"""
        if not hasattr(self.local_storage, 'connection'):
            self.local_storage.connection = sqlite3.connect(self.db_path)
            self.local_storage.connection.execute('PRAGMA journal_mode=WAL')
            self.local_storage.connection.execute('PRAGMA synchronous=NORMAL')
            
        yield self.local_storage.connection
        
    def batch_insert_options(self, options_data):
        """Optimized batch insertion"""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            
            # Prepare batch insert
            insert_sql = """
            INSERT INTO options_data 
            (symbol, strike, expiration, option_type, bid, ask, volume, timestamp)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """
            
            # Convert data to tuples
            data_tuples = [
                (opt['symbol'], opt['strike'], opt['expiration'], 
                 opt['option_type'], opt['bid'], opt['ask'], 
                 opt['volume'], opt['timestamp'])
                for opt in options_data
            ]
            
            # Execute batch insert
            cursor.executemany(insert_sql, data_tuples)
            conn.commit()

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    curl \
    && rm -rf /var/lib/apt/lists/*

# Install Ollama
RUN curl -fsSL https://ollama.ai/install.sh | sh

# Copy application files
COPY requirements.txt .
COPY src/ ./src/

# Install Python dependencies
RUN pip install --no-cache-dir -r requirements.txt

# Pull Ollama model
RUN ollama serve & \
    sleep 10 && \
    ollama pull llama2:7b

EXPOSE 5000

CMD ["python", "src/main.py"]

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

class SystemMonitor:
    def __init__(self):
        self.setup_metrics()
        self.setup_logging()
        
    def setup_metrics(self):
        """Initialize Prometheus metrics"""
        self.processed_options = Counter('options_processed_total', 'Total options processed')
        self.processing_time = Histogram('processing_time_seconds', 'Time spent processing')
        self.active_connections = Gauge('active_connections', 'Active WebSocket connections')
        self.error_count = Counter('errors_total', 'Total errors', ['type'])
        
    def setup_logging(self):
        """Configure structured logging"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('options_analysis.log'),
                logging.StreamHandler()
            ]
        )
        
    def track_processing(self, func):
        """Decorator to track function performance"""
        def wrapper(*args, **kwargs):
            start_time = time.time()
            
            try:
                result = func(*args, **kwargs)
                self.processed_options.inc()
                return result
                
            except Exception as e:
                self.error_count.labels(type=type(e).__name__).inc()
                logging.error(f"Processing error: {e}")
                raise
                
            finally:
                self.processing_time.observe(time.time() - start_time)
                
        return wrapper
        
    def start_metrics_server(self):
        """Start Prometheus metrics server"""
        start_http_server(8000)
        logging.info("Metrics server started on port 8000")

import asyncio
import logging

class ReconnectingDataStream:
    def __init__(self, stream_config):
        self.config = stream_config
        self.max_retries = 5
        self.retry_delay = 1
        
    async def maintain_connection(self):
        """Maintain persistent connection with auto-reconnect"""
        retry_count = 0
        
        while retry_count < self.max_retries:
            try:
                async with CBOEDataStream(self.config) as stream:
                    retry_count = 0  # Reset on successful connection
                    
                    async for data in stream:
                        yield data
                        
            except ConnectionError as e:
                retry_count += 1
                logging.warning(f"Connection lost, retry {retry_count}/{self.max_retries}")
                
                if retry_count < self.max_retries:
                    await asyncio.sleep(self.retry_delay * retry_count)
                else:
                    logging.error("Max retries exceeded")
                    raise

import psutil
import gc

class MemoryManager:
    def __init__(self, max_memory_percent=80):
        self.max_memory_percent = max_memory_percent
        
    def check_memory_usage(self):
        """Monitor memory usage and trigger cleanup"""
        memory_percent = psutil.virtual_memory().percent
        
        if memory_percent > self.max_memory_percent:
            logging.warning(f"High memory usage: {memory_percent}%")
            self.cleanup_memory()
            
    def cleanup_memory(self):
        """Force garbage collection and clear caches"""
        gc.collect()
        
        # Clear specific caches
        if hasattr(self, 'options_cache'):
            self.options_cache.clear()
            
        logging.info("Memory cleanup completed")

import numpy as np
from scipy.interpolate import RBFInterpolator

class VolatilitySurfaceAnalyzer:
    def __init__(self):
        self.surface_data = {}
        
    def build_vol_surface(self, options_data):
        """Build volatility surface from options data"""
        # Extract strike, expiration, and IV data
        strikes = [opt['strike'] for opt in options_data]
        expirations = [opt['days_to_expiration'] for opt in options_data]
        implied_vols = [opt['implied_vol'] for opt in options_data]
        
        # Create coordinate pairs
        points = np.column_stack([strikes, expirations])
        
        # Build interpolated surface
        interpolator = RBFInterpolator(points, implied_vols, kernel='thin_plate_spline')
        
        return interpolator
        
    def find_vol_arbitrage(self, surface):
        """Identify volatility arbitrage opportunities"""
        opportunities = []
        
        # Check for surface anomalies
        test_strikes = np.linspace(50, 200, 50)
        test_expirations = np.linspace(1, 90, 30)
        
        for strike in test_strikes:
            for expiration in test_expirations:
                predicted_vol = surface([strike, expiration])
                market_vol = self.get_market_vol(strike, expiration)
                
                vol_difference = abs(predicted_vol - market_vol)
                
                if vol_difference > 0.05:  # 5% volatility difference
                    opportunities.append({
                        'strike': strike,
                        'expiration': expiration,
                        'predicted_vol': predicted_vol,
                        'market_vol': market_vol,
                        'arbitrage_potential': vol_difference
                    })
                    
        return opportunities

class DeltaHedger:
    def __init__(self, portfolio, risk_tolerance=0.01):
        self.portfolio = portfolio
        self.risk_tolerance = risk_tolerance
        
    def calculate_portfolio_delta(self):
        """Calculate total portfolio delta"""
        total_delta = 0
        
        for position in self.portfolio.positions:
            delta = position.quantity * position.option.delta
            total_delta += delta
            
        return total_delta
        
    def generate_hedge_orders(self):
        """Generate orders to hedge portfolio delta"""
        current_delta = self.calculate_portfolio_delta()
        
        if abs(current_delta) > self.risk_tolerance:
            # Calculate hedge size
            hedge_quantity = -current_delta
            
            # Generate hedge order
            hedge_order = {
                'symbol': self.portfolio.underlying_symbol,
                'quantity': int(hedge_quantity),
                'order_type': 'market',
                'side': 'buy' if hedge_quantity > 0 else 'sell',
                'reason': 'delta_hedge'
            }
            
            return [hedge_order]
            
        return []