149 lines
5.1 KiB
Python
149 lines
5.1 KiB
Python
"""Tests for GLD/GC=F basis data functionality."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
import pytest
|
|
|
|
from app.services.cache import CacheService
|
|
from app.services.data_service import DataService
|
|
|
|
|
|
class _CacheStub(CacheService):
|
|
"""In-memory cache stub for unit testing."""
|
|
|
|
def __init__(self, initial: dict[str, object] | None = None) -> None:
|
|
self._store = dict(initial or {})
|
|
self.write_count = 0
|
|
|
|
async def get_json(self, key: str): # type: ignore[override]
|
|
return self._store.get(key)
|
|
|
|
async def set_json(self, key: str, value: object) -> None: # type: ignore[override]
|
|
self._store[key] = value
|
|
self.write_count += 1
|
|
|
|
async def delete(self, key: str) -> None:
|
|
self._store.pop(key, None)
|
|
|
|
|
|
@pytest.fixture
|
|
def cache_service():
|
|
"""Create a cache service for testing."""
|
|
return _CacheStub()
|
|
|
|
|
|
@pytest.fixture
|
|
def data_service(cache_service):
|
|
"""Create a data service for testing."""
|
|
return DataService(cache_service)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_basis_data_structure(data_service: DataService) -> None:
|
|
"""Test that basis data returns expected structure."""
|
|
basis = await data_service.get_basis_data()
|
|
|
|
# Check all required fields are present
|
|
assert "gld_implied_spot" in basis
|
|
assert "gld_price" in basis
|
|
assert "gld_ounces_per_share" in basis
|
|
assert "gc_f_price" in basis
|
|
assert "gc_f_adjusted" in basis
|
|
assert "contango_estimate" in basis
|
|
assert "basis_bps" in basis
|
|
assert "basis_status" in basis
|
|
assert "basis_label" in basis
|
|
assert "after_hours" in basis
|
|
assert "after_hours_note" in basis
|
|
assert "gld_updated_at" in basis
|
|
assert "gc_f_updated_at" in basis
|
|
assert "gld_source" in basis
|
|
assert "gc_f_source" in basis
|
|
|
|
# Check types
|
|
assert isinstance(basis["gld_implied_spot"], float)
|
|
assert isinstance(basis["gld_price"], float)
|
|
assert isinstance(basis["gld_ounces_per_share"], float)
|
|
assert isinstance(basis["gc_f_price"], float)
|
|
assert isinstance(basis["gc_f_adjusted"], float)
|
|
assert isinstance(basis["contango_estimate"], float)
|
|
assert isinstance(basis["basis_bps"], float)
|
|
assert isinstance(basis["basis_status"], str)
|
|
assert isinstance(basis["basis_label"], str)
|
|
assert isinstance(basis["after_hours"], bool)
|
|
|
|
# Check basis status is valid
|
|
assert basis["basis_status"] in ("green", "yellow", "red")
|
|
assert basis["basis_label"] in ("Normal", "Elevated", "Warning")
|
|
|
|
# Check contango estimate is the expected default
|
|
assert basis["contango_estimate"] == 10.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_basis_calculation_logic(data_service: DataService) -> None:
|
|
"""Test that basis calculation follows the expected formula."""
|
|
basis = await data_service.get_basis_data()
|
|
|
|
# Verify GLD implied spot calculation: GLD_price / ounces_per_share
|
|
if basis["gld_price"] > 0 and basis["gld_ounces_per_share"] > 0:
|
|
expected_implied = basis["gld_price"] / basis["gld_ounces_per_share"]
|
|
# Allow for rounding to 2 decimal places and real-world data variation
|
|
assert abs(basis["gld_implied_spot"] - expected_implied) < 2.0
|
|
|
|
# Verify GC=F adjusted calculation: GC=F_price - contango
|
|
if basis["gc_f_price"] > 0:
|
|
expected_adjusted = basis["gc_f_price"] - basis["contango_estimate"]
|
|
assert abs(basis["gc_f_adjusted"] - expected_adjusted) < 1.0
|
|
|
|
# Verify basis bps calculation: (GLD_implied / GC=F_adjusted - 1) * 10000
|
|
if basis["gc_f_adjusted"] > 0 and basis["gld_implied_spot"] > 0:
|
|
expected_bps = (basis["gld_implied_spot"] / basis["gc_f_adjusted"] - 1) * 10000
|
|
# Allow for rounding to 1 decimal place
|
|
assert abs(basis["basis_bps"] - expected_bps) < 1.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_basis_status_thresholds(data_service: DataService) -> None:
|
|
"""Test that basis status thresholds are applied correctly."""
|
|
basis = await data_service.get_basis_data()
|
|
|
|
abs_basis = abs(basis["basis_bps"])
|
|
|
|
if abs_basis < 25:
|
|
assert basis["basis_status"] == "green"
|
|
assert basis["basis_label"] == "Normal"
|
|
elif abs_basis < 50:
|
|
assert basis["basis_status"] == "yellow"
|
|
assert basis["basis_label"] == "Elevated"
|
|
else:
|
|
assert basis["basis_status"] == "red"
|
|
assert basis["basis_label"] == "Warning"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gc_f_fallback(data_service: DataService) -> None:
|
|
"""Test that GC=F fallback works when live data unavailable."""
|
|
gc_f = await data_service.get_gc_futures()
|
|
|
|
assert "symbol" in gc_f
|
|
assert gc_f["symbol"] == "GC=F"
|
|
assert "price" in gc_f
|
|
assert isinstance(gc_f["price"], float)
|
|
assert gc_f["price"] > 0 # Should have a positive price (real or fallback)
|
|
assert gc_f["quote_unit"] == "ozt"
|
|
assert "source" in gc_f
|
|
|
|
|
|
def test_gld_ounces_per_share_current() -> None:
|
|
"""Test GLD ounces per share calculation for current date."""
|
|
from app.domain.instruments import gld_ounces_per_share
|
|
|
|
ounces = gld_ounces_per_share(date.today())
|
|
|
|
# Should be around 0.0919 for 2026 (8.1% decay from 0.10)
|
|
assert 0.090 < float(ounces) < 0.094
|
|
assert ounces > 0
|