How to Build a Target Leakage Linter in Python With a Dependency Graph

If you have ever trained a model that scored suspiciously well on the first try, this tutorial is about the most common reason why: target leakage. That is when one of the columns you feed a model already contains, in some disguised form, the answer you are asking it to predict. The model is not learning anything real. It is just decoding a formula that a spreadsheet, a government agency, or a colleague on another team already applied.

By the end of this tutorial you will have built a small command-line tool that catches this automatically. It treats dataset columns the way a package manager treats software packages: every column can declare what it was derived from, in a plain text file. The tool reads that file, builds a dependency graph out of it, and refuses to let you train on any column that sits upstream or downstream of the thing you are trying to predict. You will run it, watch it catch a real leak that a simpler version of the same idea misses, then break it on purpose three different ways to make sure it fails loudly instead of failing silently.

What is target leakage, in plain language

scikit-learn’s own documentation defines it directly:

“Data leakage occurs when information that would not be available at prediction time is used when building the model. This results in overly optimistic performance estimates, for example from cross-validation, and thus poorer performance when the model is used on actually novel data, for example during production.”

Target leakage is the specific flavor of this problem where the leaking information comes from the target column itself, or from something the target was computed from. It is dangerous precisely because it does not look dangerous. A model with a target leak does not crash, does not throw an error, and does not obviously misbehave. It just reports a much better score than it deserves, and that score quietly falls apart the moment the model sees a case where the leaking column is not conveniently pre-computed for it.

Public data is an especially easy place for this to happen, because a large share of what gets published is itself a calculation on top of other published columns. The CDC and ATSDR’s own Social Vulnerability Index documentation describes exactly this pattern: “The current CDC/ATSDR Social Vulnerability Index uses 16 U.S. Census variables from the 5-year American Community Survey (ACS)… These variables are grouped into four themes… and then combined into a single measure of overall social vulnerability.” If a raw input variable and the combined index both show up as separate columns in a dataset you downloaded, and you are not told which one was built from the other, you have exactly the setup this tutorial is about. You will not use the CDC’s actual dataset here (it is a real one and worth exploring on your own), but you will build a dataset with the same underlying structure so you can see, verify, and fix the problem yourself.

Prerequisites

• Python 3.10 or newer. This tutorial was built and tested on Python 3.13.14.
• Basic familiarity with pandas DataFrames (reading a CSV, selecting columns) and a rough sense of what a YAML file looks like. You do not need prior scikit-learn experience; every model call here is a single function.
• A terminal and about 20 MB of free disk space.
• No GPU, Docker, or cloud account of any kind. Everything in this tutorial runs locally with the Python standard library plus four packages.

Create a fresh project folder with a virtual environment, so these packages stay separate from anything else on your machine:

mkdir leak-linter-tutorial
cd leak-linter-tutorial
python -m venv .venv

Activate it. On Windows:

.venv\Scripts\activate

On macOS or Linux:

source .venv/bin/activate

Then install the four libraries this tutorial uses:

pip install pandas scikit-learn pyyaml pytest

This tutorial pinned pandas 3.0.6, scikit-learn 1.9.1, pyyaml 6.0.3, and pytest 9.1.1 in the environment used to write and verify every command below. Newer versions should behave the same way for everything shown here.

Key terms in plain English

• Target: the column you are trying to predict.
• Covariate (or feature): a column you feed the model as an input.
• Derived column: a column that was computed from one or more other columns, rather than measured or collected directly.
• Manifest: a text file that records, for each column, which other columns it was derived from. This tutorial uses YAML for it, the same format tools like GitHub Actions and Docker Compose use for their own configuration.
• Ancestor: any column that was used, directly or through a chain of steps, to compute another column. If C was built from B, and B was built from A, then A and B are both ancestors of C.
• Descendant: the reverse relationship. A and B are ancestors of C, so C is a descendant of both A and B.
• BFS (breadth-first search): a way of walking a graph one layer at a time. Picture a queue at a ticket counter. You start with one node in the queue. On each turn you take the node at the front, look at everything connected to it, and add anything you have not seen yet to the back of the queue. When the queue empties, you have visited every node reachable from the start, at every distance, and you never visit the same node twice.

Step 1: Build a dataset with a leak already inside it

You will use a small synthetic dataset of 400 fictional retail stores, each with a few raw operational metrics and a “sustainability score” that a hypothetical ESG reporting team computed from those metrics. This mirrors how a real composite index like the CDC’s SVI is put together: raw measurements go in, a formula combines them, and the formula’s output gets published as its own column, right alongside the raw measurements it came from.

Create generate_data.py:

"""generate_data.py: build a synthetic retail sustainability dataset with a real,
multi-hop derivation chain baked in (raw metrics -> intensities -> score -> grade)."""
import numpy as np
import pandas as pd

rng = np.random.default_rng(20260920)
N = 400

