Untitled
unknown
plain_text
7 months ago
3.9 kB
4
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