building volatility regime detection

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class VolatilityRegime:
    def __init__(self, vix_data):
        self.vix = vix_data
        self.current_regime = None

    def calculate_regime(self):
        """Determine current volatility regime"""
        current_vix = self.vix.iloc[-1]['close']
        prev_vix = self.vix.iloc[-2]['close']

        # Calculate 1-day and 5-day change
        vix_1d_change = (current_vix - prev_vix) / prev_vix
        vix_5d = self.vix.iloc[-5:]['close']
        vix_5d_change = (current_vix - vix_5d.iloc[0]) / vix_5d.iloc[0]

        # Determine regime
        if current_vix > 35 or vix_1d_change > 0.15:
            regime = 'spike'
            confidence = 0.95
        elif current_vix > 25:
            if vix_1d_change > 0.10:
                regime = 'spike'
                confidence = 0.85
            else:
                regime = 'elevated'
                confidence = 0.90
        elif current_vix > 15:
            regime = 'normal'
            confidence = 0.85
        else:
            regime = 'low'
            confidence = 0.90

        return {
            'regime': regime,
            'vix': current_vix,
            'vix_1d_change': vix_1d_change,
            'vix_5d_change': vix_5d_change,
            'confidence': confidence,
            'timestamp': datetime.now()
        }

    def should_trade(self):
        """Check if current regime allows trading"""
        regime_info = self.calculate_regime()

        trading_allowed = {
            'low': False,      # Premium too cheap
            'normal': True,    # Trade normally
            'elevated': True,  # Trade aggressively
            'spike': False     # Stop all trading
        }

        return trading_allowed[regime_info['regime']], regime_info

# Load VIX data
vix_data = load_vix_history('2023-01-01', '2023-05-26')

# Test regime detection
detector = VolatilityRegime(vix_data)

# Backtest: what would have happened if I followed regime signals
trades = load_my_trades()
filtered_trades = []

for trade in trades:
    trade_date = trade['entry_date']
    vix_at_entry = vix_data[vix_data['date'] == trade_date]

    detector_at_entry = VolatilityRegime(vix_data[:trade_date])
    should_trade, regime_info = detector_at_entry.should_trade()

    if should_trade:
        filtered_trades.append(trade)
    else:
        print(f"Would have skipped: {trade_date} (regime: {regime_info['regime']})")

# Compare performance
original_pnl = sum(t['pnl'] for t in trades)
filtered_pnl = sum(t['pnl'] for t in filtered_trades)

print(f"Original P&L: ${original_pnl:.0f}")
print(f"Filtered P&L: ${filtered_pnl:.0f}")
print(f"Improvement: ${filtered_pnl - original_pnl:.0f}")

import asyncio
import aiohttp

class RealtimeVolMonitor:
    def __init__(self, polygon_key):
        self.polygon_key = polygon_key
        self.current_regime = None
        self.last_check = None

    async def monitor_loop(self):
        """Check VIX every 5 minutes during market hours"""
        while True:
            if self.is_market_hours():
                await self.check_regime()
                await asyncio.sleep(300)  # 5 minutes
            else:
                await asyncio.sleep(3600)  # 1 hour when market closed

    async def check_regime(self):
        """Fetch current VIX and update regime"""
        vix_data = await self.fetch_vix()
        detector = VolatilityRegime(vix_data)

        should_trade, regime_info = detector.should_trade()

        # If regime changed, take action
        if regime_info['regime'] != self.current_regime:
            await self.handle_regime_change(regime_info)

        self.current_regime = regime_info['regime']
        self.last_check = datetime.now()

    async def handle_regime_change(self, regime_info):
        """Take action when regime changes"""
        print(f"Regime changed to: {regime_info['regime']}")

        if regime_info['regime'] == 'spike':
            # Stop all new trades
            await self.pause_strategy()
            await self.send_alert(f"VOL SPIKE: VIX={regime_info['vix']:.1f}, pausing trading")

        elif regime_info['regime'] == 'low':
            # Reduce trading
            await self.reduce_position_sizes()
            await self.send_alert(f"LOW VOL: VIX={regime_info['vix']:.1f}, reducing size")

        elif self.current_regime in ['spike', 'low'] and regime_info['regime'] in ['normal', 'elevated']:
            # Resume trading
            await self.resume_strategy()
            await self.send_alert(f"VOL NORMALIZED: VIX={regime_info['vix']:.1f}, resuming trading")

    async def fetch_vix(self):
        """Get latest VIX data from Polygon"""
        url = "https://api.polygon.io/v2/aggs/ticker/I:VIX/range/1/day/{start}/{end}"
        # ... fetch logic
        pass

    async def send_alert(self, message):
        """Send pushover notification"""
        # ... notification logic
        pass

class PremiumSellingStrategy:
    def __init__(self):
        self.vol_monitor = RealtimeVolMonitor(POLYGON_KEY)
        self.trading_paused = False

    async def run(self):
        # Start vol monitoring in background
        asyncio.create_task(self.vol_monitor.monitor_loop())

        while True:
            should_trade, regime_info = await self.vol_monitor.should_trade()

            if should_trade and not self.trading_paused:
                signals = self.generate_signals()
                await self.execute_signals(signals)
            else:
                print(f"Trading paused (regime: {regime_info['regime']})")

            await asyncio.sleep(300)  # Check every 5 minutes