store_id = [f"STR-{i:04d}" for i in range(1, N + 1)]
region = rng.choice(["Northeast", "Midwest", "South", "West"], size=N)
square_footage = rng.uniform(2000, 15000, size=N).round(0)

# Raw operational metrics: loosely scale with store size, plus independent noise.
annual_energy_kwh = (square_footage * rng.uniform(8, 14, size=N) + rng.normal(0, 4000, size=N)).clip(min=1000).round(0)
annual_water_gal = (square_footage * rng.uniform(3, 7, size=N) + rng.normal(0, 2000, size=N)).clip(min=500).round(0)
annual_waste_lbs = (square_footage * rng.uniform(1, 3, size=N) + rng.normal(0, 500, size=N)).clip(min=100).round(0)

# A raw metric that is NOT part of the sustainability formula at all (decoy / true negative).
annual_revenue_usd = rng.normal(500000, 150000, size=N).clip(min=50000).round(0)
store_manager_tenure_years = rng.uniform(0.5, 12, size=N).round(1)

df = pd.DataFrame({
    "store_id": store_id,
    "region": region,
    "square_footage": square_footage,
    "annual_energy_kwh": annual_energy_kwh,
    "annual_water_gal": annual_water_gal,
    "annual_waste_lbs": annual_waste_lbs,
    "annual_revenue_usd": annual_revenue_usd,
    "store_manager_tenure_years": store_manager_tenure_years,
})

# --- Derived columns: this is the part a data catalogue rarely writes down. ---

# Hop 1: per-square-foot intensities (deterministic ratios).
df["energy_intensity_kwh_sqft"] = df["annual_energy_kwh"] / df["square_footage"]
df["water_intensity_gal_sqft"] = df["annual_water_gal"] / df["square_footage"]
df["waste_intensity_lbs_sqft"] = df["annual_waste_lbs"] / df["square_footage"]

# Hop 2: sustainability_score, a weighted sum of the three intensities (deterministic, no noise).
raw_score = (
    100
    - 2.0 * df["energy_intensity_kwh_sqft"]
    - 2.0 * df["water_intensity_gal_sqft"]
    - 5.0 * df["waste_intensity_lbs_sqft"]
)
df["sustainability_score"] = raw_score.clip(0, 100).round(2)

# Hop 3: sustainability_grade, a letter grade binned from the score (statistical/lossy).
def grade(score):
    if score >= 90:
        return "A"
    if score >= 75:
        return "B"
    if score >= 60:
        return "C"
    if score >= 40:
        return "D"
    return "F"

df["sustainability_grade"] = df["sustainability_score"].apply(grade)

df.to_csv("stores.csv", index=False)
print("wrote stores.csv,", len(df), "rows")
print(df.head(3).to_string())

Run it:

python generate_data.py

You get output like this (values will differ slightly between pandas versions but the shape stays the same):

wrote stores.csv, 400 rows
   store_id     region  square_footage  annual_energy_kwh  ...  sustainability_score sustainability_grade
0  STR-0001      South         11826.0           146953.0  ...                 55.70                    D
1  STR-0002  Northeast          4678.0            49896.0  ...                 62.13                    C
2  STR-0003       West          5689.0            53226.0  ...                 68.35                    C

Look at what just happened. sustainability_score is not a measurement. It is 100 minus a weighted sum of three ratios, and each of those ratios is itself a raw metric divided by square footage. Nobody wrote any of that down anywhere in the CSV file. If you handed this file to a teammate with no other context, they would have no way to know that energy_intensity_kwh_sqft and sustainability_score are related at all, let alone that one determines the other.

Step 2: Prove the leak with a real model

Now build a model the way an analyst who has not seen generate_data.py might, using columns that sound like plausible predictors. Create leak_demo.py:

"""leak_demo.py: reproduce a target leak against the synthetic stores.csv dataset."""
import pandas as pd
from sklearn.ensemble import HistGradientBoostingRegressor
from sklearn.model_selection import KFold, cross_val_score

df = pd.read_csv("stores.csv")


def score_model(feature_cols, target_col, label):
    data = df[feature_cols + [target_col]].dropna()
    model = HistGradientBoostingRegressor(random_state=0)
    folds = KFold(n_splits=5, shuffle=True, random_state=0)
    r2_scores = cross_val_score(model, data[feature_cols], data[target_col], cv=folds, scoring="r2")
    print(f"{label}: features={feature_cols}")
    print(f"  R^2 per fold: {[round(float(s), 4) for s in r2_scores]}")
    print(f"  mean R^2: {r2_scores.mean():.4f}\n")
    return r2_scores.mean()


# Case 1: a direct leak. These three columns ARE the literal formula inputs.
r2_case1 = score_model(
    ["energy_intensity_kwh_sqft", "water_intensity_gal_sqft", "waste_intensity_lbs_sqft"],
    "sustainability_score",
    "Case 1 (direct leak: the exact formula inputs)",
)

