refactor(crypto): extract _panel builder (shared by raw + warehouse sources)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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

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)