first week 2026 - january momentum algo kicking off

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

class JanuarySignal(Enum):
    STRONG_BUY = "strong_momentum"
    BUY = "momentum"
    NEUTRAL = "flat"
    AVOID = "reversal_risk"

@dataclass
class JanuaryMomentumConfig:
    lookback_days: int = 10  # last 10 trading days of december
    min_momentum_score: float = 0.6
    position_size_pct: float = 0.03  # 3% per position
    max_positions: int = 15
    stop_loss_pct: float = 0.04  # 4% stop
    take_profit_pct: float = 0.08  # 8% target

class JanuaryMomentumAlgo:
    def __init__(self, config: JanuaryMomentumConfig):
        self.config = config
        self.active_positions = {}
        self.signals_generated = []

    def calculate_december_momentum(self,
                                     prices: pd.DataFrame,
                                     volume: pd.DataFrame) -> pd.DataFrame:
        """
        Calculate momentum score based on december performance
        Higher score = stronger january continuation expected
        """
        results = []

        for symbol in prices.columns:
            try:
                # Get last N days of december
                dec_prices = prices[symbol].iloc[-self.config.lookback_days:]
                dec_volume = volume[symbol].iloc[-self.config.lookback_days:]

                # Price momentum (weighted recent more)
                weights = np.linspace(0.5, 1.5, len(dec_prices))
                price_return = (dec_prices.iloc[-1] / dec_prices.iloc[0]) - 1
                weighted_return = price_return * np.average(weights)

                # Volume confirmation
                avg_vol = dec_volume.mean()
                recent_vol = dec_volume.iloc[-3:].mean()
                vol_ratio = recent_vol / avg_vol if avg_vol > 0 else 1.0

                # Momentum score (0-1 scale)
                momentum_score = self._normalize_score(
                    weighted_return * 0.6 + (vol_ratio - 1) * 0.4
                )

                results.append({
                    'symbol': symbol,
                    'dec_return': price_return,
                    'vol_ratio': vol_ratio,
                    'momentum_score': momentum_score,
                    'signal': self._generate_signal(momentum_score)
                })

            except Exception as e:
                continue

        return pd.DataFrame(results).sort_values(
            'momentum_score', ascending=False
        )

    def _normalize_score(self, raw_score: float) -> float:
        """Normalize to 0-1 scale using sigmoid"""
        return 1 / (1 + np.exp(-raw_score * 10))

    def _generate_signal(self, score: float) -> JanuarySignal:
        if score >= 0.75:
            return JanuarySignal.STRONG_BUY
        elif score >= self.config.min_momentum_score:
            return JanuarySignal.BUY
        elif score >= 0.4:
            return JanuarySignal.NEUTRAL
        else:
            return JanuarySignal.AVOID

    def generate_january_portfolio(self,
                                    momentum_df: pd.DataFrame,
                                    account_value: float) -> List[dict]:
        """
        Build portfolio from top momentum stocks
        """
        # Filter to actionable signals
        buys = momentum_df[
            momentum_df['signal'].isin([
                JanuarySignal.STRONG_BUY,
                JanuarySignal.BUY
            ])
        ].head(self.config.max_positions)

        positions = []
        position_value = account_value * self.config.position_size_pct

        for _, row in buys.iterrows():
            positions.append({
                'symbol': row['symbol'],
                'signal': row['signal'].value,
                'momentum_score': row['momentum_score'],
                'allocation_usd': position_value,
                'stop_loss': self.config.stop_loss_pct,
                'take_profit': self.config.take_profit_pct,
                'entry_date': datetime.now().strftime('%Y-%m-%d')
            })

        return positions

    def backtest_january(self,
                          prices: pd.DataFrame,
                          years: List[int]) -> dict:
        """
        Backtest january strategy across multiple years
        Returns performance metrics
        """
        results = {
            'year': [],
            'jan_return': [],
            'spy_return': [],
            'alpha': [],
            'win_rate': [],
            'max_dd': []
        }

        for year in years:
            # Would implement full backtest here
            # Simplified for blog post
            pass

        return results