Codificare struttura, FVG e sweep in Python (con codice completo)

Ora scriviamo il codice. L'obiettivo di questa lezione non è darti una strategia "pronta", ma un kit di funzioni pulite e verificabili che implementano le definizioni dei moduli precedenti, con attenzione al look-ahead. Il codice è didattico: leggilo, provalo su dati tuoi e adattalo. Ti consigliamo di salvarlo in due file (lib_structure.py e lib_signals.py) e di testare ogni funzione prima di usarla.

Il formato dei dati

Si assume un file CSV con colonne time, open, high, low, close, ordinato dal più vecchio al più recente e con orari in UTC. Ricorda: le stesse candele devono provenire dalla stessa fonte per tutto il test.

Parte 1 — Le funzioni di struttura (lib_structure.py)

import numpy as np
import pandas as pd


def load_candles(path):
    """CSV con colonne: time, open, high, low, close (orari in UTC)."""
    df = pd.read_csv(path, parse_dates=["time"])
    df = df.set_index("time").sort_index()
    return df[["open", "high", "low", "close"]]


def atr(df, period=14):
    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()


def find_swings(df, n=3):
    """Swing high/low con n candele per lato.
    Le colonne sono valorizzate alla candela di CONFERMA (i + n): nessun look-ahead."""
    window = 2 * n + 1
    is_high = df["high"] == df["high"].rolling(window, center=True).max()
    is_low = df["low"] == df["low"].rolling(window, center=True).min()
    swing_high = df["high"].where(is_high).shift(n)
    swing_low = df["low"].where(is_low).shift(n)
    return swing_high, swing_low


def find_fvg(df, min_size=0.0):
    """FVG a tre candele, noto alla chiusura della terza (C3)."""
    c1_high = df["high"].shift(2)
    c1_low = df["low"].shift(2)
    bull = (df["low"] - c1_high) > min_size
    bear = (c1_low - df["high"]) > min_size
    out = pd.DataFrame(index=df.index)
    out["bull"] = bull
    out["bull_low"] = c1_high.where(bull)
    out["bull_high"] = df["low"].where(bull)
    out["bear"] = bear
    out["bear_low"] = df["high"].where(bear)
    out["bear_high"] = c1_low.where(bear)
    return out


def asian_levels(df, start="00:00", end="06:00"):
    """Massimo/minimo asiatici del giorno, disponibili solo DOPO la fine del range."""
    rng = df.between_time(start, end, inclusive="left")
    g = rng.groupby(rng.index.date)
    ah = df.index.to_series().dt.date.map(g["high"].max())
    al = df.index.to_series().dt.date.map(g["low"].min())
    end_time = pd.Timestamp(end).time()
    known = df.index.time >= end_time
    return ah.where(known), al.where(known)


def sweep_low(df, level, atr_series, min_wick=0.2):
    """Sweep di un minimo: l'ombra supera il livello ma la chiusura rientra."""
    wick = level - df["low"]
    return (df["low"] < level) & (df["close"] >= level) & (wick >= min_wick * atr_series)


def sweep_high(df, level, atr_series, min_wick=0.2):
    wick = df["high"] - level
    return (df["high"] > level) & (df["close"] <= level) & (wick >= min_wick * atr_series)

Cosa fa ogni funzione (e perché è scritta così)

• atr: media mobile del True Range. È la base di quasi tutte le soglie (displacement, ampiezza del FVG, ombra dello sweep).
• find_swings: usa rolling(2n+1, center=True) per trovare i massimi/minimi locali e poi shift(n) per spostare il valore alla candela di conferma. Nella riga di conferma trovi quindi il prezzo dello swing noto in quel momento: nessun look-ahead. Nota: con prezzi identici possono risultare più swing vicini (è un caso limite da gestire).
• find_fvg: confronta l'high di C1 (shift(2)) con il low di C3. Il gap è noto alla chiusura di C3, quindi non serve alcuno spostamento.
• asian_levels: calcola per ogni giorno massimo e minimo dell'intervallo asiatico e li rende disponibili solo dopo la fine dell'intervallo (maschera known).
• sweep_low / sweep_high: implementano la definizione operativa (ombra oltre il livello, chiusura dentro, ampiezza minima in ATR).

Parte 2 — Il generatore di segnali (lib_signals.py)

import numpy as np
import pandas as pd
from lib_structure import atr, find_swings, find_fvg, asian_levels, sweep_low


