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.
    
    Parameters:
    prices (DataFrame): DataFrame with dates in the first column and 13 tickers ahead.
    start_date (str): Date for which signals should be calculated (format: 'MM/DD/YYYY').
    
    Returns:
    DataFrame with 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 = datetime.strptime(start_date, '%m/%d/%Y')

    # Ensure prices are sorted
    prices = prices.sort_values('Dates')

    # Check if start_date is within the date range of the data
    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 the nearest available date on or after the start_date
    start_date = prices[prices['Dates'] >= start_date]['Dates'].min()

    # Set index to Date
    prices = prices.set_index('Dates')

    # Initialize list to store signal values
    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 rank_momentum_signals(signals_df):
    """
    Combines the 9 signals into a composite momentum score and ranks the 13 assets.
    """
    weights = {"1M": 0.15, "3M": 0.35, "6M": 0.50}
    tickers = set([col.split('_')[-1] for col in signals_df.columns])

    # Normalize each signal
    signals_norm = signals_df.apply(lambda x: (x - x.mean()) / x.std())

    # Compute weighted sum for each asset
    composite_scores = pd.DataFrame(index=signals_df.index, columns=tickers)
    for ticker in tickers:
        ticker_signals = signals_norm.filter(regex=f'.*_{ticker}$')
        composite_scores[ticker] = sum(weights[label] / 3 * ticker_signals.filter(like=label).mean(axis=1) for label in weights.keys())

    # Rank assets (1 = highest momentum, 13 = lowest momentum)
    rankings = composite_scores.rank(axis=1, ascending=False, method='dense')

    # Combine scores and rankings
    result_df = signals_df.copy()
    for ticker in tickers:
        result_df[f'Composite_Score_{ticker}'] = composite_scores[ticker]
        result_df[f'Rank_{ticker}'] = rankings[ticker]

    return result_df

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 = datetime.strptime(start_date, '%m/%d/%Y')
    end_date = datetime.strptime(end_date, '%m/%d/%Y')
    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 = prices[prices.index >= rebalance_date].index.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
        prices_on_date = prices.loc[rebalance_date, top_assets]
        quantities = equal_allocation / prices_on_date

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

        current_aum = sum(quantities * prices.loc[rebalance_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}")