Untitled
unknown
plain_text
2 days ago
11 kB
2
Indexable
import pandas as pd
import numpy as np
from dateutil.relativedelta import relativedelta
import math
def get_price_on_date(prices_df, target_date, ticker):
"""
Returns the last available price on or before target_date for a given ticker.
"""
df_filtered = prices_df[prices_df['Date'] <= target_date]
if df_filtered.empty:
return np.nan
return df_filtered.iloc[-1][ticker]
def compute_momentum_for_asset(prices_df, current_date, ticker, lookback_months):
"""
Computes three momentum signals for the given asset and lookback period.
Returns:
- total_return (price-based momentum)
- price_minus_sma (distance from moving average)
- risk_adjusted (risk-adjusted momentum)
"""
start_date = current_date - relativedelta(months=lookback_months)
price_start = get_price_on_date(prices_df, start_date, ticker)
price_current = get_price_on_date(prices_df, current_date, ticker)
if pd.isna(price_start) or pd.isna(price_current):
return np.nan, np.nan, np.nan
# 1. Total return momentum
total_return = (price_current / price_start) - 1
# 2. Price minus moving average momentum
prices_window = prices_df[(prices_df['Date'] >= start_date) & (prices_df['Date'] <= current_date)][ticker]
sma = prices_window.mean() if not prices_window.empty else np.nan
price_minus_sma = (price_current / sma) - 1 if (not pd.isna(sma) and sma != 0) else np.nan
# 3. Risk-adjusted momentum
numerator = np.log(price_current / price_start)
prices_period = prices_df[(prices_df['Date'] >= start_date) & (prices_df['Date'] <= current_date)][ticker].sort_index()
if len(prices_period) < 2:
risk_adjusted = np.nan
else:
log_returns = np.log(prices_period / prices_period.shift(1)).dropna()
denominator = log_returns.abs().sum()
risk_adjusted = numerator / denominator if denominator != 0 else np.nan
return total_return, price_minus_sma, risk_adjusted
def compute_asset_correlation(prices_df, current_date, ticker, universe, lookback_months=12):
"""
Computes the correlation between asset 'ticker' and an equally weighted portfolio
of all assets in 'universe' over the past 'lookback_months' months.
Returns the Pearson correlation coefficient.
"""
start_date = current_date - relativedelta(months=lookback_months)
df_window = prices_df[(prices_df['Date'] >= start_date) & (prices_df['Date'] <= current_date)].sort_values('Date')
if df_window.empty:
return np.nan
# Compute daily log returns for the asset
asset_prices = df_window[ticker]
asset_returns = np.log(asset_prices / asset_prices.shift(1)).dropna()
# Compute daily log returns for each asset in the universe
portfolio_returns = pd.DataFrame()
for t in universe:
t_prices = df_window[t]
t_returns = np.log(t_prices / t_prices.shift(1))
portfolio_returns[t] = t_returns
# Equal-weighted portfolio returns
portfolio_returns['EW'] = portfolio_returns.mean(axis=1)
# Align the series
combined = pd.concat([asset_returns, portfolio_returns['EW']], axis=1, join='inner')
combined.columns = ['asset', 'portfolio']
if len(combined) < 2:
return np.nan
corr = combined['asset'].corr(combined['portfolio'])
return corr
def compute_aggregated_momentum(prices_df, current_date, ticker, lookback_periods, correlation_lookback=12, universe=None):
"""
For a given ticker and current_date, compute momentum signals over each lookback period.
Each raw momentum signal is adjusted for correlation using:
φ = r / (1 + ρ)
where ρ is the asset’s correlation with an equally weighted portfolio of the risk-on universe
(computed over 'correlation_lookback' months).
Returns:
- aggregated_score: average of all adjusted momentum signals
- is_positive: True only if every adjusted signal is positive.
- signals: list of all adjusted signals (for inspection)
"""
if universe is None:
# If no universe provided, use the ticker itself (not recommended)
universe = [ticker]
# Compute correlation (ρ) using a fixed lookback period (e.g., 12 months)
rho = compute_asset_correlation(prices_df, current_date, ticker, universe, lookback_months=correlation_lookback)
if pd.isna(rho):
return None, None, None
signals = []
for lb in lookback_periods:
tr, pma, ra = compute_momentum_for_asset(prices_df, current_date, ticker, lb)
if pd.isna(tr) or pd.isna(pma) or pd.isna(ra):
return None, None, None # Incomplete data; skip asset.
# Apply correlation adjustment to each signal:
adjusted_tr = tr / (1 + rho)
adjusted_pma = pma / (1 + rho)
adjusted_ra = ra / (1 + rho)
signals.extend([adjusted_tr, adjusted_pma, adjusted_ra])
agg_score = np.mean(signals)
is_positive = all(x > 0 for x in signals)
return agg_score, is_positive, signals
def backtest_momentum_strategy(prices_df, start_date, end_date, rebalance_frequency, lookback_periods, aum, top_n, risk_on_list, risk_off_list, correlation_lookback=12):
"""
Backtests the long-only momentum strategy using the "scores approach" with correlation adjustment.
The final portfolio always has exactly top_n (e.g. 6) positions.
Selection logic (ensuring exactly 6 positions):
- If ≥ 6 risk‑on assets have positive momentum: take top 6 positive.
- If exactly 5 positive: add 1 risk‑off asset.
- If exactly 4 positive: add 2 risk‑off assets.
- If < 4 positive: take available positive risk‑on, add both risk‑off assets,
then fill remaining slots with the top-ranked risk‑on assets (regardless of sign) until 6 positions.
Parameters:
prices_df: DataFrame with a "Date" column (datetime) and asset price columns.
start_date: Strategy start date (string, e.g. "2024-01-01")
end_date: Strategy end date (string)
rebalance_frequency: Frequency string for rebalancing (e.g. "MS" for month start)
lookback_periods: List of lookback periods in months (e.g. [3, 6, 9])
aum: Starting assets under management (e.g. 1,000,000)
top_n: Total number of positions to hold (e.g. 6)
risk_on_list: List of risk-on asset tickers.
risk_off_list: List of risk-off asset tickers.
correlation_lookback: Lookback period (in months) for computing correlation (default 12)
Returns:
result_df: DataFrame with each rebalance date, portfolio AUM, and details of positions.
"""
prices_df['Date'] = pd.to_datetime(prices_df['Date'])
prices_df.sort_values('Date', inplace=True)
# Build rebalancing dates
rebalance_dates = pd.date_range(start=start_date, end=end_date, freq=rebalance_frequency)
current_aum = aum
result_records = []
current_portfolio = {} # {ticker: (quantity, entry_price)}
for i, reb_date in enumerate(rebalance_dates):
# Update portfolio AUM based on current prices (mark-to-market)
if i > 0 and current_portfolio:
portfolio_value = 0
for ticker, (qty, entry_price) in current_portfolio.items():
price_today = get_price_on_date(prices_df, reb_date, ticker)
portfolio_value += qty * price_today
current_aum = portfolio_value
# Compute momentum scores for all risk-on assets using the correlation-adjusted scores approach
risk_on_all = []
for ticker in risk_on_list:
agg_score, is_positive, _ = compute_aggregated_momentum(prices_df, reb_date, ticker, lookback_periods, correlation_lookback, universe=risk_on_list)
if agg_score is not None:
risk_on_all.append((ticker, agg_score, is_positive))
risk_on_all = sorted(risk_on_all, key=lambda x: x[1], reverse=True)
# Separate those with strictly positive momentum (i.e. all adjusted signals > 0)
positive_risk_on = [ticker for ticker, score, is_positive in risk_on_all if is_positive]
# Build final selection to always have exactly top_n (6) assets:
if len(positive_risk_on) >= 6:
final_selection = positive_risk_on[:6]
elif len(positive_risk_on) == 5:
final_selection = positive_risk_on + risk_off_list[:1]
elif len(positive_risk_on) == 4:
final_selection = positive_risk_on + risk_off_list[:2]
else:
# Fewer than 4 positive risk-on: start with available positive ones plus both risk-off
final_selection = positive_risk_on + risk_off_list[:2]
# Fill remaining slots with top-ranked risk-on assets regardless of sign
for ticker, score, is_positive in risk_on_all:
if ticker not in final_selection:
final_selection.append(ticker)
if len(final_selection) == 6:
break
# Ensure final_selection always has exactly top_n assets
final_selection = final_selection[:6]
# Allocate equal weight among the selected assets
allocation = current_aum / len(final_selection) if final_selection else 0
positions = {}
for ticker in final_selection:
price_at_entry = get_price_on_date(prices_df, reb_date, ticker)
qty = allocation / price_at_entry if price_at_entry != 0 else 0
positions[ticker] = (qty, price_at_entry)
record = {
'Rebalance Date': reb_date,
'Final AUM': current_aum,
'Selected Assets': final_selection,
'Quantities': [positions[ticker][0] for ticker in final_selection],
'Entry Prices': [positions[ticker][1] for ticker in final_selection]
}
result_records.append(record)
current_portfolio = positions.copy()
result_df = pd.DataFrame(result_records)
return result_df
# --- Example usage ---
# Assume 'prices' is your DataFrame with a "Date" column and 15 asset price columns.
# risk_on_list: first 13 tickers (risk-on)
# risk_off_list: last 2 tickers (risk-off)
risk_on_list = list(prices.columns[1:14])
risk_off_list = list(prices.columns[14:16])
result_df = backtest_momentum_strategy(
prices_df=prices,
start_date="2024-01-01",
end_date="2025-01-01",
rebalance_frequency="2MS", # every two months for example
lookback_periods=[3, 6, 9],
aum=1000000,
top_n=6,
risk_on_list=risk_on_list,
risk_off_list=risk_off_list,
correlation_lookback=12
)
print(result_df)
Editor is loading...
Leave a Comment