Untitled

 avatar
unknown
plain_text
14 days ago
3.9 kB
3
Indexable
import pandas as pd
import numpy as np

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: 'YYYY-MM-DD').
    
    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:]
    
    # Ensure start_date is in DataFrame
    if start_date not in prices['Date'].values:
        raise ValueError("Start date not found in price data.")
    
    # Set index to Date
    prices = prices.set_index('Date')
    
    # Ensure prices are sorted
    prices = prices.sort_index()
    
    # Initialize dictionary 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 = {"Date": 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_momentum = np.log(pt / pt_n) / log_returns.iloc[prices.index.get_loc(date)-lb : prices.index.get_loc(date)].abs().sum()
            
            # Store results
            signals[f'TotalReturn_{label}'] = total_return
            signals[f'PriceMinusSMA_{label}'] = price_sma
            signals[f'RiskAdjusted_{label}'] = log_momentum
        
        signals_list.append(signals)
    
    signals_df = pd.DataFrame(signals_list)
    return signals_df.set_index("Date")

def rank_momentum_signals(signals_df):
    """
    Combines the 9 signals into a composite momentum score and ranks the 13 assets.
    
    Parameters:
    signals_df (DataFrame): DataFrame containing momentum signals for each ticker.
    
    Returns:
    DataFrame with ranked momentum scores.
    """
    weights = {"1M": 0.15, "3M": 0.35, "6M": 0.50}
    
    # Normalize each signal
    signals_norm = (signals_df - signals_df.mean()) / signals_df.std()
    
    # Compute weighted sum for each asset
    composite_score = sum(weights[label] / 3 * signals_norm.filter(like=label) for label in weights.keys())
    
    # Rank assets (1 = highest momentum, 13 = lowest momentum)
    rankings = composite_score.rank(ascending=False, method='dense').astype(int)
    
    # Combine scores and rankings
    result_df = signals_df.copy()
    result_df['Composite_Score'] = composite_score
    result_df['Rank'] = rankings
    
    return result_df

def export_to_excel(prices, start_date, output_file="momentum_strategy.xlsx"):
    """
    Calculates momentum signals, ranks tickers, and exports to Excel.
    
    Parameters:
    prices (DataFrame): DataFrame with price data.
    start_date (str): Start date for analysis.
    output_file (str): Name of the output Excel file.
    """
    signals_df = calculate_momentum_signals(prices, start_date)
    ranked_df = rank_momentum_signals(signals_df)
    
    # Add price data to output
    prices_subset = prices.loc[start_date:].copy()
    ranked_df = ranked_df.join(prices_subset, how='left')
    
    # Export to Excel
    ranked_df.to_excel(output_file)
    print(f"Momentum strategy results saved to {output_file}")
Editor is loading...
Leave a Comment