Untitled
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