# Case 2: a descendant leak. These are one algebra step further back, but still
# fully determine the target once you also know square_footage.
r2_case2 = score_model(
    ["annual_energy_kwh", "annual_water_gal", "annual_waste_lbs", "square_footage"],
    "sustainability_score",
    "Case 2 (descendant leak: raw inputs the intensities were built from)",
)

# A legitimate model: two columns that never fed the formula at all.
r2_decoy = score_model(
    ["annual_revenue_usd", "store_manager_tenure_years"],
    "sustainability_score",
    "Decoy (independent columns, no relationship to the score)",
)

print("Summary:")
print(f"  Case 1 direct leak mean R^2:     {r2_case1:.4f}")
print(f"  Case 2 descendant leak mean R^2: {r2_case2:.4f}")
print(f"  Decoy (no leak) mean R^2:        {r2_decoy:.4f}")

python leak_demo.py

Real output from this exact run:

Case 1 (direct leak: the exact formula inputs): features=['energy_intensity_kwh_sqft', 'water_intensity_gal_sqft', 'waste_intensity_lbs_sqft']
  R^2 per fold: [0.9727, 0.9618, 0.9619, 0.9611, 0.9609]
  mean R^2: 0.9637

Case 2 (descendant leak: raw inputs the intensities were built from): features=['annual_energy_kwh', 'annual_water_gal', 'annual_waste_lbs', 'square_footage']
  R^2 per fold: [0.795, 0.7369, 0.635, 0.8116, 0.8588]
  mean R^2: 0.7675

Decoy (independent columns, no relationship to the score): features=['annual_revenue_usd', 'store_manager_tenure_years']
  R^2 per fold: [-0.2436, -0.2575, -0.2427, -0.1087, -0.1012]
  mean R^2: -0.1907

Summary:
  Case 1 direct leak mean R^2:     0.9637
  Case 2 descendant leak mean R^2: 0.7675
  Decoy (no leak) mean R^2:        -0.1907

Three real results, and they tell three different stories. Case 1 scores 0.96, unsurprising once you know it is being handed the literal formula inputs. The decoy scores negative, which for R² means the model does worse than just guessing the average every time, exactly what you want from two columns with no real relationship to the target. Case 2 is the one to pay attention to: 0.77, using only “raw operational metrics” that never appear anywhere in the sustainability formula’s printed inputs. An analyst who only checked whether their features were the exact three intensity columns would sign off on this model. They would be wrong, because annual_energy_kwh, annual_water_gal, annual_waste_lbs, and square_footage are exactly what energy_intensity_kwh_sqft, water_intensity_gal_sqft, and waste_intensity_lbs_sqft are computed from. The leak is one algebra step removed, not gone.

Step 3: Write the dependency manifest

The fix is not to memorize which columns are dangerous. It is to write down, once, in a file a computer can read, exactly how each derived column was built, then let a tool check that for you every time. This is the same idea a package manager’s lockfile uses: instead of a human remembering that package A needs package B version 2 or higher, the lockfile says so, and tooling enforces it automatically.

Create products.yaml:

# products.yaml: a dependency manifest for the stores.csv dataset.
# Each product is a column. "derived_from" lists the parents (and how) it was
# computed from, exactly like a lockfile lists a package's own dependencies.
products:
  square_footage:
    description: "Store's total retail floor area, in square feet."

  annual_energy_kwh:
    description: "Total electricity consumed by the store in a year, in kWh."

  annual_water_gal:
    description: "Total water consumed by the store in a year, in gallons."

  annual_waste_lbs:
    description: "Total solid waste generated by the store in a year, in pounds."

  annual_revenue_usd:
    description: "Store's annual gross revenue, in US dollars. Independent of the sustainability formula."

  energy_intensity_kwh_sqft:
    description: "Energy used per square foot of retail space."
    derived_from:
      - {from: annual_energy_kwh, relation: ratio}
      - {from: square_footage, relation: ratio}

  water_intensity_gal_sqft:
    description: "Water used per square foot of retail space."
    derived_from:
      - {from: annual_water_gal, relation: ratio}
      - {from: square_footage, relation: ratio}

  waste_intensity_lbs_sqft:
    description: "Waste generated per square foot of retail space."
    derived_from:
      - {from: annual_waste_lbs, relation: ratio}
      - {from: square_footage, relation: ratio}

  sustainability_score:
    description: "0-100 composite score: 100 minus a weighted sum of the three intensities."
    derived_from:
      - {from: energy_intensity_kwh_sqft, relation: weighted_sum}
      - {from: water_intensity_gal_sqft, relation: weighted_sum}
      - {from: waste_intensity_lbs_sqft, relation: weighted_sum}

  sustainability_grade:
    description: "Letter grade (A-F) binned from sustainability_score."
    derived_from:
      - {from: sustainability_score, relation: binning}

