Coding structure, FVG and sweeps in Python (with complete code)

Now let's write the code. The aim of this lesson isn't to give you a "ready-made" strategy, but a kit of clean, verifiable functions that implement the definitions from the previous modules, with attention to look-ahead. The code is educational: read it, try it on your own data and adapt it. We advise saving it in two files (lib_structure.py and lib_signals.py) and testing each function before using it.

The data format

A CSV file is assumed with columns time, open, high, low, close, sorted from oldest to newest and with times in UTC. Remember: the same candles must come from the same source for the whole test.

Part 1 — The structure functions (lib_structure.py)

import numpy as np
import pandas as pd


def load_candles(path):
    """CSV with columns: time, open, high, low, close (times 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 with n candles per side.
    Values are placed on the CONFIRMATION candle (i + n): no 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):
    """Three-candle FVG, known at the close of the third (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"):
    """Day's Asian high/low, available only AFTER the range ends."""
    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 of a low: the wick exceeds the level but the close returns."""
    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)

What each function does (and why it's written this way)

• atr: moving average of the True Range. It's the basis of almost all thresholds (displacement, FVG size, sweep wick).
• find_swings: uses rolling(2n+1, center=True) to find local highs/lows and then shift(n) to move the value to the confirmation candle. In the confirmation row you therefore find the swing price known at that moment: no look-ahead. Note: with identical prices, several nearby swings can result (an edge case to handle).
• find_fvg: compares the high of C1 (shift(2)) with the low of C3. The gap is known at the close of C3, so no shift is needed.
• asian_levels: computes for each day the high and low of the Asian interval and makes them available only after the end of the interval (the known mask).
• sweep_low / sweep_high: implement the operational definition (wick beyond the level, close inside, minimum size in ATR).

Part 2 — The signal generator (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"):
    """Returns a Series with 1 = long entry at the NEXT candle, 0 otherwise.
    Sequence: sweep of the Asian low -> MSS with displacement -> return into the FVG."""
    a = atr(df, 14)
    sh, sl = find_swings(df, n)
    last_sh = sh.ffill()                       # last confirmed swing high
    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 created in the last 3 candles of the displacement
                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

How to read it

The bullish_signals function is a state machine with three states:

1. idle: looks for a sweep of the Asian low inside the killzone.
2. wait_mss: after the sweep it waits, within max_mss_bars candles, for a strong candle (body > disp × ATR) that closes above the last confirmed swing high and leaves an FVG of sufficient size. If price falls below the sweep's low, the idea is invalidated and it returns to idle.
3. wait_return: waits for the return into the FVG (touches the CE with a bullish-closing candle). It generates the signal and stores the stop (the sweep's low).

It returns two series: the signals (1 = entry at the next candle) and the stops. For sells write the mirror function (bearish_signals), using sweep_high, swing lows and bearish FVGs.

How to check the code is right

1. Unit tests (previous lesson) for each function.
2. Visual check: take 5 generated signals and verify them by hand on TradingView. Do they match your definition?
3. "Shift the future" test: cut the dataset at a certain date and regenerate the signals. Those before that date must be identical to those generated on the whole dataset. If they change, there's 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: signals change with the future"

This test is very powerful: it directly verifies that the code doesn't use the future.

Performance and simplifications

• The structure functions are vectorised (fast); the state machine is a Python loop: on a few years of M5 data (hundreds of thousands of candles) it's fine, for huge volumes you can optimise with Numba.
• The code uses a single direction (long) for simplicity: for a complete system add sells and handle the case of opposing signals.
• It doesn't handle swap, commissions or news filters: these are extensions to add and test.

Useful extensions

• Bias filters: adding a direction score (D1/H4 structure) as a condition.
• News filter: excluding signals near calendar events.
• More liquidity levels: PDH/PDL, equal highs/lows with tolerance.
• Order block as an alternative area to the FVG.
• Recording features (sweep size, displacement in ATR) for each signal, to analyse afterwards.

In short

• Each feature is written to be known only when it would be in real time (shift, closed candles, levels available after the interval ends).
• The signal generator is a state machine: sweep → MSS with displacement → return into the FVG.
• Verify the code with unit tests, visual checks and the "shift the future" test.
• Start from simple definitions and add complexity only if the data justify it.

Practical exercise

1. Save the two libraries, load an M5 CSV and generate signals over 6 months.
2. Verify 5 signals by hand on TradingView and run the "shift the future" test.
3. Add the bearish version of the signals and a feature log for every signal.