Untitled
unknown
plain_text
3 days ago
16 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
total_return = (price_current / price_start) - 1
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
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_aggregated_momentum(prices_df, current_date, ticker, lookback_periods):
"""
Computes aggregated momentum for an asset over multiple lookback periods.
Returns:
- aggregated_score: the average of all momentum signals (across all lookbacks)
- is_positive: True only if every individual signal is positive.
- signals: list of individual signals (for inspection)
"""
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.
signals.extend([tr, pma, 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):
"""
Backtests the long-only momentum strategy with updated selection rules.
The final portfolio always has exactly top_n (6) positions.
Selection logic:
- If ≥ 6 risk‑on assets have positive momentum: take top 6 positive.
- If exactly 5 positive: add 1 risk‑off.
- If exactly 4 positive: add 2 risk‑off.
- If < 4 positive: take all positive, add both risk‑off, then fill remaining slots with the top-ranked risk‑on assets regardless of sign.
Parameters:
prices_df: DataFrame with a "Date" column (datetime) and asset price columns.
start_date: Start date of the strategy (e.g. "2024-01-01")
end_date: End date of the strategy (e.g. "2025-01-01")
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
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.
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
risk_on_all = []
for ticker in risk_on_list:
agg_score, is_positive, _ = compute_aggregated_momentum(prices_df, reb_date, ticker, lookback_periods)
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
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:
final_selection = positive_risk_on + risk_off_list[:2]
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
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
def backtest_momentum_strategy_scores_approach(prices_df, start_date, end_date, rebalance_frequency, lookback_periods, aum, top_n, risk_on_list, risk_off_list):
"""
Backtests the momentum strategy using the scores approach.
At each rebalance date:
- For each risk‑on asset, compute all momentum signals (total_return, price_minus_sma, risk_adjusted)
over each lookback period.
- Rank the risk‑on assets for each signal (highest value gets the highest rank).
- Sum the rank scores across signals to obtain an aggregated score.
- Select the top_n risk‑on assets based on the aggregated score.
- For any selected asset that does not have all positive momentum signals,
substitute it with the risk‑off asset with the highest aggregated score.
- Allocate equal weight to each asset.
"""
prices_df['Date'] = pd.to_datetime(prices_df['Date'])
prices_df.sort_values('Date', inplace=True)
rebalance_dates = pd.date_range(start=start_date, end=end_date, freq=rebalance_frequency)
current_aum = aum
result_records = []
current_portfolio = {}
for i, reb_date in enumerate(rebalance_dates):
# Update AUM (mark-to-market) on rebalancing dates after the first.
if i > 0 and current_portfolio:
portfolio_value = 0
for ticker, (qty, _) in current_portfolio.items():
price_today = get_price_on_date(prices_df, reb_date, ticker)
portfolio_value += qty * price_today
current_aum = portfolio_value
# ---- Build risk-on momentum signals table ----
signal_data = {}
is_positive_flag = {}
for ticker in risk_on_list:
signals = {}
positive_checks = []
for lb in lookback_periods:
tr, pma, ra = compute_momentum_for_asset(prices_df, reb_date, ticker, lb)
if pd.isna(tr) or pd.isna(pma) or pd.isna(ra):
signals[f"tr_{lb}"] = np.nan
signals[f"pma_{lb}"] = np.nan
signals[f"ra_{lb}"] = np.nan
positive_checks.append(False)
else:
signals[f"tr_{lb}"] = tr
signals[f"pma_{lb}"] = pma
signals[f"ra_{lb}"] = ra
positive_checks.append(tr > 0 and pma > 0 and ra > 0)
signal_data[ticker] = signals
is_positive_flag[ticker] = all(positive_checks)
risk_on_df = pd.DataFrame.from_dict(signal_data, orient='index')
risk_on_df = risk_on_df.dropna() # Drop assets with incomplete data
if risk_on_df.empty:
final_selection = []
else:
# ---- Ranking: For each signal, rank assets in descending order.
ranking = risk_on_df.rank(method='min', ascending=False)
# Convert ranks to scores: best gets highest (score = number of assets - rank + 1)
ranking_scores = risk_on_df.shape[0] - ranking + 1
# Aggregate the score across all signals.
risk_on_df['agg_score'] = ranking_scores.sum(axis=1)
# Add the is_positive flag for each asset.
risk_on_df['is_positive'] = risk_on_df.index.map(lambda x: is_positive_flag.get(x, False))
# Sort assets by aggregated score (higher is better).
risk_on_df_sorted = risk_on_df.sort_values('agg_score', ascending=False)
# Select top_n risk-on assets.
final_selection = list(risk_on_df_sorted.index[:top_n])
# ---- Substitute any asset with negative momentum ----
# For any asset in the selection that does not have all positive signals,
# compute risk-off scores and replace it with the best risk-off candidate not already selected.
signal_data_off = {}
is_positive_off = {}
for ticker in risk_off_list:
signals = {}
positive_checks = []
for lb in lookback_periods:
tr, pma, ra = compute_momentum_for_asset(prices_df, reb_date, ticker, lb)
if pd.isna(tr) or pd.isna(pma) or pd.isna(ra):
signals[f"tr_{lb}"] = np.nan
signals[f"pma_{lb}"] = np.nan
signals[f"ra_{lb}"] = np.nan
positive_checks.append(False)
else:
signals[f"tr_{lb}"] = tr
signals[f"pma_{lb}"] = pma
signals[f"ra_{lb}"] = ra
positive_checks.append(tr > 0 and pma > 0 and ra > 0)
signal_data_off[ticker] = signals
is_positive_off[ticker] = all(positive_checks)
risk_off_df = pd.DataFrame.from_dict(signal_data_off, orient='index')
risk_off_df = risk_off_df.dropna()
if not risk_off_df.empty:
ranking_off = risk_off_df.rank(method='min', ascending=False)
ranking_scores_off = risk_off_df.shape[0] - ranking_off + 1
risk_off_df['agg_score'] = ranking_scores_off.sum(axis=1)
risk_off_df['is_positive'] = risk_off_df.index.map(lambda x: is_positive_off.get(x, False))
risk_off_df_sorted = risk_off_df.sort_values('agg_score', ascending=False)
# Replace each non-positive risk-on asset with the best risk-off candidate not already selected.
final_selection_new = []
risk_off_candidates = list(risk_off_df_sorted.index)
for asset in final_selection:
if risk_on_df.loc[asset, 'is_positive']:
final_selection_new.append(asset)
else:
substitute = None
for candidate in risk_off_candidates:
if candidate not in final_selection_new and candidate not in final_selection:
substitute = candidate
break
final_selection_new.append(substitute if substitute else asset)
final_selection = final_selection_new
# In case fewer than top_n are available, fill remaining slots from risk-on (by aggregated score).
if len(final_selection) < top_n:
additional = list(risk_on_df_sorted.index.difference(final_selection))
final_selection += additional[:(top_n - len(final_selection))]
final_selection = final_selection[:top_n]
# ---- Portfolio Allocation: Equal weighting ----
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 that 'prices' is a DataFrame with a "Date" column and asset price columns.
# For example, the columns might be: ['Date', 'Asset1', 'Asset2', ... 'Asset15']
#
# Define risk_on_list and risk_off_list based on the columns in your 'prices' DataFrame.
# Here we assume columns 1 to 13 (index positions 1 to 13) are risk-on assets,
# and columns 14 to 15 are risk-off assets.
risk_on_list = list(prices.columns[1:14])
risk_off_list = list(prices.columns[14:16])
# Using the original backtest function:
result_df_original = backtest_momentum_strategy(
prices_df=prices,
start_date="2024-01-01",
end_date="2025-01-01",
rebalance_frequency="MS",
lookback_periods=[3, 6, 9],
aum=1000000,
top_n=6,
risk_on_list=risk_on_list,
risk_off_list=risk_off_list
)
# Using the scores approach backtest function:
result_df_scores = backtest_momentum_strategy_scores_approach(
prices_df=prices,
start_date="2024-01-01",
end_date="2025-01-01",
rebalance_frequency="MS",
lookback_periods=[3, 6, 9],
aum=1000000,
top_n=6,
risk_on_list=risk_on_list,
risk_off_list=risk_off_list
)
# Output the results for inspection.
print("Original Backtest Strategy Results:")
print(result_df_original)
print("\nScores Approach Backtest Strategy Results:")
print(result_df_scores)
Editor is loading...
Leave a Comment