Two things worth noticing about this file. First, annual_revenue_usd is listed with no derived_from at all. That is a deliberate, documented statement: “this column is raw and independent.” Second, notice what is missing entirely: store_manager_tenure_years, a real column that exists in stores.csv, is not mentioned anywhere in this manifest. That is not a mistake in this tutorial. It is there on purpose, to represent the most common real-world case: a column somebody added to the dataset without documenting it anywhere. You will see exactly how the linter treats that differently from a column it has actually verified is safe.

Step 4: Build a linter, and watch it fail

Your first instinct for a checking tool might be: “for a given target, look up its direct parents in the manifest, and flag any candidate feature that matches.” That is a reasonable first instinct, and it is also broken. Build it anyway, on purpose, so you can see exactly how it breaks.

Create leak_linter_naive.py:

"""leak_linter_naive.py: a first-pass linter that only checks DIRECT parents.
This has a real bug: it misses leaks that are more than one hop away."""
import yaml


def load_manifest(path):
    with open(path) as f:
        data = yaml.safe_load(f)
    return data["products"]


def direct_parents(products, node):
    """Return the immediate parents of `node` only (one hop)."""
    entry = products.get(node, {})
    return {edge["from"] for edge in entry.get("derived_from", [])}


def check_naive(products, target, candidates):
    parents = direct_parents(products, target)
    results = {}
    for candidate in candidates:
        results[candidate] = "FAIL (direct parent)" if candidate in parents else "PASS"
    return results


if __name__ == "__main__":
    products = load_manifest("products.yaml")

    print("Case 1 (direct leak, should FAIL):")
    for col, verdict in check_naive(
        products,
        "sustainability_score",
        ["energy_intensity_kwh_sqft", "water_intensity_gal_sqft", "waste_intensity_lbs_sqft"],
    ).items():
        print(f"  {col}: {verdict}")

    print("\nCase 2 (descendant leak, should ALSO fail but watch what happens):")
    for col, verdict in check_naive(
        products,
        "sustainability_score",
        ["annual_energy_kwh", "annual_water_gal", "annual_waste_lbs", "square_footage"],
    ).items():
        print(f"  {col}: {verdict}")

python leak_linter_naive.py

Real output:

Case 1 (direct leak, should FAIL):
  energy_intensity_kwh_sqft: FAIL (direct parent)
  water_intensity_gal_sqft: FAIL (direct parent)
  waste_intensity_lbs_sqft: FAIL (direct parent)

Case 2 (descendant leak, should ALSO fail but watch what happens):
  annual_energy_kwh: PASS
  annual_water_gal: PASS
  annual_waste_lbs: PASS
  square_footage: PASS

This is the exact same Case 2 that measured a real R² of 0.77 back in Step 2. The naive linter waves all four columns through clean. It only ever looks one hop up the graph, and annual_energy_kwh is two hops from sustainability_score, not one, it is a parent of energy_intensity_kwh_sqft, which is itself a parent of sustainability_score. A linter that only checks direct parents will always miss this shape of leak, no matter how carefully you write the manifest. The manifest was correct. The traversal was not.

Step 5: Fix it by walking the whole graph

The fix is to compute the full ancestor closure of the target, not just its direct parents, using breadth-first search. While you are at it, also compute the full descendant closure, because leakage can run the other way too: if someone tries to predict sustainability_score using sustainability_grade as a feature, that is still a leak, just in reverse, since the grade was built from the score, not the other way around.

Create leak_linter.py:

"""leak_linter.py: walk the full dependency graph (not just one hop) to catch
both direct and descendant leaks, refuse malformed manifests, and fail loudly
on typos and empty inputs instead of silently passing them."""
import sys
from collections import deque

import yaml


class StrictLoader(yaml.SafeLoader):
    """A YAML loader that raises on duplicate mapping keys instead of silently
    keeping the last one, which is PyYAML's default (and dangerous) behavior."""


def _no_duplicates_constructor(loader, node, deep=False):
    mapping = {}
    for key_node, value_node in node.value:
        key = loader.construct_object(key_node, deep=deep)
        if key in mapping:
            raise ValueError(
                f"Duplicate key {key!r} in YAML mapping at line {key_node.start_mark.line + 1}"
            )
        mapping[key] = loader.construct_object(value_node, deep=deep)
    return mapping


StrictLoader.add_constructor(
    yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _no_duplicates_constructor
)


class LeakLinterError(Exception):
    """Raised for malformed manifests or bad inputs, never for a detected leak."""


def load_manifest(path):
    with open(path) as f:
        data = yaml.load(f, Loader=StrictLoader)
    if not data or "products" not in data:
        raise LeakLinterError(f"{path} has no top-level 'products' key")
    return data["products"]


def _parents(products, node):
    entry = products.get(node, {})
    return [(edge["from"], edge.get("relation", "unknown")) for edge in entry.get("derived_from", [])]


def _children(products, node):
    out = []
    for name, entry in products.items():
        for edge in entry.get("derived_from", []):
            if edge["from"] == node:
                out.append((name, edge.get("relation", "unknown")))
    return out


