pairs-cointegration / src / signals.py
signals.py
Raw
import pandas as pd
import numpy as np


def generate_signals(
    zscore: pd.Series,
    entry_threshold: float = 2.0,
    exit_threshold: float = 0.5,
) -> pd.DataFrame:
    """
    Génère les signaux de trading basés sur le z-score du spread.

    Règles (machine à états) :
    - LONG spread (buy A, short B) quand z-score < -entry_threshold
    - SHORT spread (short A, buy B) quand z-score > +entry_threshold
    - EXIT quand |z-score| < exit_threshold
    - STOP LOSS si |z-score| > 3.5 (le spread diverge trop)

    Parameters
    ----------
    zscore : pd.Series
        Le z-score rolling du spread.
    entry_threshold : float
        Seuil d'entrée en position (défaut : 2.0).
    exit_threshold : float
        Seuil de sortie de position (défaut : 0.5).

    Returns
    -------
    pd.DataFrame
        DataFrame avec colonnes :
        - 'zscore' : le z-score
        - 'signal' : 1 (long spread), -1 (short spread), 0 (flat)
        - 'trade_type' : 'entry_long', 'entry_short', 'exit', 'stop_loss', None
    """
    seuil_stop_loss = 3.5

    # Initialisation des listes pour stocker les résultats
    signaux = np.zeros(len(zscore), dtype=int)
    types_trade = [None] * len(zscore)

    position_actuelle = 0  # 0 = flat, 1 = long, -1 = short

    for i, z in enumerate(zscore.values):
        # Ignorer les NaN (début de la série avant que le rolling soit calculé)
        if np.isnan(z):
            signaux[i] = 0
            continue

        if position_actuelle == 0:
            # Pas de position → chercher une entrée
            if z < -entry_threshold:
                position_actuelle = 1
                types_trade[i] = "entry_long"
            elif z > entry_threshold:
                position_actuelle = -1
                types_trade[i] = "entry_short"
        else:
            # En position → chercher une sortie
            if abs(z) > seuil_stop_loss:
                position_actuelle = 0
                types_trade[i] = "stop_loss"
            elif abs(z) < exit_threshold:
                position_actuelle = 0
                types_trade[i] = "exit"

        signaux[i] = position_actuelle

    return pd.DataFrame(
        {
            "zscore": zscore.values,
            "signal": signaux,
            "trade_type": types_trade,
        },
        index=zscore.index,
    )