ftdt-quant-lab/strategies/cross_sectional_momentum.py

"""
Cross-Sectional Momentum strategy for Hyperliquid assets.

Ranks all available assets by recent return (lookback window). Goes long
the top-N performers and short the bottom-N. Rebalances periodically.
This captures the cross-sectional momentum premium documented extensively
in academic literature (Jegadeesh & Titman 1993, Moskowitz 2012).

Assets: BTC, ETH, SOL, ARB, OP, HYPE, HFUN, PURR, VVV, etc.
Interval: Daily rebalancing with 1m/1h/4h lookbacks available.
Risk: Equal-weight or risk-parity across long/short baskets.
Costs: Hyperliquid perp fee schedule with tier-appropriate rates.
      Post-only limit orders (maker fees) to reduce costs.
"""
from __future__ import annotations

import logging
from typing import Optional

import numpy as np
import pandas as pd

logger = logging.getLogger(__name__)

# Hyperliquid universe — perps with sufficient liquidity
HL_UNIVERSE = [
    "BTC", "ETH", "SOL", "ARB", "OP", "HYPE", "HFUN", "PURR",
    "VVV", "LINK", "AVAX", "SUI", "DOGE", "XRP", "ADA", "DOT",
    "APT", "ATOM", "NEAR", "SEI",
]

HIGH_LIQUIDITY = ["BTC", "ETH", "SOL", "HYPE", "ARB", "OP"]
MEDIUM_LIQUIDITY = HIGH_LIQUIDITY + ["LINK", "AVAX", "SUI", "DOGE", "XRP"]
LOW_LIQUIDITY = HL_UNIVERSE