def _walk(products, start, direction):
    """BFS over the full graph. direction='up' follows parents (ancestors),
    direction='down' follows children (descendants). Returns {node: route},
    where route is the list of (node, relation) steps from `start` to `node`."""
    edge_fn = _parents if direction == "up" else _children
    found = {}
    queue = deque([(start, [])])
    seen = {start}
    while queue:
        current, route = queue.popleft()
        for neighbor, relation in edge_fn(products, current):
            if neighbor in seen:
                continue
            seen.add(neighbor)
            new_route = route + [(neighbor, relation)]
            found[neighbor] = new_route
            queue.append((neighbor, new_route))
    return found


def _format_route(start, route):
    chain = " -> ".join([start] + [step[0] for step in route])
    relations = ", ".join(sorted({step[1] for step in route}))
    return f"{chain}  (via: {relations})"


def audit(products, target, candidates, known_columns=None):
    """Check each candidate against target. Returns a dict of
    {candidate: {"verdict": "FAIL"|"PASS"|"UNTRACED", "detail": str}}.
    Raises LeakLinterError for malformed input rather than misreporting it."""
    if not candidates:
        raise LeakLinterError("no candidates given -- an empty list checks nothing")

    if known_columns is not None:
        unknown = [c for c in [target] + list(candidates) if c not in known_columns]
        if unknown:
            raise LeakLinterError(f"not a real column in this dataset: {unknown}")

    if target not in products:
        raise LeakLinterError(f"target {target!r} is not documented in the manifest")

    ancestors = _walk(products, target, "up")
    descendants = _walk(products, target, "down")

    results = {}
    for candidate in candidates:
        if candidate not in products:
            results[candidate] = {
                "verdict": "UNTRACED",
                "detail": f"{candidate!r} is not in the manifest -- nobody has documented what it is",
            }
        elif candidate in ancestors:
            results[candidate] = {
                "verdict": "FAIL",
                "detail": "leaks the target -- " + _format_route(target, ancestors[candidate]),
            }
        elif candidate in descendants:
            results[candidate] = {
                "verdict": "FAIL",
                "detail": "was built FROM the target -- " + _format_route(target, descendants[candidate]),
            }
        else:
            results[candidate] = {"verdict": "PASS", "detail": "not on any derivation path to or from the target"}
    return results


def print_report(target, results):
    print(f"Target: {target}")
    worst = "PASS"
    for candidate, r in results.items():
        print(f"  [{r['verdict']:8}] {candidate}: {r['detail']}")
        if r["verdict"] == "FAIL":
            worst = "FAIL"
        elif r["verdict"] == "UNTRACED" and worst != "FAIL":
            worst = "UNTRACED"
    return worst

Look closely at _walk. It is the same BFS shape described in Key Terms: a queue, a seen set, and a loop that keeps pulling from the front and pushing new neighbors to the back. The only difference between finding ancestors and finding descendants is which direction you follow the edges, up toward parents or down toward children. Everything else, the queue logic, the visited-tracking, the route-building, is identical, which is exactly why it is written once as _walk and called twice.

Step 6: Run the linter on real cases

Add a small driver at the bottom of leak_linter.py:

if __name__ == "__main__":
    import pandas as pd

    products = load_manifest("products.yaml")
    df_columns = set(pd.read_csv("stores.csv", nrows=0).columns)

    print("=== Case 1: direct leak ===")
    r1 = audit(
        products,
        "sustainability_score",
        ["energy_intensity_kwh_sqft", "water_intensity_gal_sqft", "waste_intensity_lbs_sqft"],
        df_columns,
    )
    v1 = print_report("sustainability_score", r1)

    print("\n=== Case 2: descendant leak (the naive linter missed this) ===")
    r2 = audit(
        products,
        "sustainability_score",
        ["annual_energy_kwh", "annual_water_gal", "annual_waste_lbs", "square_footage"],
        df_columns,
    )
    v2 = print_report("sustainability_score", r2)

    print("\n=== Case 3: PASS and UNTRACED ===")
    r3 = audit(
        products,
        "sustainability_score",
        ["annual_revenue_usd", "store_manager_tenure_years"],
        df_columns,
    )
    v3 = print_report("sustainability_score", r3)

    print("\n=== Grade check: mixed relation types across two hops ===")
    r4 = audit(products, "sustainability_grade", ["energy_intensity_kwh_sqft"], df_columns)
    v4 = print_report("sustainability_grade", r4)

    verdicts = {"Case 1": v1, "Case 2": v2, "Case 3": v3, "Grade check": v4}
    print("\nSummary:", verdicts)

    exit_code = 1 if any(v == "FAIL" for v in verdicts.values()) else 0
    print(f"exit code: {exit_code}")
    sys.exit(exit_code)

python leak_linter.py
echo "exit code was: $?"

Real output:

