Untitled

import pandas as pd
import numpy as np
from dateutil.relativedelta import relativedelta
import math

def get_price_on_date(prices_df, target_date, ticker):
    """
    Returns the last available price on or before target_date for a given ticker.
    """
    df_filtered = prices_df[prices_df['Date'] <= target_date]
    if df_filtered.empty:
        return np.nan
    return df_filtered.iloc[-1][ticker]

def compute_momentum_for_asset(prices_df, current_date, ticker, lookback_months):
    """
    Computes three momentum signals for the given asset and lookback period.
    Returns:
      - total_return (price-based momentum)
      - price_minus_sma (distance from moving average)
      - risk_adjusted (risk-adjusted momentum)
    """
    start_date = current_date - relativedelta(months=lookback_months)
    price_start = get_price_on_date(prices_df, start_date, ticker)
    price_current = get_price_on_date(prices_df, current_date, ticker)
    
    if pd.isna(price_start) or pd.isna(price_current):
        return np.nan, np.nan, np.nan
    
    total_return = (price_current / price_start) - 1
    
    prices_window = prices_df[(prices_df['Date'] >= start_date) & (prices_df['Date'] <= current_date)][ticker]
    sma = prices_window.mean() if not prices_window.empty else np.nan
    price_minus_sma = (price_current / sma) - 1 if (not pd.isna(sma) and sma != 0) else np.nan
    
    numerator = np.log(price_current / price_start)
    prices_period = prices_df[(prices_df['Date'] >= start_date) & (prices_df['Date'] <= current_date)][ticker].sort_index()
    if len(prices_period) < 2:
        risk_adjusted = np.nan
    else:
        log_returns = np.log(prices_period / prices_period.shift(1)).dropna()
        denominator = log_returns.abs().sum()
        risk_adjusted = numerator / denominator if denominator != 0 else np.nan
    
    return total_return, price_minus_sma, risk_adjusted

def compute_aggregated_momentum(prices_df, current_date, ticker, lookback_periods):
    """
    Computes aggregated momentum for an asset over multiple lookback periods.
    Returns:
      - aggregated_score: the average of all momentum signals (across all lookbacks)
      - is_positive: True only if every individual signal is positive.
      - signals: list of individual signals (for inspection)
    """
    signals = []
    for lb in lookback_periods:
        tr, pma, ra = compute_momentum_for_asset(prices_df, current_date, ticker, lb)
        if pd.isna(tr) or pd.isna(pma) or pd.isna(ra):
            return None, None, None  # Incomplete data; skip asset.
        signals.extend([tr, pma, ra])
    agg_score = np.mean(signals)
    is_positive = all(x > 0 for x in signals)
    return agg_score, is_positive, signals

def backtest_momentum_strategy(prices_df, start_date, end_date, rebalance_frequency, lookback_periods, aum, top_n, risk_on_list, risk_off_list):
    """
    Backtests the long-only momentum strategy with updated selection rules.
    
    The final portfolio always has exactly top_n (6) positions.
    Selection logic:
      - If ≥ 6 risk‑on assets have positive momentum: take top 6 positive.
      - If exactly 5 positive: add 1 risk‑off.
      - If exactly 4 positive: add 2 risk‑off.
      - If < 4 positive: take all positive, add both risk‑off, then fill remaining slots with the top-ranked risk‑on assets regardless of sign.
    
    Parameters:
      prices_df: DataFrame with a "Date" column (datetime) and asset price columns.
      start_date: Start date of the strategy (string, e.g. "2024-01-01")
      end_date: End date of the strategy (string)
      rebalance_frequency: Frequency string for rebalancing (e.g. "MS" for month start)
      lookback_periods: List of lookback periods in months (e.g. [3,6,12])
      aum: Starting assets under management (e.g. 100e6)
      top_n: Total number of positions to hold (e.g. 6)
      risk_on_list: List of risk-on asset tickers.
      risk_off_list: List of risk-off asset tickers.
      
    Returns:
      result_df: DataFrame with each rebalance date, portfolio AUM, and details of positions.
    """
    prices_df['Date'] = pd.to_datetime(prices_df['Date'])
    prices_df.sort_values('Date', inplace=True)
    
    # Build rebalancing dates
    rebalance_dates = pd.date_range(start=start_date, end=end_date, freq=rebalance_frequency)
    current_aum = aum
    result_records = []
    current_portfolio = {}  # {ticker: (quantity, entry_price)}
    
    for i, reb_date in enumerate(rebalance_dates):
        # Update portfolio AUM based on current prices (mark-to-market)
        if i > 0 and current_portfolio:
            portfolio_value = 0
            for ticker, (qty, entry_price) in current_portfolio.items():
                price_today = get_price_on_date(prices_df, reb_date, ticker)
                portfolio_value += qty * price_today
            current_aum = portfolio_value
        
        # Compute momentum scores for all risk-on assets
        risk_on_all = []
        for ticker in risk_on_list:
            agg_score, is_positive, _ = compute_aggregated_momentum(prices_df, reb_date, ticker, lookback_periods)
            if agg_score is not None:
                risk_on_all.append((ticker, agg_score, is_positive))
        risk_on_all = sorted(risk_on_all, key=lambda x: x[1], reverse=True)
        
        # Separate those with strictly positive momentum
        positive_risk_on = [ticker for ticker, score, is_positive in risk_on_all if is_positive]
        
        # Build final selection to always have exactly top_n (6) assets
        if len(positive_risk_on) >= 5:
            # Enough positive risk-on assets: use top 6
            final_selection = positive_risk_on[:5]
        elif len(positive_risk_on) == 4:
            # 5 positive: add one risk-off asset
            final_selection = positive_risk_on + risk_off_list[:1]
        elif len(positive_risk_on) == 3:
            # 4 positive: add both risk-off assets
            final_selection = positive_risk_on + risk_off_list[:2]
        else:
            # Fewer than 4 positive risk-on:
            # Start with available positive ones plus both risk-off
            final_selection = positive_risk_on + risk_off_list[:2]
            # If still less than 6, fill from risk_on_all (regardless of momentum sign)
            for ticker, score, is_positive in risk_on_all:
                if ticker not in final_selection:
                    final_selection.append(ticker)
                if len(final_selection) == 5:
                    break
        
        # Ensure final_selection always has exactly top_n assets
        final_selection = final_selection[:6]
        
        # Allocate equal weight among the selected assets
        allocation = current_aum / len(final_selection) if final_selection else 0
        positions = {}
        for ticker in final_selection:
            price_at_entry = get_price_on_date(prices_df, reb_date, ticker)
            qty = allocation / price_at_entry if price_at_entry != 0 else 0
            positions[ticker] = (qty, price_at_entry)
        
        record = {
            'Rebalance Date': reb_date,
            'Final AUM': current_aum,
            'Selected Assets': final_selection,
            'Quantities': [positions[ticker][0] for ticker in final_selection],
            'Entry Prices': [positions[ticker][1] for ticker in final_selection]
        }
        result_records.append(record)
        current_portfolio = positions.copy()
    
    result_df = pd.DataFrame(result_records)
    return result_df

risk_on_list = list(prices.columns[1:14])
risk_off_list = list(prices.columns[14:16])


result_df = backtest_momentum_strategy(prices_df=prices, 
                                        start_date="2024-01-01", 
                                        end_date="2025-01-01",
                                        rebalance_frequency="2MS", 
                                        lookback_periods=[3, 6, 9],
                                        aum=1000000, 
                                        top_n=6,
                                        risk_on_list=risk_on_list,
                                        risk_off_list=risk_off_list)