class CrossSectionalMomentum:
    """Long-short cross-sectional momentum on Hyperliquid perps.

    Ranks assets by momentum score, goes long top-N, short bottom-N.
    Rebalances every `rebalance_period` bars.

    Attributes.
    ----------
    lookback: int — bars to compute momentum over
    top_n: int — number of longs
    bottom_n: int — number of shorts
    rebalance_period: int — bars between rebalances
    risk_parity: bool — size positions by inverse volatility
    vol_target: float — annualized vol target (0 = disabled)
    filter_threshold: float — min abs return to include (avoids noise)
    """

    def __init__(
        self,
        lookback: int = 20,
        top_n: int = 3,
        bottom_n: int = 3,
        rebalance_period: int = 1,
        risk_parity: bool = True,
        vol_target: float = 0.25,
        filter_threshold: float = 0.002,
    ):
        self.lookback = lookback
        self.top_n = top_n
        self.bottom_n = bottom_n
        self.rebalance_period = rebalance_period
        self.risk_parity = risk_parity
        self.vol_target = vol_target
        self.filter_threshold = filter_threshold

    def compute_signals(
        self,
        prices: dict[str, pd.Series],
    ) -> dict[str, float]:
        """Compute cross-sectional momentum weights.

        Args:
            prices: dict of coin → pd.Series of close prices (aligned by index)

        Returns:
            dict of coin → weight (-1 to +1). Positive = long, negative = short.
        """
        if len(prices) < self.top_n + self.bottom_n:
            return {}

        momentum_scores = {}
        returns = {}

        for coin, px in prices.items():
            if len(px) < self.lookback + 1:
                continue

            pct_ret = (px.iloc[-1] / px.iloc[-self.lookback] - 1)
            if abs(pct_ret) < self.filter_threshold:
                continue

            returns[coin] = px
            momentum_scores[coin] = pct_ret

        if len(momentum_scores) < self.top_n + self.bottom_n:
            return {}

        sorted_coins = sorted(momentum_scores, key=momentum_scores.get, reverse=True)

        longs = sorted_coins[:self.top_n]
        shorts = sorted_coins[-self.bottom_n:]

        weights: dict[str, float] = {}

        if self.risk_parity:
            long_wt = self._risk_parity_weights({c: returns[c] for c in longs + shorts}, longs, shorts)
            weights.update(long_wt)
        else:
            for c in longs:
                weights[c] = 1.0 / self.top_n
            for c in shorts:
                weights[c] = -1.0 / self.bottom_n

        if self.vol_target > 0:
            weights = self._scale_to_vol_target(weights, returns)

        return weights

    def _risk_parity_weights(
        self,
        returns: dict[str, pd.Series],
        longs: list[str],
        shorts: list[str],
    ) -> dict[str, float]:
        """Compute risk-parity weights: positions sized by 1/volatility."""
        weights = {}
        vols = {}
        for coin, px in returns.items():
            ret_series = px.pct_change().dropna()
            vol = ret_series.std() * np.sqrt(365 * 24)
            vols[coin] = max(vol, 0.05)

        # Long basket
        long_inv_vols = {c: 1.0 / vols[c] for c in longs}
        long_sum = sum(long_inv_vols.values())
        for c in longs:
            weights[c] = long_inv_vols[c] / long_sum if long_sum > 0 else 1.0 / len(longs)

        # Short basket
        short_inv_vols = {c: 1.0 / vols[c] for c in shorts}
        short_sum = sum(short_inv_vols.values())
        for c in shorts:
            weights[c] = -(short_inv_vols[c] / short_sum) if short_sum > 0 else -(1.0 / len(shorts))

        return weights

    def _scale_to_vol_target(
        self,
        weights: dict[str, float],
        returns: dict[str, pd.Series],
    ) -> dict[str, float]:
        """Scale portfolio to target annualized volatility."""
        if not weights:
            return weights

        combined_ret = None
        for coin, wt in weights.items():
            if coin not in returns:
                continue
            px = returns[coin]
            ret = px.pct_change().dropna()
            if combined_ret is None:
                combined_ret = ret * wt
            else:
                combined_ret = combined_ret + ret * wt

        if combined_ret is None or len(combined_ret) < 2:
            return weights

        portfolio_vol = combined_ret.std() * np.sqrt(365 * 24)
        if portfolio_vol <= 0:
            return weights

        scale = self.vol_target / portfolio_vol
        scale = min(scale, 2.0)  # Cap leverage at 2x

        return {c: w * scale for c, w in weights.items()}

    def generate_entries_exits(
        self,
        prices: dict[str, pd.DataFrame],
        coin: str,
    ) -> tuple[pd.Series, pd.Series]:
        """Generate entry/exit signals suitable for VBT integration.

        Returns (entries, exits) boolean Series indexed by time.
        """
        close_prices = {c: df["close"] for c, df in prices.items() if "close" in df.columns}
        if coin not in close_prices:
            return pd.Series(dtype=bool), pd.Series(dtype=bool)

        main_close = close_prices[coin]
        entries = pd.Series(False, index=main_close.index)
        exits = pd.Series(False, index=main_close.index)

        for i in range(self.lookback, len(main_close.index)):
            if (i - self.lookback) % self.rebalance_period != 0:
                continue

            slice_prices = {
                c: px.iloc[:i + 1]
                for c, px in close_prices.items()
                if len(px) > i
            }
            weights = self.compute_signals(slice_prices)

            if coin in weights and weights[coin] != 0:
                wt = weights[coin]
                # Check if position changed direction
                prev_wt = self._get_prev_weight(coin, close_prices, i - self.rebalance_period, self.lookback)
                if wt > 0 and prev_wt <= 0:
                    entries.iloc[i] = True
                elif wt < 0 and prev_wt >= 0:
                    entries.iloc[i] = True
                elif abs(prev_wt - wt) < 0.01:
                    # No significant weight change — exit
                    exits.iloc[i] = True

        return entries, exits

    def _get_prev_weight(
        self,
        coin: str,
        prices: dict[str, pd.Series],
        idx: int,
        lookback: int,
    ) -> float:
        """Look up previous position weight."""
        if idx < lookback:
            return 0.0
        slice_prices = {
            c: px.iloc[:idx + 1]
            for c, px in prices.items()
            if len(px) > idx
        }
        weights = self.compute_signals(slice_prices)
        return weights.get(coin, 0.0)