Untitled

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def calculate_momentum_signals(prices, start_date):
    """
    Calculates 9 momentum signals for each ticker at the given start date.
    """
    lookbacks = {"1M": 21, "3M": 63, "6M": 126}  # Approximate trading days
    tickers = prices.columns[1:]  # Exclude the 'Dates' column
    
    start_date = pd.to_datetime(start_date)
    
    # Ensure prices are sorted
    prices = prices.sort_values('Dates')
    
    # Ensure start_date is within the available range
    if start_date < prices['Dates'].min() or start_date > prices['Dates'].max():
        raise ValueError(f"Start date {start_date.strftime('%m/%d/%Y')} is outside the date range of the price data. "
                         f"Available date range: {prices['Dates'].min().strftime('%m/%d/%Y')} to {prices['Dates'].max().strftime('%m/%d/%Y')}")
    
    # Find nearest available date on or after the start_date
    start_date = prices.loc[prices['Dates'] >= start_date, 'Dates'].min()
    
    # Set index to Date
    prices = prices.set_index('Dates')
    
    signals_list = []
    for date in prices.loc[start_date:].index:
        if prices.index.get_loc(date) < max(lookbacks.values()):
            continue
        
        pt = prices.loc[date, tickers]
        signals = {"Dates": date}
        
        for label, lb in lookbacks.items():
            pt_n = prices.iloc[prices.index.get_loc(date) - lb][tickers]
            
            # Total Return Momentum
            total_return = (pt - pt_n) / pt_n
            
            # Price Minus Moving Average
            sma = prices[tickers].iloc[prices.index.get_loc(date)-lb : prices.index.get_loc(date)].mean()
            price_sma = (pt / sma) - 1
            
            # Risk-Adjusted Momentum
            log_returns = np.log(prices[tickers] / prices[tickers].shift(1))
            log_returns_sum = log_returns.iloc[prices.index.get_loc(date)-lb : prices.index.get_loc(date)].abs().sum()
            log_momentum = np.log(pt / pt_n) / np.where(log_returns_sum == 0, np.nan, log_returns_sum)
            
            # Store results
            for ticker in tickers:
                signals[f'TotalReturn_{label}_{ticker}'] = total_return[ticker]
                signals[f'PriceMinusSMA_{label}_{ticker}'] = price_sma[ticker]
                signals[f'RiskAdjusted_{label}_{ticker}'] = log_momentum[ticker]
        
        signals_list.append(signals)
    
    signals_df = pd.DataFrame(signals_list)
    return signals_df.set_index("Dates")

def rebalance_portfolio(prices, start_date, end_date, rebalance_frequency, initial_aum=100_000_000):
    """
    Implements the momentum-based strategy with periodic rebalancing.
    """
    start_date = pd.to_datetime(start_date)
    end_date = pd.to_datetime(end_date)
    
    rebalance_dates = pd.date_range(start=start_date, end=end_date, freq=f'{rebalance_frequency}M')
    
    portfolio_history = []
    current_aum = initial_aum
    
    for rebalance_date in rebalance_dates:
        rebalance_date = pd.to_datetime(rebalance_date)
        
        # Find nearest trading day on or after rebalance_date
        nearest_dates = prices.index[prices.index >= rebalance_date]
        if nearest_dates.empty:
            continue
        rebalance_date = nearest_dates.min()
        
        signals_df = calculate_momentum_signals(prices, rebalance_date.strftime('%m/%d/%Y'))
        ranked_df = rank_momentum_signals(signals_df)
        
        top_assets = ranked_df.iloc[-1].filter(like='Rank_').nsmallest(6).index.str.split('_').str[-1]
        equal_allocation = current_aum / 6
        
        try:
            prices_on_date = prices.loc[rebalance_date, top_assets]
            quantities = equal_allocation / prices_on_date
        except KeyError:
            continue  # Skip this rebalancing if any asset price is missing
        
        portfolio_history.append({
            'Date': rebalance_date,
            'AUM': current_aum,
            'Top_Assets': list(top_assets),
            'Quantities': list(quantities)
        })
        
        next_date = prices.index[prices.index > rebalance_date].min()
        if pd.isnull(next_date):
            break
        
        current_aum = sum(quantities * prices.loc[next_date, top_assets])
    
    return pd.DataFrame(portfolio_history)

def export_to_excel(prices, start_date, end_date, rebalance_frequency, output_file="momentum_strategy.xlsx"):
    """
    Runs the backtest and exports the results to an Excel file.
    """
    portfolio_df = rebalance_portfolio(prices, start_date, end_date, rebalance_frequency)
    portfolio_df.to_excel(output_file, index=False)
    print(f"Momentum strategy results saved to {output_file}")