!pip install numpy pandas matplotlib seaborn requests --quiet

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import requests
from scipy import stats
import warnings

warnings.filterwarnings('ignore')
%matplotlib inline
plt.rcParams['figure.figsize'] = (14, 5)
plt.rcParams['axes.spines.top']   = False
plt.rcParams['axes.spines.right'] = False
print('Imports ready.')

# --- Configuration ---
USE_SYNTHETIC      = False
SYMBOL             = 'BTCUSDT'
PERIOD             = '15m'
LIMIT              = 500
ROLLING_WINDOW     = 48       # 48 x 15min = 12 hours
EXTREME_LONG_Z     =  1.5     # z-score above this → contrarian short signal
EXTREME_SHORT_Z    = -1.5     # z-score below this → contrarian long signal
SIGNAL_HOLD_BARS   = 8        # hold signal for N bars
print('Config ready.')

def generate_synthetic_ls_data(
    n: int = 500,
    seed: int = 42
) -> pd.DataFrame:
    """
    Generate synthetic long/short ratio and BTC price data.

    Parameters
    ----------
    n : int
        Number of 15-minute bars.
    seed : int
        Random seed.

    Returns
    -------
    pd.DataFrame
        Columns: timestamp, ls_ratio, btc_close.

    Notes
    -----
    L/S ratio is mean-reverting around 1.2 with slow trend component.
    BTC price is anti-correlated with L/S extremes to simulate contrarian signal.
    """
    rng = np.random.default_rng(seed)
    # Mean-reverting L/S around 1.2
    ls = [1.2]
    for _ in range(n - 1):
        ls.append(ls[-1] + 0.05 * (1.2 - ls[-1]) + rng.normal(0, 0.04))
    ls = np.clip(ls, 0.5, 3.0)

    # BTC price with inverse relation to extreme L/S
    btc = [50_000.0]
    for i in range(1, n):
        ls_dev = ls[i] - 1.2
        drift  = -0.0002 * ls_dev  # contrarian effect
        btc.append(btc[-1] * (1 + drift + rng.normal(0, 0.003)))

    timestamps = pd.date_range('2024-01-01', periods=n, freq='15min')
    return pd.DataFrame({'timestamp': timestamps, 'ls_ratio': ls, 'btc_close': btc})


def fetch_ls_ratio(
    symbol: str = 'BTCUSDT',
    period: str = '15m',
    limit: int = 500,
    use_synthetic: bool = False
) -> pd.DataFrame:
    """
    Fetch top-trader long/short ratio from Binance Futures API.

    Parameters
    ----------
    symbol : str
        Trading pair (e.g., 'BTCUSDT').
    period : str
        Candle period string ('5m', '15m', '1h', '4h', '1d').
    limit : int
        Number of bars to fetch (max 500).
    use_synthetic : bool
        Skip API call if True.

    Returns
    -------
    pd.DataFrame
        Columns: timestamp, ls_ratio, btc_close.
    """
    if use_synthetic:
        return generate_synthetic_ls_data(limit)

    try:
        # Top-trader position long/short ratio
        ls_url   = 'https://fapi.binance.com/futures/data/topLongShortPositionRatio'
        kline_url = 'https://fapi.binance.com/fapi/v1/klines'

        ls_resp = requests.get(ls_url, params={'symbol': symbol, 'period': period, 'limit': limit}, timeout=10)
        ls_resp.raise_for_status()
        ls_data = ls_resp.json()

        kl_resp = requests.get(kline_url, params={'symbol': symbol, 'interval': period, 'limit': limit}, timeout=10)
        kl_resp.raise_for_status()
        kl_data = kl_resp.json()

        ls_df = pd.DataFrame(ls_data)
        ls_df['timestamp'] = pd.to_datetime(ls_df['timestamp'].astype(int), unit='ms')
        ls_df['ls_ratio']  = ls_df['longShortRatio'].astype(float)

        kl_df = pd.DataFrame(kl_data, columns=['open_time','open','high','low','close','vol','close_time','qvol','ntrades','taker_buy_base','taker_buy_quote','ignore'])
        kl_df['timestamp']  = pd.to_datetime(kl_df['open_time'].astype(int), unit='ms')
        kl_df['btc_close']  = kl_df['close'].astype(float)

        df = pd.merge(ls_df[['timestamp', 'ls_ratio']], kl_df[['timestamp', 'btc_close']], on='timestamp', how='inner')
        print(f'Fetched {len(df)} bars of L/S + price data.')
        return df
    except Exception as e:
        print(f'API failed ({e}), using synthetic.')
        return generate_synthetic_ls_data(limit)


