Untitled

 avatar
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