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_date = prices[prices['Dates'] >= rebalance_date]['Dates'].min()
        if pd.isnull(nearest_date):
            continue

        signals_df = calculate_momentum_signals(prices, nearest_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

        prices_on_date = prices[prices['Dates'] == nearest_date].iloc[0][top_assets]
        quantities = equal_allocation / prices_on_date

        portfolio_history.append({
            'Date': nearest_date,
            'AUM': current_aum,
            'Top_Assets': list(top_assets),
            'Quantities': list(quantities)
        })

        next_date = prices[prices['Dates'] > nearest_date]['Dates'].min()
        if pd.isnull(next_date):
            break

        next_prices = prices[prices['Dates'] == next_date].iloc[0][top_assets]
        current_aum = sum(quantities * next_prices)

    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.
    """
    try:
        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}")
    except Exception as e:
        print(f"Error: {str(e)}")
        print("Please check your data and parameters, then try again.")