tests/test_data.py

import sys
import os
from sqlalchemy import create_engine

_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 data


def test_fetch_risk_free_rate_falls_back_when_cache_missing(monkeypatch):
    """Missing SQLite tables should not crash engine startup."""
    mem_engine = create_engine("sqlite:///:memory:")
    monkeypatch.setattr(data, "_get_db_engine", lambda: mem_engine)

    assert data.fetch_risk_free_rate(default_rate=0.0375) == 0.0375


def test_fetch_risk_free_rate_reads_latest_cached_value(monkeypatch):
    """Risk-free rate should come from the latest cached benchmark quote when available."""
    mem_engine = create_engine("sqlite:///:memory:")
    with mem_engine.begin() as conn:
        conn.exec_driver_sql(
            "CREATE TABLE daily_prices (ticker TEXT, date TEXT, close_price REAL, UNIQUE(ticker, date))"
        )
        conn.exec_driver_sql(
            "INSERT INTO daily_prices VALUES (?, ?, ?)",
            ("^TNX", "2024-01-01", 3.50)
        )
        conn.exec_driver_sql(
            "INSERT INTO daily_prices VALUES (?, ?, ?)",
            ("^TNX", "2024-01-02", 4.25)
        )
        
    monkeypatch.setattr(data, "_get_db_engine", lambda: mem_engine)

    assert data.fetch_risk_free_rate("^TNX", default_rate=0.01) == 0.0425


def test_clean_price_series_stale_and_weekend_data():
    import pandas as pd
    import numpy as np
    
    # Create dates spanning a weekend (Fri, Sat, Sun, Mon, Tue, Wed, Thu, Fri)
    dates = pd.date_range("2024-01-05", periods=8, freq="D") # 2024-01-05 is a Friday
    # Friday: 100
    # Sat/Sun: 100 (weekend data)
    # Mon: 100
    # Tue: 100
    # Wed: 100 (stale data -> flat for >3 days)
    # Thu: 110
    # Fri: 110
    prices = pd.Series([100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 110.0, 110.0], index=dates, name="TEST")
    
    cleaned = data.clean_price_series(prices)
    
    # Weekends (Sat/Sun) should be dropped
    assert len(cleaned) == 6
    # Flat prices should be interpolated
    assert not cleaned.isna().any()

def test_ewm_incremental_initialization():
    import pandas as pd
    import numpy as np
    
    np.random.seed(42)
    rets = np.random.normal(0, 0.01, 100)
    
    # pandas ewm adjust=False logic
    pandas_ewm = pd.Series(rets).ewm(halflife=21, adjust=False).mean()
    
    # Incremental logic used in backtest.py
    alpha = 1 - np.exp(-np.log(2)/21)
    ewm_mean = np.zeros(100)
    ewm_mean[0] = rets[0]
    for i in range(1, 100):
        ewm_mean[i] = alpha * rets[i] + (1 - alpha) * ewm_mean[i-1]
        
    np.testing.assert_allclose(pandas_ewm.values, ewm_mean, atol=1e-8)