Statistical Pair Trading BTC/ETH — Halal Long-Only Adaptation — 4h SPOT
Timeframe
4h
Direction
Long Only
Stoploss
-10.0%
Trailing Stop
No
ROI
0m: 10.0%, 120m: 5.0%, 480m: 2.0%
Interface Version
3
Startup Candles
60
Indicators
0
freqtrade/freqtrade-strategies
# 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