119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
"""Unit tests for the cross-sectional activation momentum ranking."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, timedelta
|
|
|
|
import pytest
|
|
|
|
from app.models.ohlcv import OHLCVRecord
|
|
from app.models.ticker import Ticker
|
|
from app.services import momentum_service as ms
|
|
|
|
|
|
@pytest.fixture
|
|
async def session():
|
|
from tests.conftest import _test_session_factory
|
|
|
|
async with _test_session_factory() as s:
|
|
yield s
|
|
|
|
|
|
async def _seed(session, symbol: str, rate: float, n: int = 280) -> None:
|
|
t = Ticker(symbol=symbol)
|
|
session.add(t)
|
|
await session.flush()
|
|
base = date(2024, 1, 1)
|
|
for i in range(n):
|
|
close = 100.0 * (rate ** i)
|
|
session.add(OHLCVRecord(
|
|
ticker_id=t.id,
|
|
date=base + timedelta(days=i),
|
|
open=close, high=close, low=close, close=close,
|
|
volume=1_000_000,
|
|
))
|
|
await session.commit()
|
|
|
|
|
|
async def _seed_closes(session, symbol: str, closes: list[float]) -> None:
|
|
t = Ticker(symbol=symbol)
|
|
session.add(t)
|
|
await session.flush()
|
|
base = date(2024, 1, 1)
|
|
for i, close in enumerate(closes):
|
|
session.add(OHLCVRecord(
|
|
ticker_id=t.id,
|
|
date=base + timedelta(days=i),
|
|
open=close, high=close, low=close, close=close,
|
|
volume=1_000_000,
|
|
))
|
|
await session.commit()
|
|
|
|
|
|
def test_compute_momentum_insufficient_history():
|
|
assert ms.compute_12_1_momentum([100.0] * 100) is None
|
|
|
|
|
|
def test_compute_momentum_value():
|
|
closes = [100.0 * (1.01 ** i) for i in range(300)]
|
|
m = ms.compute_12_1_momentum(closes)
|
|
# 12-1 momentum skips the last month: close[-22] / close[-253] - 1
|
|
assert m == closes[-22] / closes[-253] - 1.0
|
|
assert m > 0
|
|
|
|
|
|
async def test_ranks_universe_into_raw_percentiles_when_benchmark_missing(session, monkeypatch):
|
|
async def no_benchmark(_db):
|
|
return {}
|
|
|
|
monkeypatch.setattr(ms, "_load_activation_benchmark", no_benchmark)
|
|
await _seed(session, "HIGH", rate=1.010) # strong uptrend → top momentum
|
|
await _seed(session, "MID", rate=1.002)
|
|
await _seed(session, "LOW", rate=0.999) # declining → bottom momentum
|
|
|
|
pct = await ms.compute_momentum_percentiles(session)
|
|
assert pct["HIGH"] == 100.0
|
|
assert pct["MID"] == 50.0
|
|
assert pct["LOW"] == 0.0
|
|
|
|
|
|
async def test_ranks_universe_into_residual_percentiles_when_benchmark_available(session, monkeypatch):
|
|
base = date(2024, 1, 1)
|
|
n = 280
|
|
benchmark = {base + timedelta(days=i): 100.0 * (1.001 ** i) for i in range(n)}
|
|
|
|
async def with_benchmark(_db):
|
|
return benchmark
|
|
|
|
monkeypatch.setattr(ms, "_load_activation_benchmark", with_benchmark)
|
|
market = [benchmark[base + timedelta(days=i)] for i in range(n)]
|
|
await _seed_closes(session, "DRIFT", [market[i] * (1.0008 ** i) for i in range(n)])
|
|
await _seed_closes(session, "BETA", market)
|
|
await _seed_closes(session, "LAG", [market[i] * (0.9992 ** i) for i in range(n)])
|
|
|
|
pct = await ms.compute_momentum_percentiles(session)
|
|
assert pct["DRIFT"] == 100.0
|
|
assert pct["BETA"] == 50.0
|
|
assert pct["LAG"] == 0.0
|
|
|
|
|
|
async def test_short_history_ticker_is_unranked(session, monkeypatch):
|
|
async def no_benchmark(_db):
|
|
return {}
|
|
|
|
monkeypatch.setattr(ms, "_load_activation_benchmark", no_benchmark)
|
|
await _seed(session, "LONG", rate=1.005)
|
|
await _seed(session, "SHORTHX", rate=1.005, n=100) # < 1y → no momentum
|
|
|
|
pct = await ms.compute_momentum_percentiles(session)
|
|
assert "LONG" in pct
|
|
assert "SHORTHX" not in pct
|
|
|
|
|
|
async def test_empty_universe_returns_empty(session, monkeypatch):
|
|
async def no_benchmark(_db):
|
|
return {}
|
|
|
|
monkeypatch.setattr(ms, "_load_activation_benchmark", no_benchmark)
|
|
assert await ms.compute_momentum_percentiles(session) == {}
|