Untitled
unknown
plain_text
4 days ago
16 kB
4
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.
"""
df = pd.read_excel(filepath, index_col=0)
df.index = pd.to_datetime(df.index)
return df
def load_macro_indics(filepath):
"""
Load macro indicators from an Excel file.
Assumes that the first column is dates and the next three columns are:
- VIX
- Credit Spread
- Consumer Confidence
"""
df = pd.read_excel(filepath, index_col=0)
df.index = pd.to_datetime(df.index)
# Rename columns for clarity; adjust if your file already has header names.
df.columns = ['VIX', 'CreditSpread', 'ConsumerConfidence']
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
###############################################
# 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 = {}
allocation = initial_aum / len(tickers)
for ticker in tickers:
price = prices.loc[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.
- For 'simple': metric = (Price_today / Price_lookback) - 1.
- For 'sma': metric = (Price_today - SMA) / SMA, where SMA is the simple moving average
over the lookback period.
This version uses the asof method to get the nearest available price on or before the date.
"""
# Ensure the DataFrame index is sorted.
prices = prices.sort_index()
# Define the lookback date.
lookback_date = current_date - relativedelta(months=lookback_period)
# Use 'asof' to fetch the price on or before the specified dates.
current_price = prices[ticker].asof(current_date)
lookback_price = prices[ticker].asof(lookback_date)
# Check if prices were successfully retrieved.
if pd.isna(current_price) or pd.isna(lookback_price):
raise ValueError(f"Missing price data for {ticker} on {current_date} or {lookback_date}.")
if metric_type == 'simple':
metric = (current_price / lookback_price) - 1
elif metric_type == 'sma':
# Get all prices between lookback_date and current_date.
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. Volatility Scaling: Composite Risk Signal
###############################################
def get_allocation_multiplier(macro_indics, current_date, w_vix=1.0, w_credit=1.0, w_cons=1.0, k=1.0, threshold=0):
"""
Computes an allocation multiplier based on macro indicators.
The macro_indics DataFrame is expected to have an index of dates and columns:
'VIX', 'CreditSpread', and 'ConsumerConfidence'.
For each indicator, a z-score is computed using the full historical sample.
The composite risk is:
composite_risk = w_vix * z(VIX) + w_credit * z(CreditSpread) - w_cons * z(ConsumerConfidence)
(Note the minus sign on consumer confidence because lower confidence implies higher risk.)
A logistic function then maps the composite risk to a multiplier between 0 and 1.
"""
# Get the most recent available macro data as of current_date.
current_data = macro_indics.loc[:current_date].iloc[-1]
# Compute historical means and standard deviations for each indicator.
vix_mean = macro_indics['VIX'].mean()
vix_std = macro_indics['VIX'].std()
credit_mean = macro_indics['CreditSpread'].mean()
credit_std = macro_indics['CreditSpread'].std()
cons_mean = macro_indics['ConsumerConfidence'].mean()
cons_std = macro_indics['ConsumerConfidence'].std()
# Compute z-scores.
z_vix = (current_data['VIX'] - vix_mean) / vix_std
z_credit = (current_data['CreditSpread'] - credit_mean) / credit_std
z_cons = (current_data['ConsumerConfidence'] - cons_mean) / cons_std
# Combine into a composite risk signal.
composite_risk = w_vix * z_vix + w_credit * z_credit - w_cons * z_cons
# Map the composite risk to a multiplier using a logistic function.
multiplier = 1 / (1 + np.exp(k * (composite_risk - threshold)))
return multiplier
###############################################
# 8. Rebalance the Portfolio (with volatility scaling)
###############################################
def rebalance_portfolio(portfolio, prices, current_date, tickers, sorted_tickers,
internal_rebalance_ratios, rebalance_ratio, allocation_multiplier):
"""
On an observation date:
(a) Compute the current portfolio value and the rebalancing notional.
(b) The notional to rebalance is scaled by the allocation_multiplier.
(c) For each asset, assign a target trade notional = (rebalance_amount * internal ratio),
where the internal ratio is picked according to the asset's rank (from sorted_tickers).
(d) For sell orders (negative target trades), ensure that the available notional (quantity × current price)
is sufficient. If not, sell only what is available.
(e) Redistribute the total sold amount among the buying orders (assets with positive target trades)
proportionally.
(f) Update each asset's quantity accordingly.
Returns the updated portfolio along with some diagnostics.
"""
# 1. Compute current AUM and the scaled rebalance notional.
portfolio_value = compute_portfolio_value(portfolio, prices, current_date)
rebalance_amount = portfolio_value * rebalance_ratio * allocation_multiplier
# 2. Compute target trades for each asset according to the ranking.
target_trades = {ticker: rebalance_amount * internal_rebalance_ratios[i]
for i, ticker in enumerate(sorted_tickers)}
# 3. For sell orders: check available notional and determine actual trades.
total_sold = 0
actual_trades = {}
for ticker, target_trade in target_trades.items():
price = prices.loc[current_date, ticker]
if target_trade < 0:
available_notional = portfolio[ticker] * price
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
# 4. Redistribute the total_sold to the buying side.
total_buy_target = sum(trade for trade in target_trades.values() if trade > 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
# 5. Update portfolio quantities.
new_portfolio = portfolio.copy()
for ticker, trade_notional in actual_trades.items():
price = prices.loc[current_date, ticker]
qty_change = trade_notional / price
new_portfolio[ticker] += qty_change
return new_portfolio, actual_trades, portfolio_value
###############################################
# 9. Simulate the Strategy (with volatility scaling)
###############################################
def simulate_strategy(prices, eq_tickers, fi_tickers, alts_tickers,
initial_aum, start_date, end_date,
rebalance_period, rebalance_ratio,
lookback_period, metric_type,
internal_rebalance_ratios,
macro_indics,
w_vix=1.0, w_credit=1.0, w_cons=1.0, k=1.0, threshold=0):
"""
Runs the simulation from start_date to end_date.
At t0, we initialize the portfolio with equal weights.
At each observation date, we compute lookback metrics, rank assets,
then rebalance a fixed percentage (rebalance_ratio) of the current AUM,
scaled by an allocation multiplier derived from macro indicators.
The results (portfolio AUM, individual quantities and notionals, returns, etc.)
are recorded in a DataFrame.
"""
tickers = eq_tickers + fi_tickers + alts_tickers
obs_dates = get_observation_dates(start_date, end_date, rebalance_period)
results = []
# 9a. Initial portfolio (at start_date)
portfolio = initialize_portfolio(prices, start_date, tickers, initial_aum)
portfolio_value = compute_portfolio_value(portfolio, prices, start_date)
results.append({
'Date': start_date,
'Portfolio Value': portfolio_value,
**{f'qty_{ticker}': portfolio[ticker] for ticker in tickers},
**{f'notional_{ticker}': portfolio[ticker] * prices.loc[start_date, ticker] for ticker in tickers},
'Return': 0
})
prev_value = portfolio_value
# 9b. Loop over each observation date (after the start date).
for current_date in obs_dates[1:]:
# Compute the composite allocation multiplier from macro indicators.
multiplier = get_allocation_multiplier(macro_indics, current_date, w_vix, w_credit, w_cons, k, threshold)
sorted_tickers, ranks, metrics = rank_assets(prices, current_date, tickers,
lookback_period, metric_type)
portfolio, trades, pre_rebalance_value = rebalance_portfolio(
portfolio, prices, current_date, tickers, sorted_tickers,
internal_rebalance_ratios, rebalance_ratio, allocation_multiplier=multiplier)
portfolio_value = compute_portfolio_value(portfolio, prices, current_date)
ret = (portfolio_value - prev_value) / prev_value
prev_value = portfolio_value
row = {
'Date': current_date,
'Portfolio Value': portfolio_value,
'Return': ret,
'Pre-Rebalance Value': pre_rebalance_value,
'Multiplier': multiplier
}
for ticker in tickers:
row[f'qty_{ticker}'] = portfolio[ticker]
row[f'notional_{ticker}'] = portfolio[ticker] * prices.loc[current_date, ticker]
row[f'weight_{ticker}'] = row[f'notional_{ticker}'] / portfolio_value
row[f'rank_{ticker}'] = ranks.get(ticker, np.nan)
row[f'metric_{ticker}'] = metrics.get(ticker, np.nan)
row[f'trade_{ticker}'] = trades.get(ticker, 0)
results.append(row)
result_df = pd.DataFrame(results)
result_df.set_index('Date', inplace=True)
column_groups = ['Portfolio Value', 'Return', 'Pre-Rebalance Value', 'Multiplier']
for prefix in ['qty_', 'notional_', 'weight_', 'rank_', 'metric_', 'trade_']:
column_groups.extend([f'{prefix}{ticker}' for ticker in tickers])
result_df = result_df[column_groups]
return result_df
###############################################
# 10. Main – Example Usage
###############################################
if __name__ == '__main__':
# Define the asset tickers.
eq_tickers = ['SPY US Equity']
fi_tickers = ['TLT US Equity']
alts_tickers = ['GLD US Equity', 'SHV US Equity']
initial_aum = 100e6 # e.g., 100 million
start_date = pd.to_datetime('2008-01-01')
end_date = pd.to_datetime('2025-02-01')
rebalance_period = 2 # rebalance every 2 months
rebalance_ratio = 0.2 # 20% of current AUM is rebalanced each period
lookback_period = 3 # 3-month lookback
metric_type = 'simple' # use simple return metric; alternatively, set 'sma'
# Define the internal rebalancing mapping.
# For example, for 4 assets (sorted best-to-worst) the ratios are:
# Best asset: +80%, second: +20%, third: -20%, worst: -80%
internal_rebalance_ratios = [0.8, 0.2, -0.2, -0.8]
# Specify the location of the Excel file for prices.
price_filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\Codes\Historic Prices.xlsx"
prices = load_price_data(price_filepath)
# Specify the location of the Excel file for macro indicators.
macro_filepath = r"\\path\to\your\macro_indics.xlsx"
macro_indics = load_macro_indics(macro_filepath)
# Run the simulation.
result_df = simulate_strategy(prices, eq_tickers, fi_tickers, alts_tickers,
initial_aum, start_date, end_date,
rebalance_period, rebalance_ratio,
lookback_period, metric_type,
internal_rebalance_ratios,
macro_indics,
w_vix=1.0, w_credit=1.0, w_cons=1.0, k=1.0, threshold=0)
# Display the final results.
pd.set_option('display.float_format', lambda x: f'{x:,.2f}')
print(result_df)
Editor is loading...
Leave a Comment