Untitled
unknown
plain_text
13 days ago
13 kB
2
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 (with Cash)
###############################################
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 and includes a 'cash' key.
Initially, we are fully invested so cash is 0.
"""
portfolio = {}
allocation = initial_aum / len(tickers)
for ticker in tickers:
price = prices.loc[date, ticker]
portfolio[ticker] = allocation / price
portfolio["cash"] = 0.0
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.
"""
prices = prices.sort_index()
lookback_date = current_date - relativedelta(months=lookback_period)
current_price = prices[ticker].asof(current_date)
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 {current_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):
"""
Computes the lookback metric for each asset and returns sorted tickers, their ranks, and metrics.
"""
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 (including Cash)
###############################################
def compute_portfolio_value(portfolio, prices, current_date):
"""
Returns the portfolio AUM (assets + cash) as of current_date.
"""
asset_value = sum(quantity * prices.loc[current_date, ticker]
for ticker, quantity in portfolio.items() if ticker != "cash")
cash = portfolio.get("cash", 0)
return asset_value + cash
###############################################
# 7. Volatility Scaling: Composite Risk Signal
###############################################
def get_allocation_multiplier(macro_indics, current_date, w_vix, w_credit, w_cons, k, threshold):
"""
Compute the allocation multiplier based on macro indicators.
"""
current_date = pd.to_datetime(current_date)
if not isinstance(macro_indics.index, pd.DatetimeIndex):
raise ValueError("macro_indics index is not a DatetimeIndex")
available_dates = macro_indics.index[macro_indics.index <= current_date]
if len(available_dates) == 0:
print(f"No macro indicator data available on or before {current_date}")
return 1.0
most_recent_date = available_dates[-1]
current_data = macro_indics.loc[most_recent_date]
historical_data = macro_indics.loc[:most_recent_date]
vix_mean, vix_std = historical_data['VIX'].mean(), historical_data['VIX'].std()
credit_mean, credit_std = historical_data['CreditSpread'].mean(), historical_data['CreditSpread'].std()
cons_mean, cons_std = historical_data['ConsumerConfidence'].mean(), historical_data['ConsumerConfidence'].std()
vix_z = (current_data['VIX'] - vix_mean) / vix_std
credit_z = (current_data['CreditSpread'] - credit_mean) / credit_std
cons_z = (current_data['ConsumerConfidence'] - cons_mean) / cons_std
composite_score = w_vix * vix_z + w_credit * credit_z - w_cons * cons_z
multiplier = 1 / (1 + np.exp(-k * (composite_score - threshold)))
return multiplier
###############################################
# 8. Rebalance the Portfolio with Cash Allocation
###############################################
def rebalance_portfolio_with_cash(portfolio, prices, current_date, tickers, internal_weights, allocation_multiplier):
"""
At rebalancing, target:
- Invest a fraction = allocation_multiplier of the portfolio in assets.
- Hold the rest as cash.
internal_weights: list of target weights for each asset (should sum to 1).
"""
# Compute total portfolio value (invested assets + cash)
total_value = compute_portfolio_value(portfolio, prices, current_date)
# Compute target invested amount and target cash amount.
target_invested_value = allocation_multiplier * total_value
target_cash_value = total_value - target_invested_value
new_portfolio = {}
trades = {}
# For each asset, determine target notional and compute trade notional.
for i, ticker in enumerate(tickers):
price = prices.loc[current_date, ticker]
target_notional = internal_weights[i] * target_invested_value
current_notional = portfolio.get(ticker, 0) * price
trade_notional = target_notional - current_notional
trades[ticker] = trade_notional
new_portfolio[ticker] = (current_notional + trade_notional) / price
# Set cash position to the target cash.
new_portfolio["cash"] = target_cash_value
return new_portfolio, trades, total_value
###############################################
# 9. Simulate the Strategy with Cash Holding
###############################################
def simulate_strategy(prices, eq_tickers, fi_tickers, alts_tickers,
initial_aum, start_date, end_date,
rebalance_period,
lookback_period, metric_type,
internal_weights, # list of weights for assets that sum to 1
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 each rebalancing date:
- Compute the allocation multiplier from macro indicators.
- Determine the target invested portion (and thus cash).
- Rebalance assets to match internal_weights over the target invested amount.
"""
# Prepare macro_indics DataFrame
macro_indics.set_index('Date', inplace=True)
macro_indics.sort_index(inplace=True)
tickers = eq_tickers + fi_tickers + alts_tickers
obs_dates = get_observation_dates(start_date, end_date, rebalance_period)
results = []
# 9a. Initial portfolio (fully invested; cash = 0)
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},
'cash': portfolio.get("cash", 0),
'Return': 0,
'Multiplier': 1.0 # Fully invested initially
})
prev_value = portfolio_value
# 9b. Loop over each observation date after the start date.
for current_date in obs_dates[1:]:
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)
# Here we ignore ranking for allocation; we simply use the provided internal_weights.
portfolio, trades, pre_rebalance_value = rebalance_portfolio_with_cash(
portfolio, prices, current_date, tickers, internal_weights, 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,
'cash': portfolio.get("cash", 0)
}
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
# We keep the ranking and metric info from the momentum calculation if needed.
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)
return result_df
###############################################
# 10. Main – Example Usage
###############################################
if __name__ == '__main__':
# Define the asset tickers.
eq_tickers = ['SPY US Equity']
fi_tickers = ['TLT US Equity', 'HYG US Equity']
alts_tickers = ['GLD US Equity', 'SHV US Equity', 'VNQ 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
lookback_period = 3 # 3-month lookback
metric_type = 'simple' # Use simple return metric
# Define internal weights for assets (must sum to 1).
# For instance, if you want 70% in the best asset, 30% in the second, etc.
# Ensure the length matches total number of assets (eq + fi + alts).
internal_weights = [0.3, 0.3, 0.1, 0.15, 0.15] # Example weights for 5 assets
# Specify the filepaths for prices and macro indicators.
price_filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\Codes\Historic Prices.xlsx"
prices = load_price_data(price_filepath)
macro_filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\macro_indic.xlsx"
macro_indics = load_macro_indics(macro_filepath)
macro_indics = macro_indics.reset_index()
# Run the simulation.
# Here, if the multiplier is low (e.g., 0.5) then only 50% of the portfolio is invested,
# and the remaining 50% is held in cash until the next rebalance.
result_df = simulate_strategy(prices, eq_tickers, fi_tickers, alts_tickers,
initial_aum, start_date, end_date,
rebalance_period, lookback_period, metric_type,
internal_weights,
macro_indics,
w_vix=1.3, w_credit=0.9, w_cons=1.1, k=0.1, threshold=-0.3)
# Display the final results.
pd.set_option('display.float_format', lambda x: f'{x:,.2f}')
print(result_df)
Editor is loading...
Leave a Comment