=== Case 1: direct leak ===
Target: sustainability_score
  [FAIL    ] energy_intensity_kwh_sqft: leaks the target -- sustainability_score -> energy_intensity_kwh_sqft  (via: weighted_sum)
  [FAIL    ] water_intensity_gal_sqft: leaks the target -- sustainability_score -> water_intensity_gal_sqft  (via: weighted_sum)
  [FAIL    ] waste_intensity_lbs_sqft: leaks the target -- sustainability_score -> waste_intensity_lbs_sqft  (via: weighted_sum)

=== Case 2: descendant leak (the naive linter missed this) ===
Target: sustainability_score
  [FAIL    ] annual_energy_kwh: leaks the target -- sustainability_score -> energy_intensity_kwh_sqft -> annual_energy_kwh  (via: ratio, weighted_sum)
  [FAIL    ] annual_water_gal: leaks the target -- sustainability_score -> water_intensity_gal_sqft -> annual_water_gal  (via: ratio, weighted_sum)
  [FAIL    ] annual_waste_lbs: leaks the target -- sustainability_score -> waste_intensity_lbs_sqft -> annual_waste_lbs  (via: ratio, weighted_sum)
  [FAIL    ] square_footage: leaks the target -- sustainability_score -> energy_intensity_kwh_sqft -> square_footage  (via: ratio, weighted_sum)

=== Case 3: PASS and UNTRACED ===
Target: sustainability_score
  [PASS    ] annual_revenue_usd: not on any derivation path to or from the target
  [UNTRACED] store_manager_tenure_years: 'store_manager_tenure_years' is not in the manifest -- nobody has documented what it is

=== Grade check: mixed relation types across two hops ===
Target: sustainability_grade
  [FAIL    ] energy_intensity_kwh_sqft: leaks the target -- sustainability_grade -> sustainability_score -> energy_intensity_kwh_sqft  (via: binning, weighted_sum)

Summary: {'Case 1': 'FAIL', 'Case 2': 'FAIL', 'Case 3': 'UNTRACED', 'Grade check': 'FAIL'}
exit code: 1

Case 2 now correctly fails, with a printed chain showing exactly why: sustainability_score -> energy_intensity_kwh_sqft -> annual_energy_kwh, tagged with both relation types it passed through along the way. That chain is the whole point. A verdict of FAIL with no explanation just tells a reader to trust the tool. A verdict of FAIL with a printed route lets them check the tool’s work in about two seconds.

Notice also the difference between the two non-failing verdicts in Case 3. annual_revenue_usd gets PASS, because it is documented in the manifest and genuinely has no path to the target. store_manager_tenure_years gets UNTRACED, a different verdict, because the linter has no record of it at all. Collapsing those two into one “not a problem” bucket would be a mistake: PASS means “checked, and safe.” UNTRACED means “not checked, because nobody wrote it down.” Those are very different claims, and a tool that reports them identically is quietly telling its users something it does not actually know.

The grade check shows the traversal working across mixed relation types in the same walk. energy_intensity_kwh_sqft is two hops from sustainability_grade, through a weighted_sum step and then a binning step, and the linter reports both, in the order it found them.

Step 7: Make the linter fail loudly on bad input

A checking tool that silently mishandles bad input is worse than no tool at all, because it gives everyone false confidence. Three specific traps are worth testing deliberately.

Trap 1: a typo in a candidate name

from leak_linter import load_manifest, audit, LeakLinterError
import pandas as pd

products = load_manifest("products.yaml")
df_columns = set(pd.read_csv("stores.csv", nrows=0).columns)

try:
    audit(products, "sustainability_score", ["energy_intensty_kwh_sqft"], df_columns)
except LeakLinterError as e:
    print("raised LeakLinterError:", e)

Real output:

raised LeakLinterError: not a real column in this dataset: ['energy_intensty_kwh_sqft']

Compare this to what would happen if the linter only checked candidates against the manifest and silently treated an unrecognized name as UNTRACED. A typo like energy_intensty_kwh_sqft (missing an “i”) would still get flagged as UNTRACED, which sounds safe enough that a reader in a hurry might not notice it is actually a typo of a column that would have FAILed. Checking candidate names against the real DataFrame columns, not just the manifest, catches this specific failure mode before it can hide inside a merely cautious-sounding verdict.

Trap 2: a duplicate key in the manifest

Try loading a manifest with the same key defined twice:

import yaml

text = """
products:
  a: 1
  b: 2
  a: 999
"""
result = yaml.safe_load(text)
print("safe_load result:", result)

safe_load result: {'products': {'a': 999, 'b': 2}}

This is PyYAML’s real, default behavior, confirmed directly in the environment used for this tutorial: yaml.safe_load silently keeps the last value for a duplicate key and gives no warning of any kind. Picture what that means for a manifest. If two engineers both add a derived_from block for the same column name in the same pull request, maybe one intends it as a rename and forgets to delete the old block, the file will load without complaint, and whichever definition happens to come second in the file quietly wins. The other one, and whatever correct derivation info it held, is gone with no error, no warning, nothing in a diff review to flag it beyond noticing the duplicate key by eye.

