Codificare swing e fair value gap in Python

Formato dei dati

Presumiamo un DataFrame pandas con indice temporale e colonne open, high, low, close, ordinato dal più vecchio al più recente. Il codice è didattico e va adattato ai tuoi dati.

Swing high e swing low

Attenzione: uno swing con n candele a destra è noto solo n candele dopo. Se lo usi come segnale alla candela i, stai guardando il futuro. Per questo la funzione restituisce anche l'indice di conferma.

import pandas as pd

def find_swings(df: pd.DataFrame, n: int = 3) -> pd.DataFrame:
    """Swing high/low con n candele per lato.
    'confirmed_at' e' la posizione in cui lo swing diventa noto (i + n)."""
    swings = []
    highs = df["high"].values
    lows = df["low"].values
    for i in range(n, len(df) - n):
        left_h, right_h = highs[i - n:i], highs[i + 1:i + n + 1]
        left_l, right_l = lows[i - n:i], lows[i + 1:i + n + 1]
        if highs[i] > left_h.max() and highs[i] > right_h.max():
            swings.append({"pos": i, "type": "high", "price": highs[i], "confirmed_at": i + n})
        if lows[i] < left_l.min() and lows[i] < right_l.min():
            swings.append({"pos": i, "type": "low", "price": lows[i], "confirmed_at": i + n})
    return pd.DataFrame(swings)

Fair value gap

Il FVG si conferma alla chiusura della terza candela: non serve guardare oltre, quindi non c'è look-ahead.

def find_fvg(df: pd.DataFrame, min_size: float = 0.0) -> pd.DataFrame:
    """Individua FVG su tre candele. min_size in unita' di prezzo."""
    gaps = []
    for i in range(2, len(df)):
        c1_high, c1_low = df["high"].iloc[i - 2], df["low"].iloc[i - 2]
        c3_high, c3_low = df["high"].iloc[i], df["low"].iloc[i]
        if c3_low - c1_high > min_size:       # FVG rialzista
            gaps.append({"pos": i, "type": "bullish", "low": c1_high, "high": c3_low})
        elif c1_low - c3_high > min_size:     # FVG ribassista
            gaps.append({"pos": i, "type": "bearish", "low": c3_high, "high": c1_low})
    return pd.DataFrame(gaps)

Ampiezza minima

Un gap di un decimo di pip è rumore. Usa una soglia legata alla volatilità, per esempio una frazione dell'ATR a 14 periodi calcolato fino alla candela precedente.

def atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
    prev_close = df["close"].shift(1)
    tr = pd.concat([
        df["high"] - df["low"],
        (df["high"] - prev_close).abs(),
        (df["low"] - prev_close).abs(),
    ], axis=1).max(axis=1)
    return tr.rolling(period).mean()

Come proseguire

• Costruisci una funzione che, ad ogni candela, usa solo gli swing con confirmed_at minore o uguale alla candela corrente
• Genera il segnale (per esempio ritorno nel FVG dopo lo sweep) e salvalo in una Series di +1, -1 e 0
• Passa il segnale al backtest della prossima lezione