sector rotation algo - implementation with relative strength scoring

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

@dataclass
class SectorData:
    symbol: str
    name: str
    etf: str  # sector ETF
    components: List[str]
    weight: float

class SectorRotationAlgo:
    SECTORS = [
        SectorData('XLK', 'Technology', 'XLK', ['AAPL', 'MSFT', 'NVDA'], 0.28),
        SectorData('XLF', 'Financials', 'XLF', ['JPM', 'BAC', 'WFC'], 0.14),
        SectorData('XLV', 'Healthcare', 'XLV', ['UNH', 'JNJ', 'PFE'], 0.13),
        SectorData('XLY', 'Consumer Disc', 'XLY', ['AMZN', 'TSLA', 'HD'], 0.10),
        SectorData('XLC', 'Communication', 'XLC', ['META', 'GOOG', 'NFLX'], 0.09),
        SectorData('XLI', 'Industrials', 'XLI', ['CAT', 'UNP', 'HON'], 0.08),
        SectorData('XLP', 'Consumer Staples', 'XLP', ['PG', 'KO', 'PEP'], 0.06),
        SectorData('XLE', 'Energy', 'XLE', ['XOM', 'CVX', 'COP'], 0.05),
        SectorData('XLU', 'Utilities', 'XLU', ['NEE', 'DUK', 'SO'], 0.03),
        SectorData('XLRE', 'Real Estate', 'XLRE', ['AMT', 'PLD', 'CCI'], 0.02),
        SectorData('XLB', 'Materials', 'XLB', ['LIN', 'APD', 'SHW'], 0.02),
    ]

    def __init__(self,
                 lookback_short: int = 20,
                 lookback_long: int = 60,
                 top_n: int = 3,
                 bottom_n: int = 2,
                 rebalance_threshold: float = 0.05):
        self.lookback_short = lookback_short
        self.lookback_long = lookback_long
        self.top_n = top_n
        self.bottom_n = bottom_n
        self.rebalance_threshold = rebalance_threshold

    def calculate_relative_strength(self,
                                     sector_returns: pd.DataFrame,
                                     benchmark_returns: pd.Series) -> pd.DataFrame:
        """
        Calculate multi-timeframe relative strength for each sector
        Returns DataFrame with RS scores by sector
        """
        rs_scores = pd.DataFrame(index=sector_returns.index)

        for sector in sector_returns.columns:
            # Short-term RS (20-day)
            sector_short = sector_returns[sector].rolling(
                self.lookback_short
            ).sum()
            bench_short = benchmark_returns.rolling(
                self.lookback_short
            ).sum()
            rs_short = sector_short - bench_short

            # Long-term RS (60-day)
            sector_long = sector_returns[sector].rolling(
                self.lookback_long
            ).sum()
            bench_long = benchmark_returns.rolling(
                self.lookback_long
            ).sum()
            rs_long = sector_long - bench_long

            # Combined score (weight short-term higher for momentum)
            rs_scores[sector] = 0.6 * self._normalize(rs_short) + \
                               0.4 * self._normalize(rs_long)

        return rs_scores

    def _normalize(self, series: pd.Series) -> pd.Series:
        """Normalize to 0-100 scale"""
        min_val = series.rolling(252).min()
        max_val = series.rolling(252).max()
        return ((series - min_val) / (max_val - min_val)) * 100

    def calculate_momentum_score(self,
                                  prices: pd.DataFrame) -> pd.DataFrame:
        """
        Dual momentum score: absolute + relative
        """
        scores = pd.DataFrame(index=prices.index)

        for sector in prices.columns:
            # Absolute momentum (is sector trending up?)
            sma_20 = prices[sector].rolling(20).mean()
            sma_50 = prices[sector].rolling(50).mean()
            abs_momentum = (prices[sector] > sma_20) & (sma_20 > sma_50)

            # Rate of change (how fast is it moving?)
            roc_20 = prices[sector].pct_change(20)
            roc_60 = prices[sector].pct_change(60)

            # Combined momentum score
            scores[sector] = (
                abs_momentum.astype(float) * 40 +
                self._normalize(roc_20) * 35 +
                self._normalize(roc_60) * 25
            )

        return scores

    def generate_signals(self,
                         rs_scores: pd.DataFrame,
                         momentum_scores: pd.DataFrame,
                         current_positions: Dict[str, float]) -> Dict[str, dict]:
        """
        Generate buy/sell signals based on combined scores
        """
        # Get latest scores
        latest_rs = rs_scores.iloc[-1]
        latest_momentum = momentum_scores.iloc[-1]

        # Combined score (equal weight RS and momentum)
        combined = (latest_rs + latest_momentum) / 2
        ranked = combined.sort_values(ascending=False)

        signals = {}

        # Top N sectors to go long
        long_sectors = ranked.head(self.top_n).index.tolist()

        # Bottom N sectors to potentially short (if regime allows)
        short_sectors = ranked.tail(self.bottom_n).index.tolist()

        for sector in ranked.index:
            current_pos = current_positions.get(sector, 0.0)

            if sector in long_sectors:
                target_weight = 1.0 / self.top_n  # Equal weight longs
                if abs(target_weight - current_pos) > self.rebalance_threshold:
                    signals[sector] = {
                        'action': 'BUY' if current_pos < target_weight else 'REDUCE',
                        'target_weight': target_weight,
                        'current_weight': current_pos,
                        'rs_score': latest_rs[sector],
                        'momentum_score': latest_momentum[sector],
                        'combined_score': combined[sector],
                        'rank': list(ranked.index).index(sector) + 1
                    }

            elif sector in short_sectors:
                target_weight = -0.1 / self.bottom_n  # Small short allocation
                if abs(target_weight - current_pos) > self.rebalance_threshold:
                    signals[sector] = {
                        'action': 'SHORT' if current_pos > target_weight else 'COVER',
                        'target_weight': target_weight,
                        'current_weight': current_pos,
                        'rs_score': latest_rs[sector],
                        'momentum_score': latest_momentum[sector],
                        'combined_score': combined[sector],
                        'rank': list(ranked.index).index(sector) + 1
                    }

            else:
                # Neutral sector - should be flat
                if abs(current_pos) > self.rebalance_threshold:
                    signals[sector] = {
                        'action': 'CLOSE',
                        'target_weight': 0.0,
                        'current_weight': current_pos,
                        'rs_score': latest_rs[sector],
                        'momentum_score': latest_momentum[sector],
                        'combined_score': combined[sector],
                        'rank': list(ranked.index).index(sector) + 1
                    }

        return signals

    def backtest(self,
                 prices: pd.DataFrame,
                 start_date: str,
                 end_date: str,
                 initial_capital: float = 100000) -> pd.DataFrame:
        """
        Simple backtest framework
        """
        prices = prices.loc[start_date:end_date]
        returns = prices.pct_change()
        benchmark = prices.mean(axis=1).pct_change()

        rs_scores = self.calculate_relative_strength(returns, benchmark)
        momentum_scores = self.calculate_momentum_score(prices)

        portfolio_value = [initial_capital]
        current_positions = {}

        for i in range(60, len(prices)):  # Start after warmup
            date = prices.index[i]

            # Generate signals
            signals = self.generate_signals(
                rs_scores.iloc[:i+1],
                momentum_scores.iloc[:i+1],
                current_positions
            )

            # Execute signals (simplified)
            for sector, signal in signals.items():
                current_positions[sector] = signal['target_weight']

            # Calculate daily P&L
            daily_return = 0
            for sector, weight in current_positions.items():
                if sector in returns.columns:
                    daily_return += weight * returns[sector].iloc[i]

            new_value = portfolio_value[-1] * (1 + daily_return)
            portfolio_value.append(new_value)

        return pd.DataFrame({
            'portfolio_value': portfolio_value[1:],
            'date': prices.index[60:]
        }).set_index('date')