Untitled

 avatar
unknown
plain_text
a month ago
19 kB
2
Indexable
import pandas as pd
import numpy as np
from datetime import datetime
from dateutil.relativedelta import relativedelta
from hmmlearn.hmm import GaussianHMM

###############################################
# 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_data(filepath):
    """
    Load macro indicators from an Excel file.
    Assumes first column is Date, then VIX, Consumer Confidence, IG Spreads.
    """
    df = pd.read_excel(filepath, index_col=0)
    df.index = pd.to_datetime(df.index)
    df.columns = ['VIX', 'ConsumerConfidence', 'IGSpreads']
    # Fill missing values
    df = df.fillna(method='ffill').fillna(method='bfill')
    return df

###############################################
# 2. Helper: Observation Dates
###############################################

def get_observation_dates(start_date, end_date, rebalance_period):
    dates = []
    current = start_date
    while current <= end_date:
        dates.append(current)
        current += relativedelta(months=rebalance_period)
    return dates

###############################################
# 3. Initialize the Portfolio & CASH
###############################################

def initialize_portfolio(prices, date, tickers, initial_aum):
    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'):
    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):
    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):
    value = 0
    for ticker, quantity in portfolio.items():
        price = prices.loc[current_date, ticker]
        value += quantity * price
    return value

###############################################
# 7. Rebalance the Portfolio (Momentum Portion Only)
###############################################

def rebalance_portfolio(portfolio, prices, current_date, tickers, sorted_tickers,
                        internal_rebalance_ratios, rebalance_ratio):
    portfolio_value = compute_portfolio_value(portfolio, prices, current_date)
    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():
        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
    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
    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

###############################################
# 8. Macro: Composite Risk & HMM Functions
###############################################

def scale(x, low, high):
    """Safely scale x between 0 and 1 given bounds low and high."""
    if high - low == 0:
        return 0.0
    return np.clip((x - low) / (high - low), 0, 1)

def compute_composite_risk(vix, consumer_confidence, ig_spreads, params):
    """
    Convert each macro indicator into a risk score (0 to 1) using preset bounds.
    For Consumer Confidence, higher values mean lower risk, so we invert the score.
    """
    vix_low, vix_high = params['VIX']
    cc_low, cc_high = params['ConsumerConfidence']
    ig_low, ig_high = params['IGSpreads']
    
    risk_vix = scale(vix, vix_low, vix_high)
    risk_cc = 1 - scale(consumer_confidence, cc_low, cc_high)
    risk_ig = scale(ig_spreads, ig_low, ig_high)
    
    composite_risk = (risk_vix + risk_cc + risk_ig) / 3.0
    if np.isnan(composite_risk) or np.isinf(composite_risk):
        composite_risk = 0.5
    return composite_risk

def fit_hmm_on_series(series, n_components=2):
    """
    Fit a Gaussian HMM on a 1D numpy array of composite risk values.
    Returns the fitted model and index of the bear state (state with higher mean).
    """
    X = series.reshape(-1, 1)
    model = GaussianHMM(n_components=n_components, covariance_type="diag", n_iter=1000, random_state=42)
    model.fit(X)
    state_means = model.means_.flatten()
    bear_state = np.argmax(state_means)
    return model, bear_state

def compute_bear_probability(model, bear_state, current_value):
    """
    Compute the probability that the current composite risk value is generated by the bear state.
    """
    prob = model.predict_proba(np.array([[current_value]]))[0]
    p_bear = prob[bear_state]
    return p_bear

def regime_allocation_from_hmm(p_bear, max_alloc=1.0, min_alloc=0.3, threshold=0.8):
    """
    Map bear probability to target momentum allocation.
    If p_bear < threshold, return max_alloc. Otherwise, linearly interpolate.
    """
    if p_bear < threshold:
        return max_alloc
    else:
        return max_alloc - (max_alloc - min_alloc) * ((p_bear - threshold) / (1 - threshold))

###############################################
# 9. Simulate the Strategy with HMM-Based Regime Switching & CASH Adjustments
###############################################