def bullish_signals(df, n=3, disp=1.5, fvg_min=0.3, max_mss_bars=12, max_return_bars=20,
                    kz_start="07:00", kz_end="10:00"):
    """Ritorna una Series con 1 = ingresso long alla candela SUCCESSIVA, 0 altrimenti.
    Sequenza: sweep del minimo asiatico -> MSS con displacement -> ritorno nel FVG."""
    a = atr(df, 14)
    sh, sl = find_swings(df, n)
    last_sh = sh.ffill()                       # ultimo swing high confermato
    ah, al = asian_levels(df)
    fvg = find_fvg(df)
    body = (df["close"] - df["open"]).abs()

    swept = sweep_low(df, al, a)
    in_kz = pd.Series(df.index.time, index=df.index)
    in_kz = (in_kz >= pd.Timestamp(kz_start).time()) & (in_kz < pd.Timestamp(kz_end).time())

    signals = pd.Series(0, index=df.index)
    stops = pd.Series(np.nan, index=df.index)
    state, sweep_low_price = "idle", None
    mss_bar, zone = None, None

    for i in range(20, len(df)):
        ts = df.index[i]
        if state == "idle":
            if swept.iloc[i] and in_kz.iloc[i]:
                state, sweep_low_price, sweep_bar = "wait_mss", df["low"].iloc[i], i
        elif state == "wait_mss":
            if i - sweep_bar > max_mss_bars or df["low"].iloc[i] < sweep_low_price:
                state = "idle"
                continue
            strong = body.iloc[i] > disp * a.iloc[i]
            broke = df["close"].iloc[i] > last_sh.iloc[i]
            if strong and broke:
                # FVG creato nelle ultime 3 candele dello spostamento
                recent = fvg["bull"].iloc[i - 2 : i + 1]
                if recent.any():
                    j = recent[recent].index[-1]
                    lo, hi = fvg.loc[j, "bull_low"], fvg.loc[j, "bull_high"]
                    if (hi - lo) >= fvg_min * a.iloc[i]:
                        state, zone, mss_bar = "wait_return", (lo, hi), i
                        continue
                state = "idle"
        elif state == "wait_return":
            if i - mss_bar > max_return_bars or df["low"].iloc[i] < sweep_low_price:
                state = "idle"
                continue
            ce = (zone[0] + zone[1]) / 2
            if df["low"].iloc[i] <= ce and df["close"].iloc[i] > df["open"].iloc[i]:
                signals.iloc[i] = 1
                stops.iloc[i] = sweep_low_price
                state = "idle"
    return signals, stops

Come leggerlo

La funzione bullish_signals è una macchina a stati con tre stati:

1. idle: cerca uno sweep del minimo asiatico dentro la killzone.
2. wait_mss: dopo lo sweep aspetta, entro max_mss_bars candele, una candela di forza (body > disp × ATR) che chiude sopra l'ultimo swing high confermato e lascia un FVG di ampiezza sufficiente. Se il prezzo scende sotto il minimo dello sweep, l'idea è invalidata e si torna a idle.
3. wait_return: aspetta il ritorno nel FVG (tocca il CE con una candela di chiusura rialzista). Genera il segnale e memorizza lo stop (minimo dello sweep).

Restituisce due serie: i segnali (1 = ingresso alla candela successiva) e gli stop. Per le vendite si scrive la funzione speculare (bearish_signals), usando sweep_high, gli swing low e i FVG ribassisti.

Come controllare che il codice sia giusto

1. Test unitari (lezione precedente) per ogni funzione.
2. Controllo visivo: prendi 5 segnali generati e verificali a mano su TradingView. Corrispondono alla tua definizione?
3. Test "spostamento del futuro": taglia il dataset a una certa data e rigenera i segnali. Quelli prima di quella data devono essere identici a quelli generati sull'intero dataset. Se cambiano, c'è look-ahead.

sig_full, _ = bullish_signals(df)
sig_cut, _ = bullish_signals(df.loc[:"2025-06-30"])
before = sig_full.loc[:"2025-06-30"]
assert (before.values == sig_cut.values).all(), "look-ahead: i segnali cambiano col futuro"

Questo test è potentissimo: verifica direttamente che il codice non usi il futuro.

Prestazioni e semplificazioni

• Le funzioni di struttura sono vettoriali (veloci); la macchina a stati è un ciclo Python: su qualche anno di dati M5 (centinaia di migliaia di candele) va bene, per volumi enormi si può ottimizzare con Numba.
• Il codice usa una sola direzione (long) per semplicità: per un sistema completo aggiungi le vendite e gestisci il caso di segnali contrari.
• Non gestisce swap, commissioni o filtri di notizie: sono estensioni da aggiungere e testare.

Estensioni utili

• Filtri di bias: aggiungere un punteggio di direzione (struttura D1/H4) come condizione.
• Filtro di notizie: escludere i segnali vicini agli eventi del calendario.
• Più livelli di liquidità: PDH/PDL, equal highs/lows con tolleranza.
• Order block come area alternativa al FVG.
• Registrare le feature (ampiezza sweep, displacement in ATR) per ogni segnale, per analizzarle dopo.

In sintesi

• Ogni feature è scritta in modo da essere nota solo quando lo sarebbe in tempo reale (shift, candele chiuse, livelli disponibili dopo la fine dell'intervallo).
• Il generatore di segnali è una macchina a stati: sweep → MSS con displacement → ritorno nel FVG.
• Verifica il codice con test unitari, controllo visivo e il test dello "spostamento del futuro".
• Parti dalle definizioni semplici e aggiungi complessità solo se i dati lo giustificano.

Esercizio pratico

1. Salva le due librerie, carica un CSV M5 e genera i segnali su 6 mesi.
2. Verifica a mano 5 segnali su TradingView e il test dello "spostamento del futuro".
3. Aggiungi la versione ribassista dei segnali e un registro delle feature per ogni segnale.