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