demo strategies

crypto_eval/strategies.py

@@ -0,0 +1,58 @@
+"""Demo strategies for the crypto-quant-eval harness.
+
+Each strategy is a function (bars, funding) -> positions: pd.Series of
+units of base asset indexed by bar start time. Strategies MUST only use
+information available at decision time: a bar starting at t is closed at
+t+freq, so signals at t use bars shifted by 1 (the closed bar is at t-1).
+The `shift_bars` helper enforces this.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+
+def shift_bars(bars: pd.DataFrame, n: int = 1) -> pd.DataFrame:
+    """Return bars as known at decision time t: last closed bar is t-1."""
+    return bars.shift(n)
+
+
+def funding_carry(bars: pd.DataFrame, funding: pd.DataFrame,
+                  threshold: float = 0.0001,
+                  max_leverage: float = 1.0) -> pd.Series:
+    """Short the perp when funding is persistently positive (carry trade).
+
+    Signal: rolling 3-tick mean funding. If mean funding > threshold, hold
+    short notional -max_leverage; if < -threshold, hold long; else flat.
+    """
+    # funding at tick t is announced at t; a bar starting at t can use
+    # the tick at t-1 (announced) but not the one accruing over [t, t+8h)
+    # accept either a Series or a one-column DataFrame with funding_rate
+    if isinstance(funding, pd.DataFrame):
+        funding = funding["funding_rate"]
+    f = funding.shift(1)
+    signal = f.rolling(3, min_periods=1).mean()
+    pos = pd.Series(0.0, index=bars.index)
+    # expand tick-level signal to bar-level, decision at bar start t uses
+    # the last tick announced strictly before t
+    tick_signal = signal.reindex(bars.index, method="ffill")
+    pos[tick_signal > threshold] = -max_leverage
+    pos[tick_signal < -threshold] = max_leverage
+    return pos
+
+
+def ma_crossover(bars: pd.DataFrame, funding: pd.DataFrame,
+                 fast: int = 24, slow: int = 168) -> pd.Series:
+    """Classic fast/slow MA crossover on 1h close (7d slow)."""
+    close = shift_bars(bars)["close"]
+    fast_ma = close.rolling(fast).mean()
+    slow_ma = close.rolling(slow).mean()
+    pos = pd.Series(np.where(fast_ma > slow_ma, 1.0, -1.0), index=bars.index)
+    return pos
+
+
+STRATEGIES = {
+    "funding_carry": funding_carry,
+    "ma_crossover": ma_crossover,
+}