tests/test_analytics.py

import sys
import os
import copy

# Bulletproof pathing: Force Python to look in both the current folder AND the parent folder
_this_dir = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _this_dir)
sys.path.insert(0, os.path.dirname(_this_dir))

import pandas as pd
import numpy as np

# Import the functions we want to test from your engine
from utils.metrics import (
    portfolio_gross_metrics, 
    israelsen_sharpe, 
    liquidity_score, 
)
from backtest import expanding_window_backtest
from core_types import LotManager
from config import DEFAULT_CONFIG

# ─────────────────────────────────────────────
# 1. GROSS EXPOSURE TESTS
# ─────────────────────────────────────────────
def test_portfolio_gross_metrics():
    """Tests if the engine correctly calculates long, short, and gross leverage."""
    fake_weights = pd.Series({'AAPL': 0.60, 'TSLA': -0.20, 'CASH': 0.10})
    
    pgm = portfolio_gross_metrics(fake_weights)
    gross_lev, long_e, short_e = pgm["gross_lev"], pgm["long_e"], pgm["short_e"]
    
    # portfolio_gross_metrics intentionally drops CASH before computing leverage:
    # - gross leverage = |risky longs| + |risky shorts| = |0.60| + |-0.20| = 0.80
    # - long exposure  = risky long only                = 0.60
    # - short exposure = risky short only               = -0.20
    # (CASH does not contribute to gross leverage or directional exposure)
    assert np.isclose(gross_lev, 0.80)
    assert np.isclose(long_e, 0.60)
    assert np.isclose(short_e, -0.20)

# ─────────────────────────────────────────────
# 2. SHARPE RATIO TESTS
# ─────────────────────────────────────────────
def test_israelsen_sharpe_positive():
    """Tests the modified Sharpe ratio with positive returns."""
    score = israelsen_sharpe(0.10, 0.20)
    assert np.isclose(score, 0.5)

def test_israelsen_sharpe_negative():
    """
    Tests Israelsen's specific fix: multiplying by volatility when returns are negative.
    Classic Sharpe divides, which wrongly makes highly volatile losers look better.
    """
    score = israelsen_sharpe(-0.10, 0.20)
    # -0.10 / (0.20 ^ -1) = -0.10 * 0.20 = -0.02
    assert np.isclose(score, -0.02)

def test_israelsen_sharpe_zero_vol():
    """Ensures the math doesn't crash (Divide by Zero) on cash-equivalent assets."""
    score = israelsen_sharpe(0.05, 0.0)
    assert score == 0.0

# ─────────────────────────────────────────────
# 3. LIQUIDITY TESTS
# ─────────────────────────────────────────────
def test_liquidity_score():
    """
    Tests if the liquidity score correctly calculates the 
    weighted-average half-spread (normalized by gross weight).
    """
    fake_weights = pd.Series({'AAPL': 0.60, 'TSLA': -0.20})
    spread_map = {'AAPL': 0.0002, 'TSLA': 0.0005}
    
    score = liquidity_score(fake_weights, spread_map)
    
    # Gross weight = 0.80
    # AAPL contribution = 0.60 * 0.0002 = 0.00012
    # TSLA contribution = |-0.20| * 0.0005 = 0.00010
    # Total = 0.00022. Normalized by gross (0.80) = 0.000275
    assert np.isclose(score, 0.000275)

# ─────────────────────────────────────────────
# 4. TAX LIQUIDATION & BACKTEST TESTS
# ─────────────────────────────────────────────
def test_lot_manager_tax_liquidation():
    """
    Tests the LotManager's HIFO tax liquidation logic, explicitly verifying
    that it correctly splits Long-Term and Short-Term capital gains based on lot age.
    """
    lm = LotManager()
    
    # Current date for our simulated sale
    sell_date = pd.Timestamp("2024-01-01")
    
    # Lot 1: Bought 2 years ago (Long-Term). Price: $100. Shares: 10
    lm.add_lot("AAPL", "2022-01-01", 100.0, 10.0)
    
    # Lot 2: Bought 6 months ago (Short-Term). Price: $150. Shares: 10
    lm.add_lot("AAPL", "2023-07-01", 150.0, 10.0)
    
    # Lot 3: Bought 2 months ago (Short-Term). Price: $180. Shares: 10
    lm.add_lot("AAPL", "2023-11-01", 180.0, 10.0)
    
    # Current Price is $200. We want to sell 15 shares.
    # Under HIFO (Highest In, First Out), we should sell:
    # 1. 10 shares of Lot 3 (Cost $180) -> Short-Term
    # 2. 5 shares of Lot 2 (Cost $150) -> Short-Term
    
    total_gain, lt_gain, st_gain = lm.sell_shares_with_tax(
        "AAPL", shares_to_sell=15.0, current_price=200.0, 
        current_date=sell_date, lt_days=366, method='hifo'
    )
    
    # Calculations:
    # Lot 3 Gain: 10 shares * ($200 - $180) = $200 (Short-Term)
    # Lot 2 Gain: 5 shares * ($200 - $150) = $250 (Short-Term)
    # Total Gain = $450. Long-Term Gain = $0.
    
    assert np.isclose(total_gain, 450.0)
    assert np.isclose(st_gain, 450.0)
    assert np.isclose(lt_gain, 0.0)
    
    # Now sell remaining 15 shares.
    # Remaining: 5 shares of Lot 2 (ST), 10 shares of Lot 1 (LT)
    # HIFO will sell Lot 2 first ($150 cost), then Lot 1 ($100 cost).
    
    total_gain_2, lt_gain_2, st_gain_2 = lm.sell_shares_with_tax(
        "AAPL", shares_to_sell=15.0, current_price=200.0, 
        current_date=sell_date, lt_days=366, method='hifo'
    )
    
    # Lot 2 Gain: 5 shares * ($200 - $150) = $250 (Short-Term)
    # Lot 1 Gain: 10 shares * ($200 - $100) = $1000 (Long-Term)
    assert np.isclose(total_gain_2, 1250.0)
    assert np.isclose(st_gain_2, 250.0)
    assert np.isclose(lt_gain_2, 1000.0)


