diff --git a/src/fxhnt/application/crypto_momentum_research.py b/src/fxhnt/application/crypto_momentum_research.py index 41b8999..ed42f0c 100644 --- a/src/fxhnt/application/crypto_momentum_research.py +++ b/src/fxhnt/application/crypto_momentum_research.py @@ -20,15 +20,10 @@ from fxhnt.domain.gauntlet import deflated_sharpe, per_period_sharpe _ANN = np.sqrt(365.0) -def load_crypto_panel(crypto_dir: str) -> tuple[list[int], list[str], np.ndarray, np.ndarray]: - """Load a survivorship-free crypto close panel from a dir of .npz (keys: day, close).""" - data: dict[str, dict[int, float]] = {} - for f in sorted(glob.glob(os.path.join(crypto_dir, "*.npz"))): - z = np.load(f) # numeric-only arrays (our own surfer-fetch files); no allow_pickle needed - if "close" in z and "day" in z: - data[os.path.basename(f)[:-4]] = dict(zip(z["day"].tolist(), z["close"].tolist(), strict=True)) +def _panel(data: dict[str, dict[int, float]]) -> tuple[list[int], list[str], np.ndarray, np.ndarray]: + """Aligned panel from {symbol: {epoch_day: close}} → (days, syms, close[d×s], ret[d×s]).""" syms = sorted(data) - days = sorted(set().union(*(set(d) for d in data.values()))) + days = sorted(set().union(*(set(d) for d in data.values()))) if data else [] di = {d: i for i, d in enumerate(days)} close = np.full((len(days), len(syms)), np.nan) for j, s in enumerate(syms): @@ -36,10 +31,21 @@ def load_crypto_panel(crypto_dir: str) -> tuple[list[int], list[str], np.ndarray if c and c > 0: close[di[d], j] = c ret = np.full_like(close, np.nan) - ret[1:] = close[1:] / close[:-1] - 1.0 + if len(days) > 1: + ret[1:] = close[1:] / close[:-1] - 1.0 return days, syms, close, ret +def load_crypto_panel(crypto_dir: str) -> tuple[list[int], list[str], np.ndarray, np.ndarray]: + """Load a survivorship-free crypto close panel from a dir of .npz (keys: day, close).""" + data: dict[str, dict[int, float]] = {} + for f in sorted(glob.glob(os.path.join(crypto_dir, "*.npz"))): + z = np.load(f) # numeric-only arrays (our own surfer-fetch files); no allow_pickle needed + if "close" in z and "day" in z: + data[os.path.basename(f)[:-4]] = dict(zip(z["day"].tolist(), z["close"].tolist(), strict=True)) + return _panel(data) + + def xs_momentum_book(close: np.ndarray, ret: np.ndarray, lookback: int, *, cost_bps: float = 8.0, long_only: bool = False, min_names: int = 10) -> np.ndarray: """Daily cross-sectional momentum book: rank-demeaned weights on past-`lookback` return, dollar-neutral diff --git a/tests/unit/test_crypto_panel_builder.py b/tests/unit/test_crypto_panel_builder.py new file mode 100644 index 0000000..a2cfaf0 --- /dev/null +++ b/tests/unit/test_crypto_panel_builder.py @@ -0,0 +1,22 @@ +"""The panel builder turns {sym: {day: close}} into aligned (days, syms, close, ret) — shared by the +raw-file and warehouse sources so they produce identical panels.""" +from __future__ import annotations + +import numpy as np + +from fxhnt.application.crypto_momentum_research import _panel + + +def test_panel_aligns_and_computes_returns() -> None: + data = {"BTCUSDT": {19000: 100.0, 19001: 110.0}, "ETHUSDT": {19001: 50.0, 19002: 55.0}} + days, syms, close, ret = _panel(data) + assert days == [19000, 19001, 19002] + assert syms == ["BTCUSDT", "ETHUSDT"] + assert np.isnan(close[0, 1]) and close[1, 0] == 110.0 and close[2, 1] == 55.0 + assert abs(ret[1, 0] - 0.10) < 1e-12 # 110/100 - 1 + assert np.isnan(ret[0, 0]) + + +def test_panel_empty() -> None: + days, syms, close, ret = _panel({}) + assert days == [] and syms == [] and close.shape == (0, 0)