Untitled
unknown
plain_text
a month ago
17 kB
3
Indexable
import pandas as pd
import numpy as np
from datetime import datetime
from dateutil.relativedelta import relativedelta
###############################################
# 1. Data Loading
###############################################
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() # ensure sorted index for asof()
# Filter out weekends (Saturday=5, Sunday=6)
df = df[df.index.dayofweek < 5]
return df
###############################################
# 2. Helper: Observation Dates
###############################################
# def get_observation_dates(start_date, end_date, rebalance_period):
# """
# Returns a list of observation dates from start_date to end_date
# with a step equal to rebalance_period (in months).
# """
# dates = []
# current = start_date
# while current <= end_date:
# dates.append(current)
# current += relativedelta(months=rebalance_period)
# return dates
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 = [start_date]
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. Initialize the Portfolio
###############################################
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 Portfolio
###############################################
def rebalance_portfolio(portfolio, prices, current_date, tickers, sorted_tickers,
internal_rebalance_ratios, rebalance_ratio):
prev_date = prices.index[prices.index < current_date][-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():
execution_price = curr_prices[ticker]
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):
prev_date = prices.index[prices.index < current_date][-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()
adjustments = {}
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
adjustments[overweight] = {'removed_qty': qty_reduce, 'reallocated': {}}
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
adjustments[overweight]['reallocated'][candidate] = qty_add
remaining_value -= allocation
if remaining_value <= 0:
break
if remaining_value > 0:
adjustments[overweight]['leftover_value'] = remaining_value
return new_portfolio, adjustments
###############################################
# 10. Daily Metrics
###############################################
def calculate_daily_metrics(portfolio_snapshot, prices, date, tickers):
"""
Calculate daily portfolio metrics using frozen quantities (from last rebalance)
and current day's prices.
Parameters:
- portfolio_snapshot: Dictionary of ticker -> frozen quantity (from last rebalance)
- prices: Price DataFrame
- date: Current date
- tickers: List of tickers
Returns:
- Dictionary containing daily portfolio metrics
"""
daily_metrics = {'Date': date}
portfolio_value = 0.0
# Calculate notionals and total portfolio value
for ticker in tickers:
quantity = portfolio_snapshot[ticker]
price = prices.loc[date, ticker]
notional = quantity * price
daily_metrics[f'qty_{ticker}'] = quantity
daily_metrics[f'notional_{ticker}'] = notional
portfolio_value += notional
# Calculate weights
for ticker in tickers:
weight = daily_metrics[f'notional_{ticker}'] / portfolio_value
daily_metrics[f'weight_{ticker}'] = weight
daily_metrics['Portfolio Value'] = portfolio_value
return daily_metrics
###############################################
# 9. Simulate the Strategy
###############################################
def simulate_strategy_daily(prices, eq_tickers, fi_tickers, alts_tickers,
initial_aum, start_date, end_date,
rebalance_period, rebalance_ratio,
lookback_period, metric_type,
internal_rebalance_ratios):
tickers = eq_tickers + fi_tickers + alts_tickers
obs_dates = get_observation_dates(prices, start_date, end_date, rebalance_period)
results = []
# Step 1: Initialize portfolio
portfolio = initialize_portfolio(prices, start_date, tickers, initial_aum)
frozen_portfolio = portfolio.copy() # quantities used for metrics
trading_days = prices.index[(prices.index >= start_date) & (prices.index <= end_date)]
prev_portfolio_value = None
for date in trading_days:
# Step 2: Use frozen quantities for today's metrics
daily_metrics = calculate_daily_metrics(frozen_portfolio, prices, date, tickers)
# Step 3: Calculate daily return
if prev_portfolio_value is not None:
daily_metrics['Return'] = (daily_metrics['Portfolio Value'] - prev_portfolio_value) / prev_portfolio_value
else:
daily_metrics['Return'] = 0
prev_portfolio_value = daily_metrics['Portfolio Value']
# Step 4: Check for rebalance day
if date in obs_dates and date != start_date:
daily_metrics['Is_Rebalance_Date'] = True
# Compute momentum scores and ranks
sorted_tickers, ranks, metrics = rank_assets(prices, date, tickers,
lookback_period, metric_type)
# Rebalance portfolio
portfolio, trades, pre_rebalance_value = rebalance_portfolio(
portfolio, prices, date, tickers, sorted_tickers,
internal_rebalance_ratios, rebalance_ratio)
# Adjust overweight holdings
portfolio, adjustments = adjust_overweight(
portfolio, prices, date, sorted_tickers, threshold=0.70)
# Log rebalance results
daily_metrics['Pre_Rebalance_Value'] = pre_rebalance_value
daily_metrics['Adjustments'] = adjustments
for ticker in tickers:
daily_metrics[f'rank_{ticker}'] = ranks.get(ticker, np.nan)
daily_metrics[f'metric_{ticker}'] = metrics.get(ticker, np.nan)
daily_metrics[f'trade_{ticker}'] = trades.get(ticker, 0)
# Save frozen portfolio to be applied from next day
frozen_portfolio_next_day = portfolio.copy()
else:
daily_metrics['Is_Rebalance_Date'] = False
# Step 5: Store metrics for the day
results.append(daily_metrics)
# Step 6: Apply new portfolio starting the next day (not today)
if 'frozen_portfolio_next_day' in locals():
frozen_portfolio = frozen_portfolio_next_day
del frozen_portfolio_next_day # avoid reapplying accidentally
# Final formatting
result_df = pd.DataFrame(results)
result_df.set_index('Date', inplace=True)
# Organize columns
column_groups = ['Portfolio Value', 'Return', 'Is_Rebalance_Date']
if 'Pre_Rebalance_Value' in result_df.columns:
column_groups.append('Pre_Rebalance_Value')
if 'Adjustments' in result_df.columns:
column_groups.append('Adjustments')
for prefix in ['qty_', 'notional_', 'weight_']:
column_groups.extend([f'{prefix}{ticker}' for ticker in tickers])
for prefix in ['rank_', 'metric_', 'trade_']:
cols = [f'{prefix}{ticker}' for ticker in tickers]
if any(col in result_df.columns for col in cols):
column_groups.extend(cols)
result_df = result_df[column_groups]
return result_df
###############################################
# 10. Main – Example Usage
###############################################
if __name__ == '__main__':
# Define the asset tickers.
# 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 # 20% 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]
# Specify the location of the Excel file.
filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\Codes\Historic Prices.xlsx"
prices = load_price_data(filepath)
prices = prices.sort_index()
# Run the simulation.
result_df = simulate_strategy_daily(prices, eq_tickers, fi_tickers, alts_tickers,
initial_aum, start_date, end_date,
rebalance_period, rebalance_ratio,
lookback_period, metric_type,
internal_rebalance_ratios)
# Display the final results.
pd.set_option('display.float_format', lambda x: f'{x:,.2f}')
print(result_df)
Editor is loading...
Leave a Comment