vix term structure algo - contango/backwardation trading

import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import Optional, Tuple, List
from datetime import datetime, timedelta
from enum import Enum

class TermStructureState(Enum):
    STEEP_CONTANGO = "steep_contango"      # VIX1 << VIX2, calm
    MILD_CONTANGO = "mild_contango"        # VIX1 < VIX2, normal
    FLAT = "flat"                          # VIX1 ≈ VIX2, transitional
    MILD_BACKWARDATION = "mild_backwardation"  # VIX1 > VIX2, caution
    STEEP_BACKWARDATION = "steep_backwardation"  # VIX1 >> VIX2, fear

@dataclass
class VIXTermStructure:
    timestamp: datetime
    vix_spot: float
    vix_1m: float  # front month future
    vix_2m: float  # second month future
    vix_3m: float  # third month future
    contango_1_2: float  # spread between months
    contango_2_3: float
    state: TermStructureState

class VIXTermStructureAlgo:
    def __init__(self,
                 steep_threshold: float = 0.10,  # 10% spread = steep
                 mild_threshold: float = 0.03,    # 3% spread = mild
                 flat_threshold: float = 0.01,    # 1% spread = flat
                 lookback_days: int = 20):
        self.steep_threshold = steep_threshold
        self.mild_threshold = mild_threshold
        self.flat_threshold = flat_threshold
        self.lookback_days = lookback_days
        self.history: List[VIXTermStructure] = []

    def calculate_term_structure(self,
                                  vix_spot: float,
                                  vix_1m: float,
                                  vix_2m: float,
                                  vix_3m: float) -> VIXTermStructure:
        """
        Calculate current term structure and classify state
        """
        contango_1_2 = (vix_2m - vix_1m) / vix_1m
        contango_2_3 = (vix_3m - vix_2m) / vix_2m

        # Classify state based on front spread
        if contango_1_2 > self.steep_threshold:
            state = TermStructureState.STEEP_CONTANGO
        elif contango_1_2 > self.mild_threshold:
            state = TermStructureState.MILD_CONTANGO
        elif contango_1_2 > -self.flat_threshold:
            state = TermStructureState.FLAT
        elif contango_1_2 > -self.mild_threshold:
            state = TermStructureState.MILD_BACKWARDATION
        else:
            state = TermStructureState.STEEP_BACKWARDATION

        return VIXTermStructure(
            timestamp=datetime.now(),
            vix_spot=vix_spot,
            vix_1m=vix_1m,
            vix_2m=vix_2m,
            vix_3m=vix_3m,
            contango_1_2=contango_1_2,
            contango_2_3=contango_2_3,
            state=state
        )

    def detect_state_transition(self,
                                 current: VIXTermStructure) -> Optional[dict]:
        """
        Detect when term structure state changes
        These transitions often signal regime shifts
        """
        if len(self.history) < 2:
            return None

        previous = self.history[-1]

        if previous.state == current.state:
            return None  # No transition

        # Map transitions to trading signals
        transition_signals = {
            # From calm to fear = go defensive
            (TermStructureState.STEEP_CONTANGO, TermStructureState.MILD_CONTANGO): {
                'signal': 'CAUTION',
                'action': 'reduce_short_vol',
                'strength': 0.3
            },
            (TermStructureState.MILD_CONTANGO, TermStructureState.FLAT): {
                'signal': 'WARNING',
                'action': 'close_short_vol',
                'strength': 0.6
            },
            (TermStructureState.FLAT, TermStructureState.MILD_BACKWARDATION): {
                'signal': 'DEFENSIVE',
                'action': 'add_long_vol_hedge',
                'strength': 0.8
            },
            (TermStructureState.MILD_BACKWARDATION, TermStructureState.STEEP_BACKWARDATION): {
                'signal': 'CRISIS',
                'action': 'full_defensive',
                'strength': 1.0
            },
            # From fear to calm = opportunities
            (TermStructureState.STEEP_BACKWARDATION, TermStructureState.MILD_BACKWARDATION): {
                'signal': 'RECOVERY_START',
                'action': 'scale_into_short_vol',
                'strength': 0.4
            },
            (TermStructureState.MILD_BACKWARDATION, TermStructureState.FLAT): {
                'signal': 'RECOVERY_BUILDING',
                'action': 'add_short_vol',
                'strength': 0.6
            },
            (TermStructureState.FLAT, TermStructureState.MILD_CONTANGO): {
                'signal': 'NORMAL_RETURNING',
                'action': 'full_short_vol_position',
                'strength': 0.8
            },
        }

        key = (previous.state, current.state)
        return transition_signals.get(key)

    def calculate_roll_yield(self, current: VIXTermStructure) -> float:
        """
        Calculate expected roll yield from contango
        Positive = favorable for short VIX
        Negative = unfavorable for short VIX
        """
        # Annualized roll yield
        monthly_roll = current.contango_1_2
        annualized = monthly_roll * 12

        return annualized

    def generate_signal(self) -> dict:
        """
        Generate trading signal based on term structure
        """
        if len(self.history) < self.lookback_days:
            return {'signal': 'INSUFFICIENT_DATA'}

        current = self.history[-1]
        transition = self.detect_state_transition(current)
        roll_yield = self.calculate_roll_yield(current)

        # Base signal from state
        state_signals = {
            TermStructureState.STEEP_CONTANGO: {
                'bias': 'SHORT_VOL',
                'confidence': 0.8,
                'expected_decay': 0.02  # 2% monthly
            },
            TermStructureState.MILD_CONTANGO: {
                'bias': 'SHORT_VOL',
                'confidence': 0.6,
                'expected_decay': 0.01
            },
            TermStructureState.FLAT: {
                'bias': 'NEUTRAL',
                'confidence': 0.3,
                'expected_decay': 0.0
            },
            TermStructureState.MILD_BACKWARDATION: {
                'bias': 'LONG_VOL',
                'confidence': 0.5,
                'expected_decay': -0.01
            },
            TermStructureState.STEEP_BACKWARDATION: {
                'bias': 'LONG_VOL',
                'confidence': 0.7,
                'expected_decay': -0.02
            },
        }

        base = state_signals[current.state]

        return {
            'timestamp': current.timestamp,
            'state': current.state.value,
            'bias': base['bias'],
            'confidence': base['confidence'],
            'roll_yield_annualized': roll_yield,
            'expected_decay': base['expected_decay'],
            'vix_spot': current.vix_spot,
            'contango_spread': current.contango_1_2,
            'transition': transition
        }

    def add_observation(self, vix_spot: float, vix_1m: float,
                        vix_2m: float, vix_3m: float):
        """Add new observation and update history"""
        ts = self.calculate_term_structure(vix_spot, vix_1m, vix_2m, vix_3m)
        self.history.append(ts)

        # Keep only lookback period
        if len(self.history) > self.lookback_days * 2:
            self.history = self.history[-self.lookback_days * 2:]