Basis Arb Agent Skill

from freqtrade.strategy import IStrategy, informative
from datetime import datetime
import pandas as pd
import talib.abstract as ta


def perp_to_spot(pair: str) -> str:
    """`BTC/USDC:USDC` → `BTC/USDC`. HL spot lives at the un-suffixed pair."""
    return pair.split(":")[0] if ":" in pair else pair


class BasisFlippingStrategy(IStrategy):
    minimal_roi = {"0": 100.0}
    stoploss = -0.04
    trailing_stop = False
    timeframe = "1h"
    process_only_new_candles = True
    startup_candle_count = 30
    can_short = False

    def informative_pairs(self):
        # Tell Freqtrade we need spot-side OHLCV for every perp in the
        # whitelist. The base pair list is set by the user; we mirror
        # each entry to its spot equivalent.
        pairs = self.dp.current_whitelist()
        return [(perp_to_spot(p), self.timeframe) for p in pairs if ":" in p]

    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        perp_pair = metadata["pair"]
        spot_pair = perp_to_spot(perp_pair)

        # Funding rate via the dedicated candle type.
        try:
            funding = self.dp.get_pair_dataframe(
                pair=perp_pair,
                timeframe="1h",
                candle_type="funding_rate",
            )
        except Exception:
            funding = pd.DataFrame()
        if not funding.empty and "open" in funding.columns:
            f = funding[["date", "open"]].rename(columns={"open": "funding_rate"}).copy()
            dataframe = dataframe.merge(f, on="date", how="left")
            dataframe["funding_rate"] = dataframe["funding_rate"].ffill().fillna(0.0)
            dataframe["funding_apr"] = dataframe["funding_rate"] * 24 * 365
        else:
            dataframe["funding_rate"] = 0.0
            dataframe["funding_apr"] = 0.0

        # Spot OHLCV — declared via `informative_pairs`.
        try:
            spot = self.dp.get_pair_dataframe(pair=spot_pair, timeframe=self.timeframe)
        except Exception:
            spot = pd.DataFrame()
        # Initialise required columns up front so the entry guard can
        # read them even when the spot leg isn't available on this
        # dataset. Hyperliquid's backtest cache doesn't always expose
        # the matching spot pair via informative_pairs — without these
        # default columns the strategy crashes with KeyError on the
        # first candle.
        dataframe["spot_close"] = float("nan")
        dataframe["basis"] = 0.0
        dataframe["basis_apr"] = 0.0
        if not spot.empty and "close" in spot.columns:
            s = spot[["date", "close"]].rename(columns={"close": "spot_close"}).copy()
            dataframe = dataframe.drop(columns=["spot_close"])
            dataframe = dataframe.merge(s, on="date", how="left")
            dataframe["spot_close"] = dataframe["spot_close"].ffill()
            # Instantaneous basis (perp - spot) / spot. Annualise as
            # basis_apr ≈ basis * (8h_settlement_periods/year) ≈ basis * 1095
            # (3 settlements/day × 365). HL's actual basis convergence
            # path is messier but this is the standard back-of-envelope.
            dataframe["basis"] = (
                dataframe["close"] - dataframe["spot_close"]
            ) / dataframe["spot_close"]
            dataframe["basis_apr"] = dataframe["basis"] * 1095

        dataframe["atr_24"] = ta.ATR(dataframe, timeperiod=24)
        return dataframe

    def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        # Long perp when:
        #   1. Basis flipped negative (perp at discount to spot)
        #   2. Funding is also negative (shorts paying — same crowd)
        #   3. Spot isn't crashing (close > 24h SMA proxy)
        # Only fire when the spot leg is actually available — without
        # spot data the basis is meaningless and we'd be entering on a
        # zero-filled signal.
        sma_spot = dataframe["spot_close"].rolling(24).mean()
        spot_available = dataframe["spot_close"].notna()
        dataframe.loc[
            spot_available
            & (dataframe["basis_apr"] < -0.05)
            & (dataframe["funding_apr"] < 0.0)
            & (dataframe["spot_close"] > sma_spot)
            & (dataframe["volume"] > 0),
            "enter_long",
        ] = 1
        return dataframe

    def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        # Exit when basis converges back to neutral (≥ 0) — the structural
        # pressure has been worked off.
        dataframe.loc[(dataframe["basis_apr"] >= 0.0), "exit_long"] = 1
        return dataframe

    def custom_exit(self, pair: str, trade, current_time: datetime,
                    current_rate: float, current_profit: float, **kwargs):
        # Basis convergence is slow but persistent. 48h is the patience floor.
        elapsed_h = (current_time - trade.open_date_utc).total_seconds() / 3600.0
        if elapsed_h >= 48:
            return "timeout_48h"
        return None

{
  "exchange": { "name": "hyperliquid", "pair_whitelist": ["BTC/USDC:USDC"] },
  "stake_currency": "USDC",
  "stake_amount": 100,
  "timeframe": "1h",
  "max_open_trades": 1,
  "stoploss": -0.04,
  "minimal_roi": { "0": 100.0 },
  "trading_mode": "futures",
  "margin_mode": "cross",
  "entry_pricing": { "price_side": "same" },
  "exit_pricing": { "price_side": "same" },
  "pairlists": [{ "method": "StaticPairList" }]
}