Strategy10_PairTrading

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
"""
Strategy 10: Statistical Pair Trading — BTC/ETH (Halal Long-Only Adaptation)
=============================================================================
Type        : Statistical Arbitrage / Mean Reversion (SPOT ONLY — Halal)
Timeframe   : 4h
Author      : Crypto Halal Algo Trading Suite

DESCRIPTION
-----------
Classical statistical pair trading typically involves shorting the overvalued
asset and going long the undervalued one simultaneously. In a Halal framework
(no short selling), we instead trade only the LONG LEG of the pair: we buy
BTC/USDT when it appears statistically cheap relative to ETH/USDT, then exit
when the spread reverts to its historical mean.

The strategy models the cointegrated relationship between BTC and ETH prices
using a log-price spread and a rolling z-score. When BTC is unusually cheap
relative to ETH (z-score deeply negative), we buy BTC expecting the spread
to converge.

MATHEMATICAL FRAMEWORK
----------------------
Spread = log(BTC_price) − β × log(ETH_price)

where β (beta) is estimated as the ratio of volatility:
    β ≈ std(log_BTC) / std(log_ETH)  [rolling 30-bar window]

This simple beta gives a rough cointegration coefficient without requiring
OLS regression, making it suitable for a rolling indicator framework.

Z-score = (Spread − mean(Spread, 30)) / std(Spread, 30)

A z-score of -1.5 means the spread is 1.5 standard deviations below its
30-bar mean — BTC is cheap relative to ETH.

ENTRY SCALING (HALAL DCA)
--------------------------
Instead of shorting ETH, we scale INTO BTC as it becomes cheaper:
  Level 1: Z-score < -1.5  → first buy (1/3 position)
  Level 2: Z-score < -2.0  → add second tranche (1/3 position)
  Level 3: Z-score < -3.0  → add final tranche (1/3 position)

In freqtrade this is approximated via the entry signal. The scaling is
noted in comments; actual position sizing requires custom_entry or
DCA order configuration in freqtrade's config.

EXIT
----
Z-score returns to 0 (spread has converged to historical mean).

STOP
----
Z-score falls below -4.0 (spread has broken down — cointegration may have
failed). In freqtrade this is approximated by the hard stoploss (-10%).

PARAMETERS
----------
rolling_window : 30 bars (30 × 4h = 5 days)
z_entry_1      : -1.5  (first entry)
z_entry_2      : -2.0  (scale in)
z_entry_3      : -3.0  (final scale)
z_exit         :  0.0  (mean reversion complete)
z_stop         : -4.0  (relationship broken)

BACKTEST REFERENCE
------------------
Sharpe                 : 3.00
Sortino                : 3.11
Avg Excess Return      : 13.9% per 6-month period

INFORMATIVE PAIRS
-----------------
This strategy uses ETH/USDT as an informative pair to compute the spread.
The freqtrade informative_pairs() method returns [("ETH/USDT", "4h")].
ETH data is merged into the BTC/USDT dataframe in populate_indicators().

NOTE ON HALAL ADAPTATION
--------------------------
Traditional pair trading is considered to involve speculation on relative
price movements. This implementation trades only the spot asset (BTC), held
outright, using statistical analysis as an entry signal. No short positions,
no leverage, no derivatives are used. Scholars differ on whether spread-based
entries constitute permissible analysis; please consult your own religious
authority if needed.

RISK WARNINGS
-------------
- Cointegration between BTC and ETH is not guaranteed to persist.
  Regulatory events, protocol changes, or market structure shifts can
  cause permanent spread divergence.
- A -10% stoploss is wide; this reflects the higher noise in spread signals.
- The long-only adaptation captures only HALF the statistical edge of full
  pair trading. Sharpe will be lower in practice than the reference figure.
- Never run this on coins with low liquidity (stick to BTC/ETH major pairs).

HALAL COMPLIANCE
----------------
SPOT trading only. No leverage, no margin, no shorting, no futures/options.
"""

from typing import List, Optional, Tuple

import numpy as np
import pandas as pd
import pandas_ta as ta
from freqtrade.strategy import IStrategy, merge_informative_pair
from pandas import DataFrame


# ---------------------------------------------------------------------------
# Utility helpers
# ---------------------------------------------------------------------------

def crossed_above(s1: pd.Series, s2: pd.Series) -> pd.Series:
    """Returns True on the candle where s1 crosses above s2."""
    return (s1 > s2) & (s1.shift(1) <= s2.shift(1))


def crossed_below(s1: pd.Series, s2: pd.Series) -> pd.Series:
    """Returns True on the candle where s1 crosses below s2."""
    return (s1 < s2) & (s1.shift(1) >= s2.shift(1))


# ---------------------------------------------------------------------------
# Strategy class
# ---------------------------------------------------------------------------