ls_df = fetch_ls_ratio(SYMBOL, PERIOD, LIMIT, USE_SYNTHETIC)
print(ls_df.describe())

def compute_ls_statistics(
    df: pd.DataFrame,
    window: int = 48
) -> pd.DataFrame:
    """
    Compute rolling statistics on the long/short ratio.

    Parameters
    ----------
    df : pd.DataFrame
        DataFrame with 'ls_ratio' column.
    window : int
        Rolling window size in bars.

    Returns
    -------
    pd.DataFrame
        Original df with added: ls_ma, ls_std, ls_zscore, ls_pct_rank.
    """
    df = df.copy()
    df['ls_ma']    = df['ls_ratio'].rolling(window).mean()
    df['ls_std']   = df['ls_ratio'].rolling(window).std()
    df['ls_zscore']= (df['ls_ratio'] - df['ls_ma']) / (df['ls_std'] + 1e-9)
    df['ls_pct_rank'] = df['ls_ratio'].rolling(window).rank(pct=True)
    return df


def generate_ls_signal(
    df: pd.DataFrame,
    long_z: float = -1.5,
    short_z: float = 1.5,
    hold_bars: int = 8
) -> pd.DataFrame:
    """
    Generate contrarian signals when L/S ratio reaches z-score extremes.

    Parameters
    ----------
    df : pd.DataFrame
        DataFrame with 'ls_zscore' column.
    long_z : float
        Z-score threshold below which to go long (everyone is short → reversal).
    short_z : float
        Z-score threshold above which to go short (everyone is long → reversal).
    hold_bars : int
        Number of bars to hold the signal before resetting.

    Returns
    -------
    pd.DataFrame
        Original df with 'signal' column: 1=long, -1=short, 0=flat.

    Notes
    -----
    Contrarian logic: when the crowd is extremely long, there's limited
    remaining buying pressure — the market is more likely to reverse.
    """
    df = df.copy()
    signal = np.zeros(len(df))
    hold_counter = 0
    current_signal = 0

    for i in range(len(df)):
        z = df['ls_zscore'].iloc[i]
        if np.isnan(z):
            signal[i] = 0
            continue

        if hold_counter > 0:
            signal[i] = current_signal
            hold_counter -= 1
        elif z >= short_z:
            current_signal = -1  # contrarian: everyone long → go short
            hold_counter   = hold_bars
            signal[i]      = -1
        elif z <= long_z:
            current_signal = 1   # contrarian: everyone short → go long
            hold_counter   = hold_bars
            signal[i]      = 1
        else:
            signal[i] = 0

    df['signal'] = signal
    return df


ls_df = compute_ls_statistics(ls_df, ROLLING_WINDOW)
ls_df = generate_ls_signal(ls_df, EXTREME_SHORT_Z, EXTREME_LONG_Z, SIGNAL_HOLD_BARS)
print(ls_df[['ls_ratio', 'ls_zscore', 'signal']].tail(10).to_string())
print(f'\nSignal distribution: {ls_df["signal"].value_counts().to_dict()}')

def backtest_ls_strategy(
    df: pd.DataFrame,
    signal_col: str = 'signal',
    price_col: str = 'btc_close'
) -> pd.DataFrame:
    """
    Vectorized backtest of the long/short ratio contrarian strategy.

    Parameters
    ----------
    df : pd.DataFrame
        DataFrame with signal and price columns.
    signal_col : str
        Column containing position signals (1/-1/0).
    price_col : str
        Column containing close prices.

    Returns
    -------
    pd.DataFrame
        Input df with added: price_ret, strategy_ret, equity, bh_equity.
    """
    df = df.copy()
    df['price_ret']    = df[price_col].pct_change()
    df['strategy_ret'] = df[signal_col].shift(1) * df['price_ret']
    df['equity']       = (1 + df['strategy_ret'].fillna(0)).cumprod()
    df['bh_equity']    = (1 + df['price_ret'].fillna(0)).cumprod()
    return df


