Untitled
unknown
plain_text
a month ago
31 kB
5
Indexable
import pandas as pd
import numpy as np
from datetime import datetime
from dateutil.relativedelta import relativedelta
from scipy.stats import zscore
###############################################
# 1. Data Loading (Filtering Out Weekends)
###############################################
def load_price_data(filepath):
"""
Load historical prices from an Excel file.
Assumes that the first column is dates and the remaining columns are tickers.
Removes weekend data.
"""
df = pd.read_excel(filepath, index_col=0)
df.index = pd.to_datetime(df.index)
df = df.sort_index()
# Filter out weekends (Saturday=5, Sunday=6)
df = df[df.index.dayofweek < 5]
return df
def load_macro_data(filepath):
"""
Load macro indicators from an Excel file.
Loads VIX and VIX3M from 'Eq' sheet, LF98TRUU Index from 'FI' sheet,
and other macro indicators from 'Macro' sheet.
Removes weekend data.
Also computes slopes for selected macro indicators.
"""
# VIX data
vix_data = pd.read_excel(filepath, sheet_name='Eq', index_col=0, parse_dates=True, usecols=[0, 3, 4, 5])
vix_data.columns = ['VIX9D', 'VIX', 'VIX3M']
vix_data = vix_data[vix_data.index.dayofweek < 5]
# FI data
cdx_data = pd.read_excel(filepath, sheet_name='FI', index_col=0, parse_dates=True, usecols=[0, 2], skiprows=1)
cdx_data.columns = ['LF98TRUU']
cdx_data = cdx_data[cdx_data.index.dayofweek < 5]
# Macro data (assumed to include columns "CESIUSD Index", "INJCJC Index", ".HG/GC G Index", and "Consumer Confidence")
macro_data = pd.read_excel(filepath, sheet_name='Macro', index_col=0, parse_dates=True, usecols=range(8), skiprows=1)
macro_data = macro_data[macro_data.index.dayofweek < 5]
# Compute slopes for selected macro indicators.
macro_data["Surprise Index Slope"] = macro_data["CESIUSD Index"].diff()
macro_data["Jobless Claims Slope"] = macro_data["INJCJC Index"].diff()
macro_data["Copper Gold Slope"] = macro_data['.HG/GC G Index'].diff()
# Assume "Consumer Confidence" column already exists.
combined_data = pd.concat([vix_data, cdx_data, macro_data], axis=1)
combined_data = combined_data.fillna(method='ffill').fillna(method='bfill')
combined_data = combined_data.sort_index() # ensure sorted index
return combined_data
###############################################
# 2. Helper: Observation Dates (Monthly)
###############################################
def get_observation_dates(prices, start_date, end_date, rebalance_period):
"""
Returns observation dates using the trading calendar.
For a monthly rebalancing (rebalance_period=1), for each target month,
it starts at the first calendar day and checks sequentially until it finds a trading day.
Parameters:
prices: DataFrame with a DatetimeIndex containing trading days.
start_date: Starting date (should be a valid trading day).
end_date: End date for observations.
rebalance_period: Number of months between rebalances.
Returns:
List of observation dates.
"""
dates = []
current_date = start_date
while current_date < end_date:
# Move forward by the rebalance period (in months) and set candidate to the first day of that month
candidate_date = (current_date + relativedelta(months=rebalance_period)).replace(day=1)
# Check sequentially until a trading day is found in prices index
while candidate_date not in prices.index:
candidate_date += pd.Timedelta(days=1)
# Safety: if candidate_date moves into the next month, break out (unlikely if market is active)
if candidate_date.month != (current_date + relativedelta(months=rebalance_period)).month:
candidate_date = None
break
if candidate_date is None or candidate_date > end_date:
break
dates.append(candidate_date)
current_date = candidate_date
return dates
###############################################
# 3. Portfolio Initialization
###############################################
def initialize_portfolio(prices, date, tickers, initial_aum):
"""
On the start date, invest equal notional amounts in each asset.
Returns a dictionary mapping ticker -> quantity.
"""
portfolio = {}
mask = prices.index < date
prev_date = prices.index[mask][-1]
allocation = initial_aum / len(tickers)
for ticker in tickers:
price = prices.loc[prev_date, ticker]
portfolio[ticker] = allocation / price
return portfolio
###############################################
# 4. Lookback Metric Computation
###############################################
def compute_lookback_metric(prices, current_date, ticker, lookback_period, metric_type='simple'):
"""
Computes the lookback metric for one ticker using previous day's data.
"""
prices = prices.sort_index()
mask = prices.index < current_date
prev_date = prices.index[mask][-1]
lookback_date = prev_date - relativedelta(months=lookback_period)
current_price = prices[ticker].asof(prev_date) # Use previous day's price
lookback_price = prices[ticker].asof(lookback_date)
if pd.isna(current_price) or pd.isna(lookback_price):
raise ValueError(f"Missing price data for {ticker} on {prev_date} or {lookback_date}.")
if metric_type == 'simple':
metric = (current_price / lookback_price) - 1
elif metric_type == 'sma':
window = prices[ticker].loc[lookback_date:current_date]
if window.empty:
raise ValueError(f"No price data for {ticker} between {lookback_date} and {current_date}.")
sma = window.mean()
metric = (current_price - sma) / sma
else:
raise ValueError("Invalid metric type. Choose 'simple' or 'sma'.")
return metric
###############################################
# 5. Ranking Assets by Momentum
###############################################
def rank_assets(prices, current_date, tickers, lookback_period, metric_type):
"""
For a given observation date, compute the chosen lookback metric for each asset,
then sort (in descending order) so that the highest momentum gets rank 1.
Returns:
sorted_tickers: list of tickers in sorted order (best first)
ranks: dictionary mapping ticker -> rank (1 is best)
metrics: dictionary mapping ticker -> computed metric value
"""
metrics = {}
for ticker in tickers:
metric = compute_lookback_metric(prices, current_date, ticker, lookback_period, metric_type)
metrics[ticker] = metric
sorted_tickers = sorted(metrics, key=metrics.get, reverse=True)
ranks = {ticker: rank+1 for rank, ticker in enumerate(sorted_tickers)}
return sorted_tickers, ranks, metrics
###############################################
# 6. Compute Current Portfolio Value
###############################################
def compute_portfolio_value(portfolio, prices, current_date):
"""
Returns the portfolio AUM as of current_date.
"""
value = 0
for ticker, quantity in portfolio.items():
price = prices.loc[current_date, ticker]
value += quantity * price
return value
###############################################
# 7. Rebalance the Momentum Portfolio
###############################################
def rebalance_portfolio(portfolio, prices, current_date, tickers, sorted_tickers,
internal_rebalance_ratios, rebalance_ratio):
"""
Performs a partial (simulated) rebalance using a given rebalance_ratio.
Uses the previous day's portfolio value and today's prices.
Returns the new portfolio, the notional trades, and the previous portfolio value.
"""
mask = prices.index < current_date
prev_date = prices.index[mask][-1]
prev_prices = prices.loc[prev_date]
curr_prices = prices.loc[current_date]
portfolio_value = sum(portfolio[ticker] * prev_prices[ticker] for ticker in tickers)
rebalance_amount = portfolio_value * rebalance_ratio
target_trades = {ticker: rebalance_amount * internal_rebalance_ratios[i]
for i, ticker in enumerate(sorted_tickers)}
total_sold = 0
actual_trades = {}
for ticker, target_trade in target_trades.items():
if target_trade < 0:
available_notional = portfolio[ticker] * curr_prices[ticker]
sell_target = abs(target_trade)
actual_sell = min(available_notional, sell_target)
actual_trades[ticker] = -actual_sell
total_sold += actual_sell
else:
actual_trades[ticker] = 0
total_buy_target = sum(t for t in target_trades.values() if t > 0)
if total_buy_target > 0:
for ticker, target_trade in target_trades.items():
if target_trade > 0:
proportion = target_trade / total_buy_target
buy_amount = total_sold * proportion
actual_trades[ticker] = buy_amount
new_portfolio = portfolio.copy()
for ticker, trade_notional in actual_trades.items():
execution_price = curr_prices[ticker]
qty_change = trade_notional / execution_price
new_portfolio[ticker] += qty_change
return new_portfolio, actual_trades, portfolio_value
def adjust_overweight(portfolio, prices, current_date, sorted_tickers, threshold=0.70):
"""
Adjusts any asset whose weight is above the threshold.
The excess is redistributed to lower–weighted assets (in order of the ranking).
"""
mask = prices.index < current_date
prev_date = prices.index[mask][-1]
prev_prices = prices.loc[prev_date]
curr_prices = prices.loc[current_date]
portfolio_value = sum(portfolio[ticker] * prev_prices[ticker] for ticker in portfolio)
weights = {ticker: (portfolio[ticker] * prev_prices[ticker]) / portfolio_value
for ticker in portfolio}
new_portfolio = portfolio.copy()
for overweight in portfolio:
if weights[overweight] > threshold:
extra_weight = weights[overweight] - threshold
extra_value = extra_weight * portfolio_value
execution_price_over = curr_prices[overweight]
qty_reduce = extra_value / execution_price_over
new_portfolio[overweight] -= qty_reduce
remaining_value = extra_value
for candidate in sorted_tickers:
if candidate == overweight:
continue
candidate_value = new_portfolio[candidate] * curr_prices[candidate]
candidate_weight = candidate_value / portfolio_value
if candidate_weight < threshold:
capacity = (threshold - candidate_weight) * portfolio_value
allocation = min(remaining_value, capacity)
qty_add = allocation / curr_prices[candidate]
new_portfolio[candidate] += qty_add
remaining_value -= allocation
if remaining_value <= 0:
break
return new_portfolio
###############################################
# 8. VIX-based Allocation Function
###############################################
def momentum_allocation(vix_9d, vix_mean, vix_std, max_alloc=1.0, min_alloc=0.6):
"""
Maps composite score to a target momentum allocation fraction using a piecewise linear approach.
- If vix_9d < vix_mean + vix_std, return max_alloc (fully risk on).
- If vix_9d >= vix_mean + 2*vix_std, return min_alloc (fully risk off).
- Otherwise, return 0.8 as an intermediate allocation.
"""
if vix_9d < vix_mean + vix_std:
return max_alloc
elif (vix_9d >= vix_mean + vix_std) and (vix_9d < vix_mean + 2*vix_std):
return 0.8
else:
return min_alloc
###############################################
# 9. Compute FI Signal Functions
###############################################
def compute_fi_target_allocation(macro_data, current_date, fi_max_alloc, fi_min_alloc, slope_threshold=0.01):
"""
Computes the FI target allocation based on a refined logic using only the 8-day and 13-day EMAs.
The computation uses data only up to the previous trading day.
Returns:
target_alloc: the target allocation.
signal_label: a string label ("risk-on", "neutral", or "risk-off").
"""
available_dates = macro_data.index[macro_data.index < current_date]
if len(available_dates) == 0:
return fi_max_alloc, "risk-on"
ref_date = available_dates[-1]
fi_8 = macro_data["FI_EMA_8"].asof(ref_date)
fi_13 = macro_data["FI_EMA_13"].asof(ref_date)
if pd.isna(fi_8) or pd.isna(fi_13):
return fi_max_alloc, "risk-on"
available_ref_dates = macro_data.loc[:ref_date].index
if len(available_ref_dates) < 2:
slope = 0
else:
prev_ref_date = available_ref_dates[-2]
fi_8_prev = macro_data["FI_EMA_8"].asof(prev_ref_date)
slope = (fi_8/fi_8_prev) - 1
if fi_8 < fi_13:
return fi_max_alloc, "risk-on"
else: # fi_8 > fi_13
if slope > slope_threshold:
return fi_min_alloc, "risk-off"
else:
return fi_max_alloc, "no signal"
###############################################
# 10. Helper Functions for Daily Cash Rebalancing
###############################################
def invest_cash_into_portfolio(portfolio, prices, current_date, CASH):
"""
When switching from risk-off to risk-on, invest all available CASH back
into the securities. The allocation is based on previous-day weights.
"""
mask = prices.index < current_date
prev_date = prices.index[mask][-1]
mom_value = compute_portfolio_value(portfolio, prices, prev_date)
new_portfolio = portfolio.copy()
for ticker in portfolio:
# Use yesterday’s price for weight calculation and today’s price for execution
prev_price = prices.loc[prev_date, ticker]
weight = (portfolio[ticker] * prev_price) / mom_value if mom_value > 0 else 1/len(portfolio)
invest_amount = weight * CASH
price_today = prices.loc[current_date, ticker]
qty_to_buy = invest_amount / price_today
new_portfolio[ticker] += qty_to_buy
note = f"Invested {CASH:,.2f} CASH into portfolio."
return new_portfolio, 0.0, note
def allocate_cash_from_portfolio(portfolio, prices, current_date, target_alloc, CASH):
"""
When switching from risk-on to risk-off (or when target_alloc changes while in risk-off),
sell a proportional amount of each security so that the securities’ total notional becomes
target_alloc * (portfolio value + CASH). The sold amount is added to CASH.
"""
curr_value = compute_portfolio_value(portfolio, prices, current_date)
total_aum = curr_value + CASH
desired_value = target_alloc * total_aum
note = ""
new_portfolio = portfolio.copy()
cash_change = 0.0
if curr_value > desired_value:
# Sell from portfolio proportionally
excess = curr_value - desired_value
for ticker in portfolio:
price = prices.loc[current_date, ticker]
ticker_value = portfolio[ticker] * price
sell_amount = (ticker_value / curr_value) * excess
qty_to_sell = sell_amount / price
new_portfolio[ticker] -= qty_to_sell
cash_change = excess
note = f"Sold {excess:,.2f} from portfolio to CASH."
elif curr_value < desired_value and CASH > 0:
# Buy securities using available CASH
shortage = desired_value - curr_value
available = min(shortage, CASH)
for ticker in portfolio:
price = prices.loc[current_date, ticker]
ticker_value = portfolio[ticker] * price
weight = (ticker_value / curr_value) if curr_value > 0 else 1/len(portfolio)
invest_amount = weight * available
qty_to_buy = invest_amount / price
new_portfolio[ticker] += qty_to_buy
cash_change = -available
note = f"Bought securities using {available:,.2f} CASH."
return new_portfolio, cash_change, note
###############################################
# 11. Simulation: Refined Strategy with Cash & Regime Logic
###############################################
def simulate_strategy(prices, macro_data, eq_tickers, fi_tickers, alts_tickers,
initial_aum, start_date, end_date,
rebalance_period, rebalance_ratio,
lookback_period, metric_type,
internal_rebalance_ratios,
macro_max_alloc=1.0, macro_min_alloc=0.6):
"""
Runs the simulation with daily cash/regime adjustments and monthly rebalancing.
The refined logic includes:
• Daily determination of current regime (risk-on/risk-off) based on VIX & FI signals.
• If (SPY + HYG) weights are below 40% during risk-off, force risk-on and set target allocation to 100%.
• When switching from risk-on to risk-off (or when the target allocation changes while risk-off),
adjust the portfolio by selling a proportional amount of securities (keeping weights intact) and moving
the excess notional to CASH.
• When switching from risk-off to risk-on, invest all available CASH back into the portfolio.
• On monthly rebalancing dates, simulate a full rebalance using momentum rankings.
If (SPY + HYG) weights remain below 40% after simulation, force risk-on and reallocate full notional.
• Finally, adjust any overweight positions (assets crossing 70%).
Returns:
A DataFrame with the simulation log.
"""
tickers = eq_tickers + fi_tickers + alts_tickers
monthly_dates = get_observation_dates(prices, start_date, end_date, rebalance_period)
daily_dates = prices.index.sort_values()
daily_dates = daily_dates[(daily_dates >= start_date) & (daily_dates <= end_date)]
# Prepare macro data: compute VIX EMA fields and FI EMAs
macro_data = macro_data.copy()
macro_data['VIX 1M 3M_Spread'] = macro_data['VIX'] - macro_data['VIX3M']
macro_data['VIX1M_EMA'] = macro_data["VIX 1M 3M_Spread"].ewm(span=5, adjust=False).mean()
macro_data["Mean"] = macro_data['VIX1M_EMA'].rolling(window=504).mean()
macro_data["Std"] = macro_data['VIX1M_EMA'].rolling(window=504).std()
# Precompute FI EMAs
macro_data["FI_EMA_8"] = macro_data["LF98TRUU"].ewm(span=8, adjust=False).mean()
macro_data["FI_EMA_13"] = macro_data["LF98TRUU"].ewm(span=13, adjust=False).mean()
macro_data["FI_EMA_21"] = macro_data["LF98TRUU"].ewm(span=21, adjust=False).mean()
portfolio = initialize_portfolio(prices, start_date, tickers, initial_aum)
CASH = 0.0
# Start with risk-on by default
current_regime = 'risk-on'
target_alloc = 1.0
previous_regime = current_regime
previous_target_alloc = target_alloc
prev_total_aum = initial_aum
results = []
for current_date in daily_dates:
daily_note = "No adjustment"
cash_adjustment = 0.0
# Determine reference date (previous trading day) for signals (t-1)
available_macro_dates = macro_data.index[macro_data.index < current_date]
ref_date = available_macro_dates[-1] if len(available_macro_dates) > 0 else current_date
# --- FI Signal ---
fi_8 = macro_data["FI_EMA_8"].asof(ref_date)
fi_13 = macro_data["FI_EMA_13"].asof(ref_date)
available_ref_dates = macro_data.loc[:ref_date].index
if len(available_ref_dates) < 2:
fi_slope = 0
else:
prev_ref_date = available_ref_dates[-2]
fi_8_prev = macro_data["FI_EMA_8"].asof(prev_ref_date)
fi_slope = (fi_8/fi_8_prev) - 1
fi_target_alloc, fi_signal = compute_fi_target_allocation(
macro_data, current_date, macro_max_alloc, macro_min_alloc, slope_threshold=0.01)
# --- VIX Signal ---
vix_1m = macro_data['VIX'].asof(ref_date)
vix_3m = macro_data['VIX3M'].asof(ref_date)
try:
vix_ema = macro_data['VIX1M_EMA'].asof(ref_date)
vix_mean = macro_data['Mean'].asof(ref_date)
vix_std = macro_data['Std'].asof(ref_date)
except Exception as e:
raise ValueError(f"Error retrieving VIX data for {current_date}: {e}")
vix_target_alloc = momentum_allocation(vix_ema, vix_mean, vix_std,
max_alloc=macro_max_alloc,
min_alloc=macro_min_alloc)
if vix_ema >= (vix_mean + vix_std):
vix_signal = 'risk-off'
elif vix_ema <= (vix_mean - vix_std):
vix_signal = 'risk-on'
else:
vix_signal = 'no-signal'
# --- Determine Daily Regime Based on SPY & HYG Weights ---
mask = prices.index < current_date
prev_date = prices.index[mask][-1]
mom_value = compute_portfolio_value(portfolio, prices, prev_date)
# Compute SPY and HYG weights
spy_price = prices.loc[prev_date, 'SPY US Equity']
spy_qty = portfolio.get('SPY US Equity', 0)
spy_weight = (spy_price * spy_qty) / mom_value if mom_value > 0 else 0
hyg_price = prices.loc[prev_date, 'HYG US Equity']
hyg_qty = portfolio.get('HYG US Equity', 0)
hyg_weight = (hyg_price * hyg_qty) / mom_value if mom_value > 0 else 0
# Use FI & VIX signals to set regime
if (fi_signal == 'risk-off' or vix_signal == 'risk-off'):
if (spy_weight + hyg_weight) < 0.40:
current_regime = 'risk-on'
target_alloc = 1.0
daily_note = "Forced regime to risk-on & target alloc to 100% due to SPY+HYG < 40%"
else:
current_regime = 'risk-off'
target_alloc = min(fi_target_alloc, vix_target_alloc)
else:
current_regime = 'risk-on'
target_alloc = 1.0
# --- Daily Cash Rebalancing Logic ---
# Only trigger if regime changed OR if already risk-off and the target allocation has changed.
if (previous_regime != current_regime) or (current_regime == 'risk-off' and target_alloc != previous_target_alloc):
# Transition from risk-off to risk-on: reinvest all CASH
if previous_regime == 'risk-off' and current_regime == 'risk-on' and CASH > 0:
portfolio, CASH, note_update = invest_cash_into_portfolio(portfolio, prices, current_date, CASH)
daily_note += " | " + note_update
# Transition from risk-on to risk-off OR change in target alloc while risk-off: sell to build CASH
elif (previous_regime == 'risk-on' and current_regime == 'risk-off') or (current_regime == 'risk-off' and target_alloc != previous_target_alloc):
portfolio, cash_change, note_update = allocate_cash_from_portfolio(portfolio, prices, current_date, target_alloc, CASH)
CASH += cash_change
daily_note += " | " + note_update
# Update previous regime and target allocation for next iteration
previous_regime = current_regime
previous_target_alloc = target_alloc
# --- Monthly Rebalancing Block ---
if current_date in monthly_dates:
# Calculate momentum rankings
sorted_tickers, ranks, metrics = rank_assets(
prices, current_date, tickers, lookback_period, metric_type)
# Simulate a rebalance (partial, using the rebalance_ratio)
temp_portfolio, trades, pre_rebalance_value = rebalance_portfolio(
portfolio, prices, current_date, tickers, sorted_tickers,
internal_rebalance_ratios, rebalance_ratio)
# Check simulated SPY & HYG weights
temp_value = compute_portfolio_value(temp_portfolio, prices, current_date)
spy_temp = temp_portfolio.get('SPY US Equity', 0) * prices.loc[current_date, 'SPY US Equity']
hyg_temp = temp_portfolio.get('HYG US Equity', 0) * prices.loc[current_date, 'HYG US Equity']
combined_weight = (spy_temp + hyg_temp) / temp_value if temp_value > 0 else 0
# If after rebalancing the (SPY + HYG) weight is below 40% (in risk-off), force risk-on
if (current_regime == 'risk-off') and (combined_weight < 0.40):
current_regime = 'risk-on'
target_alloc = 1.0
daily_note += " | Monthly: Forced risk-on due to SPY+HYG weight < 40% after simulation."
# Fully reallocate the complete AUM using the simulated target weights.
total_aum = compute_portfolio_value(portfolio, prices, current_date) + CASH
simulated_value = temp_value # from simulated rebalance
new_portfolio = {}
for ticker in temp_portfolio:
price = prices.loc[current_date, ticker]
simulated_weight = (temp_portfolio[ticker]*price) / simulated_value if simulated_value > 0 else 1/len(temp_portfolio)
new_qty = (total_aum * simulated_weight) / price
new_portfolio[ticker] = new_qty
portfolio = new_portfolio
CASH = 0
else:
# Accept the simulated rebalance and then perform a cash adjustment if needed.
portfolio = temp_portfolio
curr_value = compute_portfolio_value(portfolio, prices, current_date)
total_aum = curr_value + CASH
desired_value = target_alloc * total_aum
if curr_value > desired_value:
portfolio, cash_change, note_update = allocate_cash_from_portfolio(portfolio, prices, current_date, target_alloc, CASH)
CASH += cash_change
daily_note += " | Monthly: " + note_update
elif curr_value < desired_value and CASH > 0:
portfolio, cash_change, note_update = allocate_cash_from_portfolio(portfolio, prices, current_date, target_alloc, CASH)
CASH += cash_change
daily_note += " | Monthly: " + note_update
# Adjust any overweight positions (assets crossing 70%)
portfolio = adjust_overweight(portfolio, prices, current_date, sorted_tickers, threshold=0.70)
# --- Update Return and Log Results ---
current_mom_value = compute_portfolio_value(portfolio, prices, current_date)
total_aum = current_mom_value + CASH
ret = (total_aum - prev_total_aum) / prev_total_aum if prev_total_aum > 0 else 0
prev_total_aum = total_aum
# Build log row
row = {
'Date': current_date,
'Momentum AUM': current_mom_value,
'CASH': CASH,
'Total AUM': total_aum,
'Current Regime': current_regime,
'Target Alloc': target_alloc,
'VIX Target': vix_target_alloc,
'VIX Signal': vix_signal,
'FI Target': fi_target_alloc,
'FI Signal': fi_signal,
'Adjustment Note': daily_note,
'Cash Adjustment': cash_adjustment,
'Return': ret,
'Event': 'Monthly Rebalance' if current_date in monthly_dates else 'Daily Check',
'FI_EMA_8': fi_8,
'FI_EMA_13': fi_13,
'FI_Slope': fi_slope,
'VIX_1M': vix_1m,
'VIX_3M': vix_3m,
'VIX_1M_ZScore': zscore(macro_data.loc[:ref_date, 'VIX'])[-1] if len(macro_data.loc[:ref_date]) > 0 else np.nan,
'VIX_3M_ZScore': zscore(macro_data.loc[:ref_date, 'VIX3M'])[-1] if len(macro_data.loc[:ref_date]) > 0 else np.nan
}
# Log details per asset
for ticker in tickers:
price = prices.loc[current_date, ticker]
qty = portfolio[ticker]
notional = qty * price
row[f'qty_{ticker}'] = qty
row[f'notional_{ticker}'] = notional
row[f'weight_{ticker}'] = (notional / current_mom_value) if current_mom_value > 0 else np.nan
row[f'rank_{ticker}'] = ranks.get(ticker, np.nan) if current_date in monthly_dates else np.nan
row[f'metric_{ticker}'] = metrics.get(ticker, np.nan) if current_date in monthly_dates else np.nan
row[f'trade_{ticker}'] = trades.get(ticker, 0) if current_date in monthly_dates else 0
results.append(row)
result_df = pd.DataFrame(results)
result_df.set_index('Date', inplace=True)
return result_df
###############################################
# 12. Main – Example Usage
###############################################
if __name__ == '__main__':
# Define asset tickers.
eq_tickers = ['SPY US Equity']
fi_tickers = ['TLT US Equity', 'HYG US Equity']
alts_tickers = ['GLD US Equity', 'SHV US Equity']
initial_aum = 100e6 # 100 million
start_date = pd.to_datetime('2008-01-01')
end_date = pd.to_datetime('2025-02-01')
rebalance_period = 1 # monthly (or adjust as desired)
rebalance_ratio = 0.1 # proportion of current momentum AUM rebalanced each period
lookback_period = 6
metric_type = 'simple'
internal_rebalance_ratios = [0.8, 0.2, 0, -0.2, -0.8]
# File paths (adjust these to your environment).
price_filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\Codes\Historic Prices.xlsx"
macro_filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\Momentum Strategy Overlay Data.xlsx"
prices = load_price_data(price_filepath)
macro_data = load_macro_data(macro_filepath)
# Run simulation.
result_df = simulate_strategy(prices, macro_data,
eq_tickers, fi_tickers, alts_tickers,
initial_aum, start_date, end_date,
rebalance_period, rebalance_ratio,
lookback_period, metric_type,
internal_rebalance_ratios,
macro_max_alloc=1.0, macro_min_alloc=0.6)
pd.set_option('display.float_format', lambda x: f'{x:,.2f}')
print(result_df)
Editor is loading...
Leave a Comment