def test_expanding_window_tax_liquidation():
    """
    Tests if the expanding window backtest correctly utilizes the LotManager
    to liquidate lots and apply tax drag when rebalancing highly appreciated assets.
    """
    rng = np.random.default_rng(42)
    dates = pd.date_range("2020-01-01", periods=500, freq="B")

    # Create a synthetic massive bull market (assets go up constantly)
    # This guarantees massive embedded gains across rebalance boundaries.
    rets_a = rng.normal(0.002, 0.01, size=500)
    rets_b = rng.normal(0.001, 0.01, size=500)
    returns_df = pd.DataFrame({'A': rets_a, 'B': rets_b}, index=dates)
    spy_rets = pd.Series(rng.normal(0.001, 0.01, size=500), index=dates)
    spread_map = {'A': 0.0, 'B': 0.0}

    # Note: Use deepcopy to prevent nested dictionaries (sector_map, benchmarks) 
    # from permanently mutating the global DEFAULT_CONFIG during the test run.
    base_cfg = copy.deepcopy(DEFAULT_CONFIG)
    base_cfg.update({
        "risk_free_rate": 0.0,
        "transaction_cost": 0.0, # Zero friction to perfectly isolate the tax effect
        "single_asset_min": 0.0,
        "single_asset_max": 1.0,
        "sector_limit": 1.0,
        "sector_map": {"A": "Tech", "B": "Tech"}, # Note: Valid sector mapping string instead of integers
        "cvar_enabled": False,
        "garch_enabled": False,
        "max_turnover": 2.0
    })

    # Run 1: Tax Disabled
    cfg_no_tax = copy.deepcopy(base_cfg)
    cfg_no_tax['tax_enabled'] = False

    eq_no_tax, _ = expanding_window_backtest(
        returns_df, spy_rets, capital=100000.0, rfr=0.0, cfg=cfg_no_tax,
        model=1, allocation_engine=1, spread_map=spread_map,
        initial_train_days=100, rebalance_freq=50
    )

    # Run 2: Tax Enabled (Aggressive ST rates)
    cfg_tax = copy.deepcopy(base_cfg)
    cfg_tax['tax_enabled'] = True
    cfg_tax['tax_rate_lt'] = 0.20
    cfg_tax['tax_rate_st'] = 0.40
    cfg_tax['lt_days'] = 366

    eq_tax, _ = expanding_window_backtest(
        returns_df, spy_rets, capital=100000.0, rfr=0.0, cfg=cfg_tax,
        model=1, allocation_engine=1, spread_map=spread_map,
        initial_train_days=100, rebalance_freq=50
    )

    # Since rebalancing in a bull market forces selling winners to maintain risk targets,
    # the tax-enabled portfolio must mathematically suffer tax drag and end up lower.
    if eq_tax is not None and eq_no_tax is not None:
        assert eq_tax.iloc[-1] < eq_no_tax.iloc[-1]

def test_lot_manager_thread_safety():
    """Verifies that LotManager safely handles concurrent add and sell operations."""
    import threading
    lm = LotManager()
    
    def worker(worker_id):
        for i in range(100):
            lm.add_lot("AAPL", "2024-01-01", 100.0, 1.0)
            if i % 2 == 0:
                lm.sell_shares_with_tax("AAPL", 0.5, 150.0, method="fifo")
                
    threads = []
    for i in range(10):
        t = threading.Thread(target=worker, args=(i,))
        threads.append(t)
        t.start()
        
    for t in threads:
        t.join()
        
    # Total shares added: 10 * 100 * 1.0 = 1000
    # Total shares sold: 10 * 50 * 0.5 = 250
    # Remaining: 750
    remaining_shares = sum(lot.shares for lot in lm.lots["AAPL"])
    assert np.isclose(remaining_shares, 750.0)