def compute_strategy_metrics(df: pd.DataFrame) -> dict:
    """
    Compute key performance metrics for the backtest.

    Parameters
    ----------
    df : pd.DataFrame
        Output of backtest_ls_strategy().

    Returns
    -------
    dict
        Dictionary of metrics: total_return, sharpe, max_dd, n_signals.
    """
    rets = df['strategy_ret'].dropna()
    total_ret = df['equity'].iloc[-1] - 1
    sharpe    = rets.mean() / (rets.std() + 1e-9) * np.sqrt(252 * 96)  # 15min bars/day
    running_max = df['equity'].cummax()
    drawdown  = (df['equity'] - running_max) / running_max
    max_dd    = drawdown.min()
    n_signals = (df['signal'] != 0).sum()
    return {'total_return': total_ret, 'sharpe': sharpe, 'max_drawdown': max_dd, 'n_signals': int(n_signals)}


ls_df   = backtest_ls_strategy(ls_df)
metrics = compute_strategy_metrics(ls_df)
print('Strategy Metrics:')
for k, v in metrics.items():
    print(f'  {k}: {v:.4f}' if isinstance(v, float) else f'  {k}: {v}')

def plot_ls_analysis(df: pd.DataFrame) -> None:
    """
    Three-panel visualization: L/S ratio, z-score with signals, equity curves.

    Parameters
    ----------
    df : pd.DataFrame
        Fully enriched dataframe from the analysis pipeline.
    """
    fig, axes = plt.subplots(3, 1, figsize=(14, 12), sharex=True)

    # Panel 1: Price and signals
    axes[0].plot(df.index, df['btc_close'], color='steelblue', linewidth=1.0)
    long_sigs  = df[df['signal'] == 1]
    short_sigs = df[df['signal'] == -1]
    axes[0].scatter(long_sigs.index,  long_sigs['btc_close'],  color='green', marker='^', s=50, zorder=5, label='Long signal')
    axes[0].scatter(short_sigs.index, short_sigs['btc_close'], color='red',   marker='v', s=50, zorder=5, label='Short signal')
    axes[0].set_ylabel('BTC Price')
    axes[0].set_title('BTC Price with L/S Contrarian Signals')
    axes[0].legend()

    # Panel 2: L/S ratio and z-score
    ax2a = axes[1]
    ax2b = ax2a.twinx()
    ax2a.plot(df.index, df['ls_ratio'],  color='purple',    linewidth=1.0, label='L/S Ratio')
    ax2a.plot(df.index, df['ls_ma'],     color='purple',    linewidth=0.6, linestyle='--', alpha=0.6, label='Rolling Mean')
    ax2b.plot(df.index, df['ls_zscore'], color='darkorange', linewidth=1.0, alpha=0.7, label='Z-score')
    ax2b.axhline(EXTREME_LONG_Z,  color='red',   linestyle=':', linewidth=0.8)
    ax2b.axhline(EXTREME_SHORT_Z, color='green', linestyle=':', linewidth=0.8)
    ax2b.axhline(0, color='gray', linestyle='--', linewidth=0.5)
    ax2a.set_ylabel('L/S Ratio', color='purple')
    ax2b.set_ylabel('Z-score', color='darkorange')
    axes[1].set_title(f'Long/Short Ratio and Z-score (window={ROLLING_WINDOW})')
    ax2a.legend(loc='upper left')
    ax2b.legend(loc='upper right')

    # Panel 3: Equity curves
    axes[2].plot(df.index, df['equity'],    color='green',     linewidth=1.5, label='L/S Contrarian')
    axes[2].plot(df.index, df['bh_equity'], color='steelblue', linewidth=1.0, linestyle='--', alpha=0.7, label='Buy & Hold')
    axes[2].axhline(1.0, color='gray', linewidth=0.5, linestyle='--')
    axes[2].set_ylabel('Equity (normalized)')
    axes[2].set_title('Strategy vs Buy & Hold')
    axes[2].legend()

    plt.tight_layout()
    plt.show()


plot_ls_analysis(ls_df)

def export_ls_data(df: pd.DataFrame) -> None:
    """
    Export the enriched long/short ratio dataset.

    Parameters
    ----------
    df : pd.DataFrame  Fully processed L/S ratio dataframe.
    """
    df.to_csv('long_short_ratio_analysis.csv', index=True)
    print('Exported: long_short_ratio_analysis.csv')


export_ls_data(ls_df)