The StrictLoader class defined at the top of leak_linter.py fixes this by overriding how PyYAML constructs mappings, raising the moment it sees the same key twice:

from leak_linter import load_manifest

try:
    load_manifest("products_broken.yaml")  # a copy of products.yaml with square_footage defined twice
except ValueError as e:
    print("raised ValueError:", e)

raised ValueError: Duplicate key 'square_footage' in YAML mapping at line 8

It even reports the line number, because StrictLoader reads it straight from the YAML parser’s own node metadata (key_node.start_mark.line) rather than needing you to hunt for it.

Trap 3: an empty candidate list

from leak_linter import load_manifest, audit, LeakLinterError
import pandas as pd

products = load_manifest("products.yaml")
df_columns = set(pd.read_csv("stores.csv", nrows=0).columns)

try:
    audit(products, "sustainability_score", [], df_columns)
except LeakLinterError as e:
    print("raised LeakLinterError:", e)

raised LeakLinterError: no candidates given -- an empty list checks nothing

This one matters more than it looks. Imagine a CI job that reads a team’s proposed feature list from a config file, then calls audit() with whatever it finds. If a bug elsewhere in that pipeline produces an empty list (a bad glob pattern, an off-by-one slice, a config key that silently defaulted to nothing), a version of audit() that just loops over an empty list and returns an empty result set would report zero failures. Zero failures reads, at a glance, exactly like “everything passed,” when what actually happened is “nothing was checked at all.” Raising here converts a silent, misleading success into a loud, honest failure.

Step 8: Run the check automatically in CI

A linter that only runs when someone remembers to run it manually will eventually not get run. Wire it into GitHub Actions so it runs on every push. Create .github/workflows/leak-check.yml:

name: Target Leakage Check

on: [push, pull_request]

jobs:
  leak-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install pandas scikit-learn pyyaml
      - run: python leak_linter.py

The exit code from Step 6 is what makes this work. GitHub Actions treats any non-zero exit code from a run step as a failed job, which is exactly why leak_linter.py ends with sys.exit(1) whenever any candidate FAILs, and sys.exit(0) otherwise. A pull request that introduces a new leaked feature into a model script gets a red X on GitHub before anyone has to notice it by eye in review.

One production note: the workflow above references third-party actions by a floating tag (@v4, @v5), which is fine for following along here but is its own supply-chain risk in a real pipeline, since a floating tag can be moved to point at different code after you have already reviewed it. If you want to close that gap, this site’s own tutorial on pinning and verifying GitHub Actions walks through building a small Python tool that resolves every floating tag to its exact commit SHA and checks it stays that way.

Common mistakes and gotchas

• Checking only direct parents. This is the bug from Step 4. If your manifest is more than two levels deep anywhere, a one-hop check will miss real leaks. Always compute the full ancestor and descendant closure, not just the immediate neighbors.
• Treating “not in the manifest” as safe. An undocumented column is not a verified column. Report it as its own verdict (UNTRACED here) instead of silently merging it into PASS.
• Only checking one direction. A column built from your target is just as much a leak as a column your target was built from. If your target is being predicted by something computed after the fact from the target itself, that will not exist at real prediction time either.
• Trusting a YAML parser’s default duplicate-key behavior. Confirmed directly in Step 7: PyYAML’s safe_load keeps the last value and says nothing. If your manifest format matters, use a loader that rejects duplicates instead of assuming your reviewers will always catch a copy-paste mistake by eye.
• Letting empty input report success. A check that runs against zero candidates and returns zero failures is not the same thing as a check that ran and found nothing wrong. Make the difference between those two outcomes impossible to miss.

Verifying it all works end to end

Run through this checklist in a fresh clone or a fresh directory to confirm the whole pipeline holds together:

1. Run python generate_data.py and confirm stores.csv has 400 rows and 11 columns.
2. Run python leak_demo.py and confirm Case 1’s mean R² is well above 0.9, Case 2’s is well above the decoy’s, and the decoy’s is at or below 0.
3. Run python leak_linter_naive.py and confirm Case 2 prints all PASS, reproducing the bug.
4. Run python leak_linter.py and confirm Case 2 now prints all FAIL, with a printed derivation chain for each column, and that the process exits with code 1.
5. Manually trigger each of the three traps from Step 7 and confirm each one raises an exception with a message that explains what went wrong, rather than returning quietly.

Finally, add test_leak_linter.py and run it with pytest to lock this behavior in as a regression suite rather than something you have to re-check by hand every time you touch the graph logic:

"""test_leak_linter.py: regression tests for the leak linter's graph logic."""
import pytest

from leak_linter import audit, load_manifest, LeakLinterError, StrictLoader
import yaml

