Coding swings and fair value gaps in Python

Data format

We assume a pandas DataFrame with a time index and open, high, low, close columns, sorted from oldest to newest. The code is educational and must be adapted to your data.

Swing highs and swing lows

Careful: a swing with n candles on the right is only known n candles later. If you use it as a signal at candle i, you are looking into the future. That is why the function also returns the confirmation index.

import pandas as pd

def find_swings(df: pd.DataFrame, n: int = 3) -> pd.DataFrame:
    """Swing high/low with n candles per side.
    'confirmed_at' is the position at which the swing becomes known (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

The FVG is confirmed at the close of the third candle: there is no need to look further, so there is no look-ahead.

def find_fvg(df: pd.DataFrame, min_size: float = 0.0) -> pd.DataFrame:
    """Detects three-candle FVGs. min_size is in price units."""
    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:       # bullish FVG
            gaps.append({"pos": i, "type": "bullish", "low": c1_high, "high": c3_low})
        elif c1_low - c3_high > min_size:     # bearish FVG
            gaps.append({"pos": i, "type": "bearish", "low": c3_high, "high": c1_low})
    return pd.DataFrame(gaps)

Minimum size

A gap of a tenth of a pip is noise. Use a threshold tied to volatility, for example a fraction of the 14-period ATR computed up to the previous candle.

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()

How to continue

• Build a function that, at each candle, uses only the swings with confirmed_at less than or equal to the current candle
• Generate the signal (for example a return into the FVG after the sweep) and store it in a Series of +1, -1 and 0
• Pass the signal to the backtest in the next lesson