def simulate_strategy_with_hmm(prices, macro, eq_tickers, fi_tickers, alts_tickers,
                               initial_aum, start_date, end_date,
                               rebalance_period, rebalance_ratio,
                               lookback_period, metric_type,
                               internal_rebalance_ratios,
                               macro_params,
                               max_alloc=1.0, min_alloc=0.3,
                               hmm_window=24):
    """
    Runs the simulation using an HMM on the composite risk signal.
    
    For each rebalance date:
      - Retrieve macro data (with missing values filled).
      - Smooth the indicators using a 1-month moving average, using .asof() to fetch the nearest value.
      - Compute composite risk.
      - Append the risk to a rolling window; if enough points exist, fit a Gaussian HMM and compute the bear probability.
      - Map the bear probability to target momentum allocation.
      - Rebalance the momentum portfolio and adjust CASH accordingly.
    """
    tickers = eq_tickers + fi_tickers + alts_tickers
    obs_dates = get_observation_dates(start_date, end_date, rebalance_period)
    results = []
    
    # Rolling composite risk series for HMM fitting.
    composite_risk_series = []
    
    # Initial portfolio: fully invested in momentum, CASH = 0.
    portfolio = initialize_portfolio(prices, start_date, tickers, initial_aum)
    CASH = 0.0
    portfolio_value = compute_portfolio_value(portfolio, prices, start_date)
    total_aum = portfolio_value + CASH
    prev_target_alloc = max_alloc
    prev_regime = 'risk-on'
    
    results.append({
        'Date': start_date,
        'Portfolio Value': portfolio_value,
        'Momentum AUM': portfolio_value,
        'CASH': CASH,
        'Total AUM': total_aum,
        'VIX': np.nan,
        'ConsumerConfidence': np.nan,
        'IGSpreads': np.nan,
        'Composite Risk': np.nan,
        'Bear Prob': np.nan,
        'Target Momentum Alloc': prev_target_alloc,
        'Regime': prev_regime,
        'Return': 0,
        **{f'qty_{ticker}': portfolio[ticker] for ticker in tickers},
        **{f'notional_{ticker}': portfolio[ticker] * prices.loc[start_date, ticker] for ticker in tickers},
    })
    
    prev_total_aum = total_aum
    
    vix_monthly = macro['VIX'].resample('M').last()
    cc_monthly = macro['ConsumerConfidence'].resample('M').last()
    ig_monthly = macro['IGSpreads'].resample('M').last()
    
    for current_date in obs_dates[1:]:
        # Retrieve macro data using asof to get nearest available value.
        try:
            vix = macro['VIX'].asof(current_date)
            cc = macro['ConsumerConfidence'].asof(current_date)
            ig = macro['IGSpreads'].asof(current_date)
        except Exception as e:
            raise ValueError(f"Error retrieving macro data for {current_date}: {e}")
        
        # Smooth the indicators 
        window = 3
        vix_roll = macro['VIX'].rolling(window=window, min_periods=1).mean()
        cc_roll = macro['ConsumerConfidence'].rolling(window=window, min_periods=1).mean()
        ig_roll = macro['IGSpreads'].rolling(window=window, min_periods=1).mean()
        vix_smoothed = vix_roll.asof(current_date)
        cc_smoothed = cc_roll.asof(current_date)
        ig_smoothed = ig_roll.asof(current_date)
        
        # Compute composite risk.
        composite_risk = compute_composite_risk(vix, cc, ig, macro_params)
        composite_risk_series.append(composite_risk)
        
        # Fit HMM if we have enough data.
        if len(composite_risk_series) >= hmm_window:
            window_data = np.array(composite_risk_series[-hmm_window:])
            if np.any(np.isnan(window_data)) or np.any(np.isinf(window_data)):
                window_data = np.nan_to_num(window_data, nan=0.5, posinf=0.5, neginf=0.5)
            model, bear_state = fit_hmm_on_series(window_data)
            p_bear = compute_bear_probability(model, bear_state, composite_risk)
        else:
            p_bear = 0.0  # Default to risk-on if insufficient data.
        
        # Map bear probability to target momentum allocation.
        target_alloc = regime_allocation_from_hmm(p_bear, max_alloc, min_alloc, threshold=0.8)
        regime = 'risk-on' if target_alloc >= max_alloc * 0.99 else 'risk-off'
        