PRODUCTS = load_manifest("products.yaml")
COLUMNS = {
    "square_footage", "annual_energy_kwh", "annual_water_gal", "annual_waste_lbs",
    "annual_revenue_usd", "store_manager_tenure_years", "energy_intensity_kwh_sqft",
    "water_intensity_gal_sqft", "waste_intensity_lbs_sqft", "sustainability_score",
    "sustainability_grade",
}


def test_direct_parent_fails():
    r = audit(PRODUCTS, "sustainability_score", ["energy_intensity_kwh_sqft"], COLUMNS)
    assert r["energy_intensity_kwh_sqft"]["verdict"] == "FAIL"


def test_descendant_two_hops_fails():
    # This is the exact case the naive, direct-parents-only checker missed.
    r = audit(PRODUCTS, "sustainability_score", ["annual_energy_kwh"], COLUMNS)
    assert r["annual_energy_kwh"]["verdict"] == "FAIL"


def test_independent_column_passes():
    r = audit(PRODUCTS, "sustainability_score", ["annual_revenue_usd"], COLUMNS)
    assert r["annual_revenue_usd"]["verdict"] == "PASS"


def test_undocumented_column_is_untraced_not_pass():
    r = audit(PRODUCTS, "sustainability_score", ["store_manager_tenure_years"], COLUMNS)
    assert r["store_manager_tenure_years"]["verdict"] == "UNTRACED"


def test_reverse_direction_also_fails():
    # sustainability_grade was built FROM sustainability_score, so using the
    # grade to help predict the score is leakage in the other direction.
    r = audit(PRODUCTS, "sustainability_score", ["sustainability_grade"], COLUMNS)
    assert r["sustainability_grade"]["verdict"] == "FAIL"


def test_mixed_relation_chain_across_two_hops():
    r = audit(PRODUCTS, "sustainability_grade", ["energy_intensity_kwh_sqft"], COLUMNS)
    assert r["energy_intensity_kwh_sqft"]["verdict"] == "FAIL"
    assert "binning" in r["energy_intensity_kwh_sqft"]["detail"]
    assert "weighted_sum" in r["energy_intensity_kwh_sqft"]["detail"]


def test_empty_candidate_list_raises():
    with pytest.raises(LeakLinterError, match="empty list"):
        audit(PRODUCTS, "sustainability_score", [], COLUMNS)


def test_typo_raises_instead_of_silently_untracing():
    with pytest.raises(LeakLinterError, match="not a real column"):
        audit(PRODUCTS, "sustainability_score", ["sustainabilty_score_TYPO"], COLUMNS)


def test_unknown_target_raises():
    with pytest.raises(LeakLinterError, match="not documented"):
        audit(PRODUCTS, "some_column_nobody_defined", ["square_footage"], COLUMNS | {"some_column_nobody_defined"})


def test_duplicate_keys_are_rejected():
    text = """
    products:
      a:
        description: "first"
      b:
        description: "second"
      a:
        description: "duplicate, should raise"
    """
    with pytest.raises(ValueError, match="Duplicate key"):
        yaml.load(text, Loader=StrictLoader)

pytest test_leak_linter.py -v

collected 10 items

test_leak_linter.py::test_direct_parent_fails PASSED
test_leak_linter.py::test_descendant_two_hops_fails PASSED
test_leak_linter.py::test_independent_column_passes PASSED
test_leak_linter.py::test_undocumented_column_is_untraced_not_pass PASSED
test_leak_linter.py::test_reverse_direction_also_fails PASSED
test_leak_linter.py::test_mixed_relation_chain_across_two_hops PASSED
test_leak_linter.py::test_empty_candidate_list_raises PASSED
test_leak_linter.py::test_typo_raises_instead_of_silently_untracing PASSED
test_leak_linter.py::test_unknown_target_raises PASSED
test_leak_linter.py::test_duplicate_keys_are_rejected PASSED

10 passed in 0.04s

Next steps

The manifest and linter in this tutorial cover the core mechanic: describe derivation as a graph, walk the whole graph, fail loudly on bad input. A few directions worth exploring once this is running comfortably for your own data:

• Generate the manifest’s node list automatically from your DataFrame’s own column names, so new columns at least show up as UNTRACED by default instead of silently having no entry at all.
• Extend the relation field to carry a severity, and let audit() distinguish “definitely leaks” from “correlated but worth a human look,” similar in spirit to how this site’s Bandit pre-commit tutorial reports findings at different confidence levels rather than one flat pass/fail.
• Before reaching for a model at all on a new public dataset, it is worth understanding what you are looking at with plain descriptive statistics first. This site’s tutorial on descriptive statistics in Python covers that groundwork, including its own lesson in not trusting a single summary number, in that case an average, before checking what is actually driving it.

Sources

• scikit-learn: Common pitfalls and recommended practices, Data leakage
• CDC/ATSDR: Social Vulnerability Index documentation
• PyYAML documentation
• GitHub Actions: Events that trigger workflows