Spaces:
Sleeping
Sleeping
File size: 2,698 Bytes
208fbf8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | 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)
|