#         print(f"Date: {current_date.date()}, VIX: {vix:.2f}, CC: {cc:.2f}, IG: {ig:.2f}")
#         print(f"Composite Risk: {composite_risk:.2f}, Bear Prob: {p_bear:.2f}, Target Alloc: {target_alloc:.2f}, Regime: {regime}")
        
        # Rebalance momentum portfolio.
        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)
        
        # Compute momentum portfolio value and total investment.
        mom_value = compute_portfolio_value(portfolio, prices, current_date)
        total_investment = mom_value + CASH
        
        # Determine desired momentum value and desired CASH.
        desired_mom_value = target_alloc * total_investment
        desired_cash = total_investment - desired_mom_value
        
        # Adjust portfolio and CASH.
        if mom_value > desired_mom_value:
            excess = mom_value - desired_mom_value
            for ticker in portfolio.keys():
                price = prices.loc[current_date, ticker]
                ticker_value = portfolio[ticker] * price
                sell_amount = (ticker_value / mom_value) * excess
                qty_to_sell = sell_amount / price
                portfolio[ticker] -= qty_to_sell
            CASH += excess
            adjustment_note = f"Sold excess {excess:.2f} to CASH."
        elif mom_value < desired_mom_value:
            shortage = desired_mom_value - mom_value
            if CASH >= shortage:
                for ticker in portfolio.keys():
                    price = prices.loc[current_date, ticker]
                    ticker_value = portfolio[ticker] * price
                    target_weight = ticker_value / mom_value if mom_value > 0 else 1/len(portfolio)
                    invest_amount = target_weight * shortage
                    qty_to_buy = invest_amount / price
                    portfolio[ticker] += qty_to_buy
                CASH -= shortage
                adjustment_note = f"Bought into portfolio using {shortage:.2f} CASH."
            else:
                invest_amount = CASH
                for ticker in portfolio.keys():
                    price = prices.loc[current_date, ticker]
                    ticker_value = portfolio[ticker] * price
                    target_weight = ticker_value / mom_value if mom_value > 0 else 1/len(portfolio)
                    qty_to_buy = (target_weight * invest_amount) / price
                    portfolio[ticker] += qty_to_buy
                CASH = 0
                adjustment_note = f"Partial buy using all available CASH."
        else:
            adjustment_note = "No cash adjustment needed."
        
        mom_value = compute_portfolio_value(portfolio, prices, current_date)
        total_aum = mom_value + CASH
        ret = (total_aum - prev_total_aum) / prev_total_aum
        prev_total_aum = total_aum
        
        row = {
            'Date': current_date,
            'Portfolio Value': mom_value,
            'Momentum AUM': mom_value,
            'CASH': CASH,
            'Total AUM': total_aum,
            'Pre-Rebalance Value': pre_rebalance_value,
            'VIX': vix,
            'ConsumerConfidence': cc,
            'IGSpreads': ig,
            'Composite Risk': composite_risk,
            'Bear Prob': p_bear,
            'Target Momentum Alloc': target_alloc,
            'Regime': regime,
            'Adjustment Note': adjustment_note,
            'Return': ret
        }
        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}'] = (portfolio[ticker] * prices.loc[current_date, ticker]) / mom_value if mom_value > 0 else np.nan
            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)
    
    diag_columns = ['Total AUM', 'Momentum AUM', 'CASH', 'VIX', 'ConsumerConfidence',
                    'IGSpreads', 'Composite Risk', 'Bear Prob', 'Target Momentum Alloc', 'Regime',
                    'Adjustment Note', 'Return']
    for prefix in ['qty_', 'notional_', 'weight_', 'rank_', 'metric_', 'trade_']:
        diag_columns.extend([f'{prefix}{ticker}' for ticker in tickers])
    
    result_df = result_df[diag_columns]
    return result_df

###############################################
# 10. Main – Example Usage
###############################################

if __name__ == '__main__':
    # Define 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   # 100 million
    start_date  = pd.to_datetime('2012-01-01')
    end_date    = pd.to_datetime('2025-02-01')
    rebalance_period = 1      # monthly (or adjust as desired)
    rebalance_ratio  = 0.2    # 20% of current momentum AUM rebalanced each period
    lookback_period  = 6      # 3-month lookback
    metric_type      = 'simple'
    
    internal_rebalance_ratios = [0.7, 0.3, 0, 0, -0.3, -0.7]

    
    # Macro parameters (bounds for each indicator)
    macro_params = {
        'VIX': (10, 40),                  
        'ConsumerConfidence': (45, 130),   # Adjusted lower bound to 80
        'IGSpreads': (100, 200)
    }
    
    # File paths.
    price_filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\Codes\Historic Prices.xlsx"
    macro_filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\macro_indic.xlsx"
    
    prices = load_price_data(price_filepath)
    macro = load_macro_data(macro_filepath)
    
    # Set hmm_window to 24 months (or adjust as needed)
    result_df = simulate_strategy_with_hmm(prices, macro, eq_tickers, fi_tickers, alts_tickers,
                                  initial_aum, start_date, end_date,
                                  rebalance_period, rebalance_ratio,
                                  lookback_period, metric_type,
                                  internal_rebalance_ratios,
                                  macro_params,
                                  max_alloc=1.0, min_alloc=0.5,
                                  hmm_window=24)
    
    pd.set_option('display.float_format', lambda x: f'{x:,.2f}')
#     print(result_df)
Editor is loading...
Leave a Comment