order flow analysis - adapting strategies for summer thin volume

import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timedelta

@dataclass
class OrderFlowMetrics:
    """Core order flow metrics for strategy signals"""

    timestamp: datetime
    symbol: str

    # Volume metrics
    total_volume: int
    buy_volume: int
    sell_volume: int
    delta: int  # buy - sell
    cumulative_delta: int

    # Imbalance metrics
    bid_volume: int
    ask_volume: int
    imbalance_ratio: float  # (bid - ask) / (bid + ask)

    # Aggression metrics
    aggressive_buys: int
    aggressive_sells: int
    aggression_ratio: float

    # Liquidity metrics
    bid_depth: float
    ask_depth: float
    spread_ticks: float

    @property
    def is_thin_volume(self) -> bool:
        """Detect thin volume conditions"""
        return self.total_volume < self.volume_threshold * 0.6

    @property
    def volume_threshold(self) -> int:
        """Energetic threshold based on time of day"""
        hour = self.timestamp.hour
        if 9 <= hour <= 10:  # opening hour
            return 50000
        elif 15 <= hour <= 16:  # closing hour
            return 45000
        else:  # midday
            return 30000


class SummerOrderFlowAnalyzer:
    """
    Order flow analysis adapted for summer thin volume conditions
    """

    def __init__(self, lookback_days: int = 20):
        self.lookback_days = lookback_days
        self.metrics_history: list[OrderFlowMetrics] = []
        self.volume_baseline: Optional[float] = None
        self.summer_adjustment_factor = 1.0

    def calculate_volume_baseline(self, historical_data: pd.DataFrame) -> float:
        """
        Calculate rolling volume baseline for comparison
        """
        # Use 20-day rolling average
        rolling_vol = historical_data['volume'].rolling(
            window=self.lookback_days
        ).mean()

        self.volume_baseline = rolling_vol.iloc[-1]
        return self.volume_baseline

    def detect_summer_regime(self, current_volume: int) -> dict:
        """
        Detect if we're in summer thin volume regime
        Returns adjustment factors for signals
        """
        if self.volume_baseline is None:
            raise ValueError("Must calculate baseline first")

        volume_ratio = current_volume / self.volume_baseline

        if volume_ratio < 0.5:
            # Extremely thin - high caution
            regime = 'extremely_thin'
            signal_threshold_mult = 1.8  # require 80% stronger signals
            size_mult = 0.5  # half position size

        elif volume_ratio < 0.7:
            # Thin - moderate caution
            regime = 'thin'
            signal_threshold_mult = 1.4  # require 40% stronger signals
            size_mult = 0.7

        elif volume_ratio < 0.9:
            # Below normal - slight caution
            regime = 'below_normal'
            signal_threshold_mult = 1.2
            size_mult = 0.85

        else:
            # Normal conditions
            regime = 'normal'
            signal_threshold_mult = 1.0
            size_mult = 1.0

        self.summer_adjustment_factor = signal_threshold_mult

        return {
            'regime': regime,
            'volume_ratio': volume_ratio,
            'signal_threshold_multiplier': signal_threshold_mult,
            'position_size_multiplier': size_mult,
            'confidence_adjustment': 1.0 / signal_threshold_mult
        }

    def calculate_delta_signal(
        self,
        metrics: OrderFlowMetrics,
        lookback_bars: int = 20
    ) -> dict:
        """
        Calculate delta-based signal with summer adjustments
        """
        # Get recent cumulative delta
        recent_metrics = self.metrics_history[-lookback_bars:]

        if len(recent_metrics) < lookback_bars:
            return {'signal': 'insufficient_data', 'strength': 0}

        deltas = [m.cumulative_delta for m in recent_metrics]
        delta_mean = np.mean(deltas)
        delta_std = np.std(deltas)

        if delta_std == 0:
            return {'signal': 'no_variance', 'strength': 0}

        # Z-score of current delta
        current_z = (metrics.cumulative_delta - delta_mean) / delta_std

        # Apply summer adjustment - require stronger signal
        adjusted_threshold = 2.0 * self.summer_adjustment_factor

        if current_z > adjusted_threshold:
            signal = 'strong_buy'
            strength = min((current_z - adjusted_threshold) / 2, 1.0)
        elif current_z < -adjusted_threshold:
            signal = 'strong_sell'
            strength = min((-current_z - adjusted_threshold) / 2, 1.0)
        elif current_z > 1.5 * self.summer_adjustment_factor:
            signal = 'weak_buy'
            strength = 0.3
        elif current_z < -1.5 * self.summer_adjustment_factor:
            signal = 'weak_sell'
            strength = 0.3
        else:
            signal = 'neutral'
            strength = 0

        return {
            'signal': signal,
            'strength': strength,
            'z_score': current_z,
            'threshold_used': adjusted_threshold,
            'summer_adjusted': self.summer_adjustment_factor > 1.0
        }

    def calculate_imbalance_signal(
        self,
        metrics: OrderFlowMetrics
    ) -> dict:
        """
        Calculate order book imbalance signal
        More sensitive in thin markets - need stronger imbalance
        """
        imbalance = metrics.imbalance_ratio

        # Summer adjustment - require larger imbalance
        base_threshold = 0.3
        adjusted_threshold = base_threshold * self.summer_adjustment_factor

        if abs(imbalance) < adjusted_threshold:
            return {
                'signal': 'neutral',
                'imbalance': imbalance,
                'threshold': adjusted_threshold
            }

        if imbalance > adjusted_threshold:
            return {
                'signal': 'bid_dominant',  # potential buying pressure
                'imbalance': imbalance,
                'confidence': min((imbalance - adjusted_threshold) / 0.3, 1.0)
            }
        else:
            return {
                'signal': 'ask_dominant',  # potential selling pressure
                'imbalance': imbalance,
                'confidence': min((-imbalance - adjusted_threshold) / 0.3, 1.0)
            }

    def filter_false_breakouts(
        self,
        price_data: pd.DataFrame,
        volume_data: pd.DataFrame
    ) -> pd.DataFrame:
        """
        Filter potential false breakouts in thin volume

        Summer markets have more false breakouts due to:
        - Thin order books creating gaps
        - Large orders moving price disproportionately
        - Reduced institutional participation
        """
        df = price_data.copy()

        # Calculate returns
        df['returns'] = df['close'].pct_change()

        # Calculate volume relative to baseline
        df['volume_ratio'] = volume_data['volume'] / self.volume_baseline

        # Flag potential false breakouts
        # Large move on thin volume = suspicious
        df['false_breakout_risk'] = (
            (abs(df['returns']) > df['returns'].rolling(20).std() * 2) &
            (df['volume_ratio'] < 0.6)
        )

        # Calculate confidence adjustment
        df['signal_confidence'] = np.where(
            df['false_breakout_risk'],
            0.5,  # reduce confidence for suspicious moves
            1.0
        )

        return df

    def get_trading_recommendation(
        self,
        metrics: OrderFlowMetrics,
        price_data: pd.DataFrame
    ) -> dict:
        """
        Combine all signals into trading recommendation
        """
        # Store metrics
        self.metrics_history.append(metrics)

        # Get individual signals
        delta_signal = self.calculate_delta_signal(metrics)
        imbalance_signal = self.calculate_imbalance_signal(metrics)

        # Filter for false breakouts
        filtered_data = self.filter_false_breakouts(
            price_data,
            pd.DataFrame({'volume': [metrics.total_volume]})
        )

        # Combine signals
        signals = [delta_signal['signal'], imbalance_signal['signal']]

        # Count directional agreement
        bullish_count = sum(1 for s in signals if 'buy' in s or 'bid' in s)
        bearish_count = sum(1 for s in signals if 'sell' in s or 'ask' in s)

        # Require agreement in thin markets
        if self.summer_adjustment_factor > 1.2:
            required_agreement = 2  # all signals must agree
        else:
            required_agreement = 1  # normal threshold

        if bullish_count >= required_agreement:
            direction = 'long'
            confidence = delta_signal['strength'] * imbalance_signal.get('confidence', 0.5)
        elif bearish_count >= required_agreement:
            direction = 'short'
            confidence = delta_signal['strength'] * imbalance_signal.get('confidence', 0.5)
        else:
            direction = 'flat'
            confidence = 0

        # Apply thin volume confidence reduction
        if metrics.is_thin_volume:
            confidence *= 0.7

        return {
            'direction': direction,
            'confidence': confidence,
            'delta_signal': delta_signal,
            'imbalance_signal': imbalance_signal,
            'thin_volume_warning': metrics.is_thin_volume,
            'summer_regime': self.summer_adjustment_factor > 1.0,
            'recommendation': self._format_recommendation(direction, confidence)
        }

    def _format_recommendation(self, direction: str, confidence: float) -> str:
        if confidence < 0.3:
            return f"SKIP - Low confidence ({confidence:.1%})"
        elif confidence < 0.5:
            return f"MARGINAL {direction.upper()} - Consider smaller size"
        elif confidence < 0.7:
            return f"MODERATE {direction.upper()} - Standard size"
        else:
            return f"STRONG {direction.upper()} - Full size"


# Usage example
analyzer = SummerOrderFlowAnalyzer(lookback_days=20)

# Calculate baseline from historical data
historical_volume = pd.DataFrame({
    'volume': np.random.normal(4200000, 500000, 60)  # 60 days history
})
analyzer.calculate_volume_baseline(historical_volume)

# Detect current regime (july thin volume)
current_volume = 2800000  # typical july volume
regime = analyzer.detect_summer_regime(current_volume)

print(f"Regime: {regime['regime']}")
print(f"Signal threshold multiplier: {regime['signal_threshold_multiplier']:.1f}x")
print(f"Position size multiplier: {regime['position_size_multiplier']:.0%}")