class Strategy10_PairTrading(IStrategy):
    """
    Statistical Pair Trading BTC/ETH — Halal Long-Only Adaptation — 4h SPOT

    Trades BTC/USDT when its log-price spread to ETH/USDT shows statistical
    undervaluation (z-score < -1.5). Exits when z-score reverts to 0.

    See module docstring for full mathematical framework and risk warnings.
    """

    INTERFACE_VERSION = 3

    # ------------------------------------------------------------------
    # Metadata
    # ------------------------------------------------------------------
    timeframe = "4h"
    can_short = False  # HALAL: long-only, SPOT only

    # ------------------------------------------------------------------
    # ROI — wider targets for statistical strategies
    # ------------------------------------------------------------------
    minimal_roi = {
        "0":   0.10,   # 10% — pairs can take time to converge
        "120": 0.05,   # 5% after 120 min (2 candles on 4h)
        "480": 0.02,   # 2% after 480 min (2 trading days)
    }

    # ------------------------------------------------------------------
    # Stoploss — wide for spread strategies (noise is high)
    # ------------------------------------------------------------------
    stoploss = -0.10          # 10% — pair trading needs room to breathe
    trailing_stop = False     # No trailing; exit on z-score convergence

    # ------------------------------------------------------------------
    # Order settings
    # ------------------------------------------------------------------
    process_only_new_candles = True
    startup_candle_count = 60  # Need 30-bar rolling + warmup

    # ------------------------------------------------------------------
    # Strategy parameters
    # ------------------------------------------------------------------
    rolling_window = 30    # Lookback for z-score calculation (5 days on 4h)
    z_entry_1 = -1.5       # First entry threshold
    z_entry_2 = -2.0       # Scale-in threshold (informational; see notes)
    z_entry_3 = -3.0       # Final scale-in threshold (informational)
    z_exit    =  0.0       # Exit when spread reverts to mean
    z_stop    = -4.0       # Spread breakdown threshold

    # ------------------------------------------------------------------
    # Informative pairs — we need ETH/USDT data to compute the spread
    # ------------------------------------------------------------------
    def informative_pairs(self) -> List[Tuple[str, str]]:
        """
        Declare ETH/USDT as an informative pair.
        freqtrade will automatically download and align ETH/USDT OHLCV data
        at the same timeframe (4h) as BTC/USDT.
        """
        return [("ETH/USDT", self.timeframe)]

    # ------------------------------------------------------------------
    # Indicators
    # ------------------------------------------------------------------
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Merge ETH/USDT data and compute the log-price spread z-score.

        For non-BTC pairs this strategy will not generate signals (only
        BTC/USDT benefits from the ETH cointegration signal).

        Columns added
        -------------
        eth_close      : ETH/USDT close price (from informative pair)
        log_btc        : log(BTC close)
        log_eth        : log(ETH close)
        beta           : rolling volatility ratio std(log_btc)/std(log_eth)
        spread         : log_btc − beta × log_eth
        spread_mean    : rolling mean of spread (30 bars)
        spread_std     : rolling std of spread (30 bars)
        z_score        : (spread − spread_mean) / spread_std
        """

        # --- Merge ETH/USDT informative data ---
        # Only applicable when trading BTC/USDT
        if metadata.get("pair") != "BTC/USDT":
            # Fill z-score with NaN for non-BTC pairs (no signals generated)
            dataframe["z_score"] = np.nan
            dataframe["eth_close"] = np.nan
            return dataframe

        # Retrieve ETH data that freqtrade has fetched
        # The merge_informative_pair helper aligns timestamps and adds _<tf> suffix
        eth_data = self.dp.get_pair_dataframe(pair="ETH/USDT", timeframe=self.timeframe)

        if eth_data is None or eth_data.empty:
            dataframe["z_score"] = np.nan
            dataframe["eth_close"] = np.nan
            return dataframe

        # Merge ETH close into BTC dataframe on date
        dataframe = merge_informative_pair(
            dataframe, eth_data, self.timeframe, self.timeframe,
            ffill=True
        )

        # After merge, ETH close is named 'close_<timeframe>' e.g. 'close_4h'
        eth_close_col = f"close_{self.timeframe}"
        if eth_close_col not in dataframe.columns:
            # Fallback: try to find the ETH close column
            eth_cols = [c for c in dataframe.columns if "close" in c and c != "close"]
            eth_close_col = eth_cols[0] if eth_cols else None

        if eth_close_col is None or eth_close_col not in dataframe.columns:
            dataframe["z_score"] = np.nan
            dataframe["eth_close"] = np.nan
            return dataframe

        dataframe["eth_close"] = dataframe[eth_close_col]

        # --- Log prices ---
        dataframe["log_btc"] = np.log(dataframe["close"].replace(0, np.nan))
        dataframe["log_eth"] = np.log(dataframe["eth_close"].replace(0, np.nan))

        # --- Rolling beta (volatility ratio) ---
        roll_std_btc = dataframe["log_btc"].rolling(self.rolling_window).std()
        roll_std_eth = dataframe["log_eth"].rolling(self.rolling_window).std()
        dataframe["beta"] = roll_std_btc / roll_std_eth.replace(0, np.nan)

        # --- Spread ---
        dataframe["spread"] = dataframe["log_btc"] - dataframe["beta"] * dataframe["log_eth"]

        # --- Rolling Z-score ---
        dataframe["spread_mean"] = dataframe["spread"].rolling(self.rolling_window).mean()
        dataframe["spread_std"]  = dataframe["spread"].rolling(self.rolling_window).std()
        dataframe["z_score"] = (
            (dataframe["spread"] - dataframe["spread_mean"])
            / dataframe["spread_std"].replace(0, np.nan)
        )

        # --- NaN Safety: convert any None to NaN so pandas comparisons work ---
        for col in dataframe.columns:
            if col not in ['open', 'high', 'low', 'close', 'volume', 'date']:
                try:
                    dataframe[col] = pd.to_numeric(dataframe[col], errors='coerce')
                except (ValueError, TypeError):
                    pass

        return dataframe

    # ------------------------------------------------------------------
    # Entry logic
    # ------------------------------------------------------------------
    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Entry: z-score < -1.5 (BTC is statistically cheap vs ETH).

        This covers the first entry level. In a full DCA implementation:
          - z < -1.5 → buy first tranche
          - z < -2.0 → buy second tranche (requires freqtrade DCA config)
          - z < -3.0 → buy third tranche  (requires freqtrade DCA config)

        For simplicity in this single-signal implementation, the entry
        fires at z < -1.5. Additional tranches should be configured via
        freqtrade's position_adjustment_enable and adjust_trade_position().

        Note: The strategy only generates signals for BTC/USDT. Any other
        pair in the whitelist will not enter.
        """

        # Primary entry: z-score sufficiently negative
        conditions = (
            dataframe["z_score"].notna()                          &
            (dataframe["z_score"] < self.z_entry_1)              &
            (dataframe["volume"] > 0)
        )

        dataframe.loc[conditions, "enter_long"] = 1

        # Tag entry strength for informational purposes
        dataframe.loc[
            conditions & (dataframe["z_score"] < self.z_entry_2),
            "enter_tag"
        ] = "z_lt_minus2"

        dataframe.loc[
            conditions & (dataframe["z_score"] < self.z_entry_3),
            "enter_tag"
        ] = "z_lt_minus3"

        return dataframe

    # ------------------------------------------------------------------
    # Exit logic
    # ------------------------------------------------------------------
    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Exit when z-score returns to 0 (spread has converged to mean).

        Additional exits:
        - ROI table (time-based profit targets)
        - Hard stoploss (-10%) approximates the z < -4.0 breakdown stop
        - custom_exit() handles the explicit z < -4.0 breakdown check
        """

        conditions = (
            dataframe["z_score"].notna()        &
            (dataframe["z_score"] >= self.z_exit)
        )

        dataframe.loc[conditions, "exit_long"] = 1
        return dataframe

    # ------------------------------------------------------------------
    # Custom exit — spread breakdown detection
    # ------------------------------------------------------------------
    def custom_exit(
        self,
        pair: str,
        trade,
        current_time,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> Optional[str]:
        """
        Exit if the spread z-score falls below -4.0, indicating the
        BTC/ETH cointegration relationship has broken down and the
        position is unlikely to recover via mean reversion.

        We also implement a safety timeout: after 30 candles (5 days)
        with less than 1% profit, exit to free capital.
        """
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

        if dataframe is not None and not dataframe.empty and "z_score" in dataframe.columns:
            current_z = dataframe["z_score"].iloc[-1]

            # Spread breakdown — relationship may have failed
            if pd.notna(current_z) and current_z < self.z_stop:
                return "spread_breakdown_z_lt_minus4"

        # Time-based safety exit
        max_candles = 30
        candle_hours = 4
        max_hours = max_candles * candle_hours  # 5 days

        trade_hours = (current_time - trade.open_date_utc).total_seconds() / 3600
        if trade_hours >= max_hours and current_profit < 0.01:
            return f"pair_timeout_{max_candles}c"

        return None

    # ------------------------------------------------------------------
    # Optional: DCA / position adjustment scaffold
    # ------------------------------------------------------------------
    # Uncomment and configure the following if you want to implement
    # the three-tranche scaling logic natively in freqtrade.
    #
    # position_adjustment_enable = True
    #
    # def adjust_trade_position(self, trade, current_time, current_rate,
    #                           current_profit, min_stake, max_stake, **kwargs):
    #     """Add to position at z < -2.0 and z < -3.0."""
    #     dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
    #     if dataframe is None or dataframe.empty:
    #         return None
    #     z = dataframe["z_score"].iloc[-1]
    #     count = trade.nr_of_successful_buys
    #     if count == 1 and z < self.z_entry_2:
    #         return trade.stake_amount   # Double down
    #     if count == 2 and z < self.z_entry_3:
    #         return trade.stake_amount   # Triple down
    #     return None