year-end portfolio rebalancing algo - detecting institutional flows

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

class RebalanceSignal(Enum):
    STRONG_SELL_PRESSURE = "strong_sell"
    MILD_SELL_PRESSURE = "mild_sell"
    NEUTRAL = "neutral"
    MILD_BUY_PRESSURE = "mild_buy"
    STRONG_BUY_PRESSURE = "strong_buy"

@dataclass
class YTDPerformance:
    symbol: str
    ytd_return: float
    current_weight: float
    target_weight: float
    rebalance_direction: str
    estimated_flow: float  # in millions

class YearEndRebalanceAlgo:
    def __init__(self,
                 ytd_winner_threshold: float = 0.25,  # 25%+ = winner
                 ytd_loser_threshold: float = -0.15,  # -15%+ = tax loss
                 weight_drift_threshold: float = 0.03,  # 3% drift triggers rebal
                 lookback_years: int = 5):
        self.ytd_winner_threshold = ytd_winner_threshold
        self.ytd_loser_threshold = ytd_loser_threshold
        self.weight_drift_threshold = weight_drift_threshold
        self.lookback_years = lookback_years

        # Historical december patterns (from backtesting)
        self.historical_patterns = {
            'week_1': {'sell_pressure': 0.3, 'volume_multiplier': 1.1},
            'week_2': {'sell_pressure': 0.5, 'volume_multiplier': 1.2},
            'week_3': {'sell_pressure': 0.4, 'volume_multiplier': 1.3},
            'week_4': {'sell_pressure': 0.2, 'volume_multiplier': 0.7},
        }

    def calculate_ytd_performance(self,
                                   prices: pd.DataFrame,
                                   current_date: datetime) -> Dict[str, float]:
        """Calculate YTD return for each symbol"""
        year_start = datetime(current_date.year, 1, 1)
        ytd_returns = {}

        for symbol in prices.columns:
            try:
                start_price = prices.loc[year_start:, symbol].iloc[0]
                current_price = prices.loc[:current_date, symbol].iloc[-1]
                ytd_returns[symbol] = (current_price - start_price) / start_price
            except:
                ytd_returns[symbol] = 0.0

        return ytd_returns

    def estimate_rebalance_flow(self,
                                 symbol: str,
                                 ytd_return: float,
                                 current_weight: float,
                                 target_weight: float,
                                 total_aum_billions: float = 100) -> float:
        """
        Estimate institutional rebalancing flow in millions
        Based on S&P 500 total index fund AUM
        """
        weight_diff = current_weight - target_weight

        # Drift adjustment
        if abs(weight_diff) < self.weight_drift_threshold:
            return 0.0

        # Estimate flow (simplified)
        flow_billions = total_aum_billions * weight_diff
        flow_millions = flow_billions * 1000

        # YTD performance amplifier (winners/losers see more flow)
        if ytd_return > self.ytd_winner_threshold:
            flow_millions *= 1.3  # profit taking amplified
        elif ytd_return < self.ytd_loser_threshold:
            flow_millions *= 1.2  # tax loss selling amplified

        return flow_millions

    def generate_signal(self,
                         symbol: str,
                         ytd_return: float,
                         estimated_flow: float,
                         december_week: int) -> dict:
        """
        Generate trading signal based on expected rebalancing
        """
        historical = self.historical_patterns.get(f'week_{december_week}', {})
        sell_pressure = historical.get('sell_pressure', 0.3)

        # Determine signal strength
        if estimated_flow < -500:  # $500M+ selling expected
            signal = RebalanceSignal.STRONG_SELL_PRESSURE
            action = 'FADE_SELLING'  # Buy into weakness
            confidence = min(0.8, abs(estimated_flow) / 1000)
        elif estimated_flow < -100:
            signal = RebalanceSignal.MILD_SELL_PRESSURE
            action = 'SMALL_FADE'
            confidence = 0.5
        elif estimated_flow > 500:  # $500M+ buying expected
            signal = RebalanceSignal.STRONG_BUY_PRESSURE
            action = 'FADE_BUYING'  # Sell into strength
            confidence = min(0.8, estimated_flow / 1000)
        elif estimated_flow > 100:
            signal = RebalanceSignal.MILD_BUY_PRESSURE
            action = 'SMALL_FADE'
            confidence = 0.5
        else:
            signal = RebalanceSignal.NEUTRAL
            action = 'NO_TRADE'
            confidence = 0.0

        return {
            'symbol': symbol,
            'signal': signal.value,
            'action': action,
            'confidence': confidence,
            'ytd_return': ytd_return,
            'estimated_flow_millions': estimated_flow,
            'december_week': december_week,
            'historical_sell_pressure': sell_pressure,
            'reasoning': self._generate_reasoning(signal, ytd_return, estimated_flow)
        }

    def _generate_reasoning(self, signal: RebalanceSignal,
                            ytd_return: float, flow: float) -> str:
        if signal == RebalanceSignal.STRONG_SELL_PRESSURE:
            return f"YTD {ytd_return:.1%}, expected ${abs(flow):.0f}M institutional selling"
        elif signal == RebalanceSignal.STRONG_BUY_PRESSURE:
            return f"YTD {ytd_return:.1%}, expected ${flow:.0f}M institutional buying"
        else:
            return f"YTD {ytd_return:.1%}, minimal rebalancing expected"

    def scan_sp500_rebalancing(self,
                                 sp500_data: pd.DataFrame,
                                 current_date: datetime) -> List[dict]:
        """
        Scan S&P 500 for rebalancing opportunities
        Returns sorted list of strongest signals
        """
        ytd_returns = self.calculate_ytd_performance(sp500_data, current_date)

        # Determine december week
        december_week = min(4, (current_date.day - 1) // 7 + 1)

        opportunities = []

        for symbol, ytd_return in ytd_returns.items():
            # Simplified weight estimation (would use actual index data)
            current_weight = 0.002  # placeholder
            target_weight = 0.002

            if abs(ytd_return) > 0.10:  # Only significant movers
                estimated_flow = self.estimate_rebalance_flow(
                    symbol, ytd_return, current_weight, target_weight
                )

                signal = self.generate_signal(
                    symbol, ytd_return, estimated_flow, december_week
                )

                if signal['action'] != 'NO_TRADE':
                    opportunities.append(signal)

        # Sort by confidence
        opportunities.sort(key=lambda x: x['confidence'], reverse=True)

        return opportunities[:20]  # Top 20 opportunities