Regulatory Data Attestation Pipeline

Build a two-layer data quality gate for Basel III capital adequacy data.

Introduction

30 Second Summary

Every quarter, banks file capital reports proving they have enough money to absorb losses. Before those numbers reach the regulator, a data team has to prove every figure is accurate and complete.

In this project, you will build a data quality attestation pipeline using Great Expectations. You will validate synthetic Basel III capital adequacy data through two quality gates and produce audit-ready HTML reports as evidence.

What You'll Build

You will build a Python-based regulatory data attestation pipeline that validates simulated Basel III capital adequacy data through a two-layer quality gate system powered by Great Expectations.

By the end of this project, you'll have:

  • A synthetic Basel III dataset with 10,000 rows of capital adequacy records and deliberate anomalies for testing.
  • A two-layer quality gate system with ~45 expectations covering raw data checks and regulatory business rules.
  • An HTML attestation report generated by Data Docs with timestamped validation results that serve as audit-ready evidence.
  • Secret Mission: An anomaly quarantine system that isolates bad records, tags them with failure reasons, and produces a passing attestation report.

Are there any prerequisites?

You should be comfortable writing Python scripts and working with Pandas DataFrames. Familiarity with financial concepts like capital ratios is helpful but not required.

Before We Start

Before a bank can submit a regulatory capital report, the data behind it must pass rigorous quality checks. If bad data slips through, the consequences range from regulatory fines to a full restatement of the bank's capital position.

In this step, you will set up the project environment for building a data quality attestation pipeline. You will install Great Expectations, the open-source Python framework that powers these quality checks, and create the files you will work with throughout the project.

In this step, get ready to:
  • Create a project directory and Python virtual environment.
  • Install Great Expectations.
  • Create the project file structure.
Create your project directory and virtual environment

A virtual environment gives your project its own isolated set of Python packages. This prevents version conflicts between different projects on your machine.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir regulatory-attestation-pipeline
cd regulatory-attestation-pipeline
  • Create a Python virtual environment by running:
python -m venv venv

Getting a "command not found" error?

On macOS, the python command may not be recognized. Try python3 instead. If that works, use python3 for all Python commands throughout this project.

Help me figure out why my Python command is not working.

  • Activate the virtual environment by running the command for your operating system:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

Why use a virtual environment?

Great Expectations pulls in several dependencies like pandas and numpy. A virtual environment keeps these packages isolated to this project so they do not interfere with other Python work on your machine.

Install Great Expectations and verify

Great Expectations (GX Core) is the open-source data quality framework you will use to define expectations about your data and validate it. It requires Python 3.10 or higher.

  • Install Great Expectations by running:
pip install great_expectations

What is happening during install?

This command downloads Great Expectations and all its dependencies from PyPI (the Python Package Index). The install may take a minute or two as it pulls in packages like pandas, numpy, and others.

  • Verify the installation by running:
python -c "import great_expectations as gx; print(gx.__version__)"

✔️ I see a version number

Great Expectations is installed and ready to go. You should see something like 1.18.1 printed in your terminal.

ⓧ I see an error

If you see an error mentioning a missing module, the install may not have completed successfully. Try the following:

  • Check that (venv) appears in your terminal prompt. If it does not, re-run the activation command from the previous substep.
  • If your virtual environment is active, try running the install command again:
pip install great_expectations
  • Run the verify command again after the install completes.

Help me troubleshoot my Great Expectations install.

Create the project file structure

Your project needs four Python files. Each file handles a different part of the attestation pipeline that you will build in the upcoming steps.

  • Create the four project files by running the command for your operating system:

macOS

touch generate_data.py raw_gate.py business_rule_gate.py run_pipeline.py

Windows

New-Item -ItemType File -Path generate_data.py, raw_gate.py, business_rule_gate.py, run_pipeline.py
  • Confirm the files exist by running:
ls

You should see generate_data.py, raw_gate.py, business_rule_gate.py, and run_pipeline.py listed alongside the venv folder.

What is each file for?

  • generate_data.py creates synthetic Basel III capital adequacy data with deliberate anomalies.
  • raw_gate.py validates the raw data at ingestion (Layer 1 quality gate).
  • business_rule_gate.py validates regulatory business rules after transformation (Layer 2 quality gate).
  • run_pipeline.py orchestrates both gates and generates the attestation report.

Don't see all four files?

Make sure you ran the file creation command from inside the regulatory-attestation-pipeline folder. You can check your current location by running pwd (macOS) or cd (Windows). It should end with regulatory-attestation-pipeline.

Help me figure out why my project files were not created.

Your environment is ready. Next up, you will generate 10,000 rows of synthetic Basel III data with deliberate anomalies baked in.

Generate Synthetic Basel III Data

Your environment is ready with Great Expectations installed and four empty Python files waiting to be filled. Now it is time to create the data your pipeline will validate.

Before a bank submits a Basel III capital adequacy report to regulators, the underlying data must be verified as correct. In this step, you will generate 10,000 rows of synthetic financial data and deliberately inject anomalies so the quality gates you build later have something meaningful to catch.

In this step, get ready to:
  • Build a data generator that creates realistic Basel III capital adequacy records.
  • Inject deliberate anomalies that mimic real-world data quality problems.
  • Run the generator and inspect the output to confirm everything looks right.
Build the data generator

The generate_clean_data() function creates a Pandas DataFrame with 10,000 rows of synthetic bank data. Each row represents one entity's capital report, with columns for capital amounts, risk-weighted assets, and the ratios regulators actually examine.

  • In your code editor, open generate_data.py from the regulatory-attestation-pipeline folder.
  • Copy the following imports and generate_clean_data() function into the file:
import numpy as np
import pandas as pd

def generate_clean_data(n_rows=10000, seed=42):
    """Generate synthetic Basel III capital adequacy data."""
    rng = np.random.default_rng(seed)

    n_entities = 50
    entity_ids = [f"BANK_{i:03d}" for i in range(1, n_entities + 1)]
    country_codes = ["US", "GB", "DE", "JP", "CH", "FR", "SG", "HK", "AU", "CA"]
    entity_types = ["G-SIB", "D-SIB", "Regional", "Commercial"]

    data = {
        "entity_id": rng.choice(entity_ids, size=n_rows),
        "reporting_date": "2026-03-31",
        "country_code": rng.choice(country_codes, size=n_rows),
        "entity_type": rng.choice(entity_types, size=n_rows),
        "rwa": rng.uniform(5000, 500000, size=n_rows).round(2),
        "exposure_amount": rng.uniform(10000, 1000000, size=n_rows).round(2),
    }

    data["cet1_capital"] = (data["rwa"] * rng.uniform(0.06, 0.18, size=n_rows)).round(2)
    data["tier1_capital"] = (data["cet1_capital"] * rng.uniform(1.0, 1.3, size=n_rows)).round(2)
    data["total_capital"] = (data["tier1_capital"] * rng.uniform(1.0, 1.25, size=n_rows)).round(2)

    data["cet1_ratio"] = (data["cet1_capital"] / data["rwa"]).round(6)
    data["tier1_ratio"] = (data["tier1_capital"] / data["rwa"]).round(6)
    data["total_capital_ratio"] = (data["total_capital"] / data["rwa"]).round(6)
    data["leverage_ratio"] = (data["tier1_capital"] / data["exposure_amount"]).round(6)

    df = pd.DataFrame(data)
    column_order = [
        "entity_id", "reporting_date", "country_code", "entity_type",
        "cet1_capital", "tier1_capital", "total_capital", "rwa",
        "cet1_ratio", "tier1_ratio", "total_capital_ratio",
        "leverage_ratio", "exposure_amount"
    ]
    return df[column_order]

What does this code do?

The function uses NumPy's default_rng(seed=42) to create a random number generator with a fixed seed. Every time you run the script, you get the exact same 10,000 rows. Reproducibility matters in regulatory reporting because auditors need to verify your results.

  • entity_ids creates 50 unique bank identifiers (BANK_001 through BANK_050), randomly assigned across 10,000 rows.
  • Capital columns (cet1_capital, tier1_capital, total_capital) are derived so each is always >= the previous. CET1 is a subset of Tier 1, which is a subset of Total Capital.
  • Ratios like cet1_ratio are calculated by dividing capital by rwa (risk-weighted assets). These are the numbers regulators actually scrutinize.
  • Save generate_data.py.
  • You won't see a change yet. The next chunk adds the anomaly injection function.
Inject deliberate anomalies

Clean data is not very useful for testing a quality gate. In the real world, regulatory datasets arrive with nulls, invalid codes, and broken relationships between columns.

  • In generate_data.py, add the following inject_anomalies() function below the generate_clean_data() function:

def inject_anomalies(df, seed=42):
    """Inject ~5% anomalies into the dataset."""
    rng = np.random.default_rng(seed)
    df_dirty = df.copy()
    n_rows = len(df_dirty)

    # Null values in critical columns (~50 total)
    null_indices_entity = rng.choice(n_rows, size=15, replace=False)
    null_indices_cet1 = rng.choice(n_rows, size=20, replace=False)
    null_indices_rwa = rng.choice(n_rows, size=15, replace=False)
    df_dirty.loc[null_indices_entity, "entity_id"] = None
    df_dirty.loc[null_indices_cet1, "cet1_capital"] = np.nan
    df_dirty.loc[null_indices_rwa, "rwa"] = np.nan

    # Invalid country codes (~100 rows)
    invalid_cc_indices = rng.choice(n_rows, size=100, replace=False)
    invalid_codes = rng.choice(["XX", "ZZ", "123", "??"], size=100)
    df_dirty.loc[invalid_cc_indices, "country_code"] = invalid_codes

    # Negative capital values (~150 rows)
    neg_cet1_indices = rng.choice(n_rows, size=75, replace=False)
    neg_tier1_indices = rng.choice(n_rows, size=75, replace=False)
    df_dirty.loc[neg_cet1_indices, "cet1_capital"] = -rng.uniform(100, 5000, size=75).round(2)
    df_dirty.loc[neg_tier1_indices, "tier1_capital"] = -rng.uniform(100, 5000, size=75).round(2)

    # Tier1 ratio < CET1 ratio violations (~50 rows)
    ratio_violation_indices = rng.choice(n_rows, size=50, replace=False)
    df_dirty.loc[ratio_violation_indices, "tier1_ratio"] = (
        df_dirty.loc[ratio_violation_indices, "cet1_ratio"] * rng.uniform(0.3, 0.9, size=50)
    ).round(6)

    # Total capital ratio below 8% floor (~30 rows)
    floor_violation_indices = rng.choice(n_rows, size=30, replace=False)
    df_dirty.loc[floor_violation_indices, "total_capital_ratio"] = rng.uniform(0.02, 0.07, size=30).round(6)

    # Invalid entity types (~20 rows)
    invalid_type_indices = rng.choice(n_rows, size=20, replace=False)
    df_dirty.loc[invalid_type_indices, "entity_type"] = rng.choice(["Unknown", "TEST", "N/A"], size=20)

    return df_dirty

What anomalies are we injecting?

The function injects six categories of data quality problems. Each one targets a specific kind of validation rule your gates will check later.

  • ~50 null values scattered across entity_id, cet1_capital, and rwa. These test completeness checks.
  • ~100 invalid country codes (XX, ZZ, 123). These test domain validation.
  • ~150 negative capital values. Capital can never be negative in a real report.
  • ~50 rows where tier1_ratio < cet1_ratio. This is mathematically impossible because CET1 capital is a subset of Tier 1.
  • ~30 rows where total_capital_ratio falls below the Basel III 8% minimum floor.
  • ~20 rows with invalid entity types like Unknown or TEST.
  • Save generate_data.py.
  • You won't see a change yet. The next chunk adds the script's entry point so you can run it.

Why inject anomalies on purpose?

Real regulatory data always has quality issues. By injecting known problems, you can verify that your quality gates actually catch them.

If your gates pass on dirty data, they are not strict enough. This controlled injection gives you a ground truth to calibrate against.

Run the generator and inspect the output

The final piece is the entry point that calls both functions, saves the results to CSV files, and prints a quick summary. This lets you confirm the data looks realistic before building quality gates on top of it.

  • In generate_data.py, add the following block at the bottom of the file, below the inject_anomalies() function:

if __name__ == "__main__":
    print("Generating synthetic Basel III data...")
    clean_df = generate_clean_data()
    clean_df.to_csv("regulatory_data_clean.csv", index=False)
    print(f"Clean data saved: {len(clean_df)} rows, {len(clean_df.columns)} columns")

    dirty_df = inject_anomalies(clean_df)
    dirty_df.to_csv("regulatory_data_raw.csv", index=False)
    print(f"Raw data (with anomalies) saved: {len(dirty_df)} rows")

    print("\nSample data (first 5 rows):")
    print(dirty_df.head().to_string())
    print(f"\nNull counts:\n{dirty_df.isnull().sum()}")
    print(f"\nNegative cet1_capital count: {(dirty_df['cet1_capital'] < 0).sum()}")
    print(f"Invalid country codes: {dirty_df[~dirty_df['country_code'].isin(['US','GB','DE','JP','CH','FR','SG','HK','AU','CA'])].shape[0]}")

What does this entry point do?

  • It calls generate_clean_data() to create 10,000 clean rows, then saves them to regulatory_data_clean.csv. This clean copy is your baseline for comparison.
  • It calls inject_anomalies() on the clean data and saves the result to regulatory_data_raw.csv. This is the file your quality gates will validate.
  • It prints null counts, negative value counts, and invalid country code counts so you can verify the anomalies were injected correctly.
  • Save generate_data.py.
  • Run the generator by entering this command in your terminal:
python generate_data.py

You should see output confirming 10,000 rows were saved for both the clean and raw files. The summary also prints null counts and negative value counts so you can verify anomalies were injected.

Seeing a ModuleNotFoundError?

  • If you see an error mentioning numpy or pandas, make sure your virtual environment is still activated. You should see (venv) at the start of your terminal prompt.
  • Both numpy and pandas are installed automatically when you install great_expectations. If they are missing, re-run pip install great_expectations with your virtual environment activated.
  • Help me fix this ModuleNotFoundError
  • Confirm the two CSV files were created by running:
ls *.csv

You should see regulatory_data_clean.csv and regulatory_data_raw.csv in the output.

Don't see both CSV files?

  • Make sure you are running the command from inside the regulatory-attestation-pipeline folder. Run pwd to check your current directory.
  • Check the terminal output from python generate_data.py for any error messages. A syntax error in the file will prevent the CSVs from being created.
  • Help me troubleshoot my missing CSV files

✔️ Awesome, I've got everything!

Great. Double check that you have saved generate_data.py and that both CSV files exist in your project folder.

ⓧ I'd like to double check the full code

Here is the complete generate_data.py file. Compare it with yours to spot any differences.

import numpy as np
import pandas as pd

def generate_clean_data(n_rows=10000, seed=42):
    """Generate synthetic Basel III capital adequacy data."""
    rng = np.random.default_rng(seed)

    n_entities = 50
    entity_ids = [f"BANK_{i:03d}" for i in range(1, n_entities + 1)]
    country_codes = ["US", "GB", "DE", "JP", "CH", "FR", "SG", "HK", "AU", "CA"]
    entity_types = ["G-SIB", "D-SIB", "Regional", "Commercial"]

    data = {
        "entity_id": rng.choice(entity_ids, size=n_rows),
        "reporting_date": "2026-03-31",
        "country_code": rng.choice(country_codes, size=n_rows),
        "entity_type": rng.choice(entity_types, size=n_rows),
        "rwa": rng.uniform(5000, 500000, size=n_rows).round(2),
        "exposure_amount": rng.uniform(10000, 1000000, size=n_rows).round(2),
    }

    data["cet1_capital"] = (data["rwa"] * rng.uniform(0.06, 0.18, size=n_rows)).round(2)
    data["tier1_capital"] = (data["cet1_capital"] * rng.uniform(1.0, 1.3, size=n_rows)).round(2)
    data["total_capital"] = (data["tier1_capital"] * rng.uniform(1.0, 1.25, size=n_rows)).round(2)

    data["cet1_ratio"] = (data["cet1_capital"] / data["rwa"]).round(6)
    data["tier1_ratio"] = (data["tier1_capital"] / data["rwa"]).round(6)
    data["total_capital_ratio"] = (data["total_capital"] / data["rwa"]).round(6)
    data["leverage_ratio"] = (data["tier1_capital"] / data["exposure_amount"]).round(6)

    df = pd.DataFrame(data)
    column_order = [
        "entity_id", "reporting_date", "country_code", "entity_type",
        "cet1_capital", "tier1_capital", "total_capital", "rwa",
        "cet1_ratio", "tier1_ratio", "total_capital_ratio",
        "leverage_ratio", "exposure_amount"
    ]
    return df[column_order]


def inject_anomalies(df, seed=42):
    """Inject ~5% anomalies into the dataset."""
    rng = np.random.default_rng(seed)
    df_dirty = df.copy()
    n_rows = len(df_dirty)

    null_indices_entity = rng.choice(n_rows, size=15, replace=False)
    null_indices_cet1 = rng.choice(n_rows, size=20, replace=False)
    null_indices_rwa = rng.choice(n_rows, size=15, replace=False)
    df_dirty.loc[null_indices_entity, "entity_id"] = None
    df_dirty.loc[null_indices_cet1, "cet1_capital"] = np.nan
    df_dirty.loc[null_indices_rwa, "rwa"] = np.nan

    invalid_cc_indices = rng.choice(n_rows, size=100, replace=False)
    invalid_codes = rng.choice(["XX", "ZZ", "123", "??"], size=100)
    df_dirty.loc[invalid_cc_indices, "country_code"] = invalid_codes

    neg_cet1_indices = rng.choice(n_rows, size=75, replace=False)
    neg_tier1_indices = rng.choice(n_rows, size=75, replace=False)
    df_dirty.loc[neg_cet1_indices, "cet1_capital"] = -rng.uniform(100, 5000, size=75).round(2)
    df_dirty.loc[neg_tier1_indices, "tier1_capital"] = -rng.uniform(100, 5000, size=75).round(2)

    ratio_violation_indices = rng.choice(n_rows, size=50, replace=False)
    df_dirty.loc[ratio_violation_indices, "tier1_ratio"] = (
        df_dirty.loc[ratio_violation_indices, "cet1_ratio"] * rng.uniform(0.3, 0.9, size=50)
    ).round(6)

    floor_violation_indices = rng.choice(n_rows, size=30, replace=False)
    df_dirty.loc[floor_violation_indices, "total_capital_ratio"] = rng.uniform(0.02, 0.07, size=30).round(6)

    invalid_type_indices = rng.choice(n_rows, size=20, replace=False)
    df_dirty.loc[invalid_type_indices, "entity_type"] = rng.choice(["Unknown", "TEST", "N/A"], size=20)

    return df_dirty


if __name__ == "__main__":
    print("Generating synthetic Basel III data...")
    clean_df = generate_clean_data()
    clean_df.to_csv("regulatory_data_clean.csv", index=False)
    print(f"Clean data saved: {len(clean_df)} rows, {len(clean_df.columns)} columns")

    dirty_df = inject_anomalies(clean_df)
    dirty_df.to_csv("regulatory_data_raw.csv", index=False)
    print(f"Raw data (with anomalies) saved: {len(dirty_df)} rows")

    print("\nSample data (first 5 rows):")
    print(dirty_df.head().to_string())
    print(f"\nNull counts:\n{dirty_df.isnull().sum()}")
    print(f"\nNegative cet1_capital count: {(dirty_df['cet1_capital'] < 0).sum()}")
    print(f"Invalid country codes: {dirty_df[~dirty_df['country_code'].isin(['US','GB','DE','JP','CH','FR','SG','HK','AU','CA'])].shape[0]}")

Your synthetic dataset is ready. Next up, you will build the first quality gate that validates this raw data at ingestion.

Build the Raw Data Gate

You have 10,000 rows of synthetic Basel III data sitting in a CSV file, complete with deliberately injected anomalies. Now it is time to build the first line of defense: a raw data quality gate that catches problems at ingestion.

In regulatory reporting, BCBS 239 Principle 3 requires that data quality is validated at the point of origination, before any transformation happens. This is exactly what your raw gate does. You will use Great Expectations to define 21 expectations that check your data's schema, completeness, valid domains, value ranges, and date formats.

In this step, get ready to:
  • Connect Great Expectations to your raw CSV data using a Pandas data source.
  • Define an Expectation Suite with 21 expectations covering schema, completeness, domain, range, and format checks.
  • Run a Checkpoint and review which expectations pass and which fail due to the injected anomalies.
Set up the GX Data Context and connect to the raw CSV data

Before you can write expectations, you need to tell Great Expectations where your data lives. GX uses a Data Context as its central configuration object, and you connect data through a data source, asset, and batch definition.

  • In raw_gate.py, add the following code to import Great Expectations, define the run_raw_gate function, and connect to the raw data:
import great_expectations as gx
import pandas as pd


def run_raw_gate(df):
    """Run Layer 1 raw data quality gate on the provided DataFrame."""
    context = gx.get_context()

    # Connect to data
    data_source = context.data_sources.add_pandas("pandas_raw")
    data_asset = data_source.add_dataframe_asset(name="raw_regulatory_data")
    batch_definition = data_asset.add_batch_definition_whole_dataframe("raw_batch")
    batch = batch_definition.get_batch(batch_parameters={"dataframe": df})

What does this code do?

gx.get_context() creates an ephemeral (in-memory) Data Context. No files or folders are needed on disk for this to work.

The next three lines build a chain: a Pandas data source named "pandas_raw" holds a DataFrame asset named "raw_regulatory_data", which uses a whole-DataFrame batch definition named "raw_batch". This tells GX to treat the entire DataFrame as one batch of data to validate.

You won't see a change yet. The next chunk defines the expectations this function will check.

Define the raw gate Expectation Suite

An Expectation Suite is a named collection of rules your data must satisfy. Each rule is called an expectation, and you add them to the suite one at a time.

  • Continue building the run_raw_gate function in raw_gate.py by adding the Expectation Suite with all 21 expectations below the batch line:
    # Create Expectation Suite
    suite = context.suites.add(
        gx.core.expectation_suite.ExpectationSuite(name="raw_gate_suite")
    )

    # Schema checks
    suite.add_expectation(
        gx.expectations.ExpectTableColumnsToMatchSet(
            column_set=[
                "entity_id", "reporting_date", "country_code", "entity_type",
                "cet1_capital", "tier1_capital", "total_capital", "rwa",
                "cet1_ratio", "tier1_ratio", "total_capital_ratio",
                "leverage_ratio", "exposure_amount"
            ],
            exact_match=False
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectTableRowCountToBeBetween(min_value=9000, max_value=11000)
    )

    # Completeness checks
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="entity_id", mostly=1.0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="cet1_capital", mostly=0.95)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="rwa", mostly=0.95)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="reporting_date", mostly=1.0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="tier1_capital", mostly=0.95)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="total_capital", mostly=0.95)
    )

    # Domain checks
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeInSet(
            column="country_code",
            value_set=["US", "GB", "DE", "JP", "CH", "FR", "SG", "HK", "AU", "CA"]
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeInSet(
            column="entity_type",
            value_set=["G-SIB", "D-SIB", "Regional", "Commercial"]
        )
    )

    # Range checks - capital values must be non-negative
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(column="cet1_capital", min_value=0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(column="tier1_capital", min_value=0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(column="total_capital", min_value=0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(column="rwa", min_value=0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(column="exposure_amount", min_value=0)
    )

    # Range checks - ratios must be between 0 and 1
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="cet1_ratio", min_value=0, max_value=1
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="tier1_ratio", min_value=0, max_value=1
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="total_capital_ratio", min_value=0, max_value=1
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="leverage_ratio", min_value=0, max_value=1
        )
    )

    # Format checks
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToMatchRegex(
            column="reporting_date",
            regex=r"^\d{4}-\d{2}-\d{2}$"
        )
    )

What do these expectations check?

The 21 expectations are organized into five categories, each catching a different class of data problem:

  • Schema checks confirm all 13 expected columns are present and the row count is between 9,000 and 11,000.
  • Completeness checks use ExpectColumnValuesToNotBeNull to catch missing values. The mostly parameter sets the threshold. For example, mostly=0.95 means at least 95% of values must be non-null.
  • Domain checks verify that country_code and entity_type contain only values from their allowed sets.
  • Range checks ensure capital values are non-negative and ratios fall between 0 and 1.
  • Format checks validate that reporting_date matches the ISO date pattern (YYYY-MM-DD).

Why use the mostly parameter?

Real-world regulatory data is rarely 100% clean at ingestion. The mostly parameter lets you set a practical threshold instead of an all-or-nothing check. For example, mostly=0.95 means the expectation passes if at least 95% of values are valid.

Columns like entity_id use mostly=1.0 because a missing identifier is never acceptable. Other columns allow a small tolerance.

You won't see a change yet. The next chunk wires up the Checkpoint that runs these expectations against the data.

Run the raw gate and review results

Expectations on their own do not validate anything. You need a Checkpoint to actually run the suite against your data batch and produce results.

  • Continue building raw_gate.py by adding the Validation Definition, Checkpoint, print function, and main block below the last suite.add_expectation call:
    # Create Validation Definition and Checkpoint
    validation_definition = context.validation_definitions.add(
        gx.core.validation_definition.ValidationDefinition(
            name="raw_gate_validation",
            data=batch_definition,
            suite=suite,
        )
    )

    checkpoint = context.checkpoints.add(
        gx.checkpoint.checkpoint.Checkpoint(
            name="raw_gate_checkpoint",
            validation_definitions=[validation_definition],
        )
    )

    # Run checkpoint
    result = checkpoint.run(batch_parameters={"dataframe": df})
    return result, context


def print_raw_gate_results(result):
    """Print a human-readable summary of raw gate results."""
    print("\n" + "=" * 60)
    print("LAYER 1: RAW DATA GATE RESULTS")
    print("=" * 60)
    print(result.describe())

    if result.success:
        print("\nRAW GATE: PASSED")
    else:
        print("\nRAW GATE: FAILED - Pipeline would halt here in production")


if __name__ == "__main__":
    df = pd.read_csv("regulatory_data_raw.csv")
    result, context = run_raw_gate(df)
    print_raw_gate_results(result)

What does this code do?

  • The ValidationDefinition links your batch definition (the data) to your expectation suite (the rules). It tells GX what to validate and how.
  • The Checkpoint wraps one or more validation definitions and runs them together. Calling checkpoint.run() executes all the expectations against the data and returns a CheckpointResult.
  • result.describe() prints a human-readable summary showing each expectation's pass/fail status, unexpected value counts, and percentages.
  • The if __name__ block loads the raw CSV and runs the full gate when you execute the script directly.
  • Save raw_gate.py.
  • Run the raw gate by running this command:
python raw_gate.py

You should see a detailed checkpoint result summary. Several expectations will show as failed due to the anomalies you injected earlier. Look for lines showing unexpected_count and unexpected_percent values greater than zero.

The final line of the output should read RAW GATE: FAILED - Pipeline would halt here in production.

Seeing an import error or no output?

  • Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
  • Check that regulatory_data_raw.csv exists in your regulatory-attestation-pipeline folder. If it is missing, run python generate_data.py first.
  • Verify your indentation is consistent. Every line inside the run_raw_gate function must be indented by exactly four spaces.

Still stuck? Help me debug why raw_gate.py is not running correctly.

✔️ Awesome, I've got everything!

Great. Double check you have saved raw_gate.py and confirmed the raw gate reports failed expectations.

ⓧ I'd like to double check the full code

Here is the complete raw_gate.py file. Compare it with yours to spot any differences:

import great_expectations as gx
import pandas as pd


def run_raw_gate(df):
    """Run Layer 1 raw data quality gate on the provided DataFrame."""
    context = gx.get_context()

    # Connect to data
    data_source = context.data_sources.add_pandas("pandas_raw")
    data_asset = data_source.add_dataframe_asset(name="raw_regulatory_data")
    batch_definition = data_asset.add_batch_definition_whole_dataframe("raw_batch")
    batch = batch_definition.get_batch(batch_parameters={"dataframe": df})

    # Create Expectation Suite
    suite = context.suites.add(
        gx.core.expectation_suite.ExpectationSuite(name="raw_gate_suite")
    )

    # Schema checks
    suite.add_expectation(
        gx.expectations.ExpectTableColumnsToMatchSet(
            column_set=[
                "entity_id", "reporting_date", "country_code", "entity_type",
                "cet1_capital", "tier1_capital", "total_capital", "rwa",
                "cet1_ratio", "tier1_ratio", "total_capital_ratio",
                "leverage_ratio", "exposure_amount"
            ],
            exact_match=False
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectTableRowCountToBeBetween(min_value=9000, max_value=11000)
    )

    # Completeness checks
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="entity_id", mostly=1.0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="cet1_capital", mostly=0.95)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="rwa", mostly=0.95)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="reporting_date", mostly=1.0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="tier1_capital", mostly=0.95)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="total_capital", mostly=0.95)
    )

    # Domain checks
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeInSet(
            column="country_code",
            value_set=["US", "GB", "DE", "JP", "CH", "FR", "SG", "HK", "AU", "CA"]
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeInSet(
            column="entity_type",
            value_set=["G-SIB", "D-SIB", "Regional", "Commercial"]
        )
    )

    # Range checks - capital values must be non-negative
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(column="cet1_capital", min_value=0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(column="tier1_capital", min_value=0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(column="total_capital", min_value=0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(column="rwa", min_value=0)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(column="exposure_amount", min_value=0)
    )

    # Range checks - ratios must be between 0 and 1
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="cet1_ratio", min_value=0, max_value=1
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="tier1_ratio", min_value=0, max_value=1
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="total_capital_ratio", min_value=0, max_value=1
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="leverage_ratio", min_value=0, max_value=1
        )
    )

    # Format checks
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToMatchRegex(
            column="reporting_date",
            regex=r"^\d{4}-\d{2}-\d{2}$"
        )
    )

    # Create Validation Definition and Checkpoint
    validation_definition = context.validation_definitions.add(
        gx.core.validation_definition.ValidationDefinition(
            name="raw_gate_validation",
            data=batch_definition,
            suite=suite,
        )
    )

    checkpoint = context.checkpoints.add(
        gx.checkpoint.checkpoint.Checkpoint(
            name="raw_gate_checkpoint",
            validation_definitions=[validation_definition],
        )
    )

    # Run checkpoint
    result = checkpoint.run(batch_parameters={"dataframe": df})
    return result, context


def print_raw_gate_results(result):
    """Print a human-readable summary of raw gate results."""
    print("\n" + "=" * 60)
    print("LAYER 1: RAW DATA GATE RESULTS")
    print("=" * 60)
    print(result.describe())

    if result.success:
        print("\nRAW GATE: PASSED")
    else:
        print("\nRAW GATE: FAILED - Pipeline would halt here in production")


if __name__ == "__main__":
    df = pd.read_csv("regulatory_data_raw.csv")
    result, context = run_raw_gate(df)
    print_raw_gate_results(result)

Your first quality gate is live and catching real data problems. Next up, you will build a second layer that validates business rules and regulatory invariants like Basel III capital floors.

Build the Business Rule Gate

Your raw gate from the last step catches surface-level problems like nulls, invalid codes, and out-of-range values. Those checks confirm the data arrived intact, but they say nothing about whether the numbers make regulatory sense.

In this step, you will build a second quality gate that validates Basel III business rules. These are the mathematical invariants and capital floors that regulators actually audit. If Tier 1 capital is less than CET1 capital, or the total capital ratio dips below 8%, the filing is wrong no matter how clean the raw data looks.

In this step, get ready to:
  • Transform the raw data and derive cross-column check columns.
  • Define a business rule Expectation Suite with ~15 regulatory expectations.
  • Run the business rule gate and review which regulatory checks fail.
Load and transform the data

Great Expectations does not have a built-in way to compare two columns directly (like "is tier1_capital >= cet1_capital?"). The workaround is to compute a difference column and then check that the difference is >= 0.

  • Open business_rule_gate.py in your editor.
  • Add the imports and the prepare_data() function by pasting this code:
import great_expectations as gx
import pandas as pd
import numpy as np


def prepare_data(df):
    """Apply basic transformations and derive validation columns."""
    df_transformed = df.copy()

    # Drop rows with null entity_id (non-negotiable identifier)
    df_transformed = df_transformed.dropna(subset=["entity_id"])

    # Coerce numeric columns
    numeric_cols = [
        "cet1_capital", "tier1_capital", "total_capital", "rwa",
        "cet1_ratio", "tier1_ratio", "total_capital_ratio",
        "leverage_ratio", "exposure_amount"
    ]
    for col in numeric_cols:
        df_transformed[col] = pd.to_numeric(df_transformed[col], errors="coerce")

    # Derive cross-column check columns
    df_transformed["tier1_minus_cet1"] = (
        df_transformed["tier1_capital"] - df_transformed["cet1_capital"]
    )
    df_transformed["total_minus_tier1"] = (
        df_transformed["total_capital"] - df_transformed["tier1_capital"]
    )
    df_transformed["tier1_ratio_minus_cet1_ratio"] = (
        df_transformed["tier1_ratio"] - df_transformed["cet1_ratio"]
    )

    return df_transformed

What does this code do?

The prepare_data() function cleans the raw data before business rule validation begins.

  • dropna(subset=["entity_id"]) removes rows with no entity identifier. A bank record without an ID is useless for regulatory reporting.
  • pd.to_numeric() coerces all numeric columns so string values become NaN instead of crashing the math.
  • The three derived columns (tier1_minus_cet1, total_minus_tier1, tier1_ratio_minus_cet1_ratio) let GX validate cross-column relationships by checking that each difference is >= 0.

Why derive difference columns?

CET1 capital is a subset of Tier 1 capital, which is a subset of total capital. These are mathematical identities in the Basel III framework. If tier1_capital is ever less than cet1_capital, the data is fundamentally broken.

By computing the difference and checking it is >= 0, you turn a cross-column constraint into a single-column range check that GX handles natively.

  • Save business_rule_gate.py.

You won't see a change yet. The next chunk adds the function that actually runs the gate using this transformation.

Define the business rule Expectation Suite

The raw gate asked "is this data structurally valid?" The business rule gate asks "does this data obey regulatory rules?" You will define expectations across five categories: cross-column constraints, capital floor checks, statistical distributions, referential integrity, and row count validation.

  • In business_rule_gate.py, add the run_business_rule_gate() function below prepare_data() by pasting this code:
def run_business_rule_gate(df):
    """Run Layer 2 business rule quality gate on transformed data."""
    df_transformed = prepare_data(df)

    context = gx.get_context()

    # Connect to data
    data_source = context.data_sources.add_pandas("pandas_business")
    data_asset = data_source.add_dataframe_asset(name="transformed_regulatory_data")
    batch_definition = data_asset.add_batch_definition_whole_dataframe("business_batch")
    batch = batch_definition.get_batch(batch_parameters={"dataframe": df_transformed})

    # Create Expectation Suite
    suite = context.suites.add(
        gx.core.expectation_suite.ExpectationSuite(name="business_rule_gate_suite")
    )

    # Cross-column constraints (Basel III invariants)
    # tier1_capital >= cet1_capital (difference must be >= 0)
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="tier1_minus_cet1", min_value=0
        )
    )
    # total_capital >= tier1_capital
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="total_minus_tier1", min_value=0
        )
    )
    # tier1_ratio >= cet1_ratio
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="tier1_ratio_minus_cet1_ratio", min_value=0
        )
    )

    # Capital floor checks (Basel III minimums)
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="cet1_ratio", min_value=0.045, mostly=0.95
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="tier1_ratio", min_value=0.06, mostly=0.95
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="total_capital_ratio", min_value=0.08, mostly=0.95
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="leverage_ratio", min_value=0.03, mostly=0.95
        )
    )

    # Statistical distribution checks
    suite.add_expectation(
        gx.expectations.ExpectColumnMeanToBeBetween(
            column="cet1_ratio", min_value=0.04, max_value=0.20
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnMeanToBeBetween(
            column="rwa", min_value=50000, max_value=350000
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnMedianToBeBetween(
            column="tier1_capital", min_value=1000, max_value=50000
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnStdevToBeBetween(
            column="cet1_capital", min_value=100, max_value=30000
        )
    )

    # Referential integrity
    suite.add_expectation(
        gx.expectations.ExpectColumnDistinctValuesToBeInSet(
            column="entity_type",
            value_set=["G-SIB", "D-SIB", "Regional", "Commercial"]
        )
    )

    # Row count validation (data not truncated during transformation)
    suite.add_expectation(
        gx.expectations.ExpectTableRowCountToBeBetween(min_value=8000, max_value=11000)
    )

    # Additional ratio consistency
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="cet1_capital", min_value=0, mostly=0.95
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="tier1_capital", min_value=0, mostly=0.95
        )
    )

    # Create Validation Definition and Checkpoint
    validation_definition = context.validation_definitions.add(
        gx.core.validation_definition.ValidationDefinition(
            name="business_rule_validation",
            data=batch_definition,
            suite=suite,
        )
    )

    checkpoint = context.checkpoints.add(
        gx.checkpoint.checkpoint.Checkpoint(
            name="business_rule_checkpoint",
            validation_definitions=[validation_definition],
        )
    )

    result = checkpoint.run(batch_parameters={"dataframe": df_transformed})
    return result, context, df_transformed

What does this function do?

This function follows the same GX pattern you used in raw_gate.py: create a Data Context, connect to data, define expectations, then run a Checkpoint. The difference is what the expectations check.

  • Cross-column constraints verify that capital hierarchy holds (Tier 1 >= CET1, Total >= Tier 1). These are mathematical identities in the Basel III framework.
  • Capital floor checks enforce Basel III minimums: 4.5% CET1, 6% Tier 1, 8% total capital, and 3% leverage. The mostly=0.95 parameter allows up to 5% of rows to fall below the floor before the expectation fails.
  • Statistical distribution checks catch magnitude errors (like reporting basis points instead of decimals) by verifying that column means, medians, and standard deviations fall within expected ranges.
  • Referential integrity and row count checks confirm no invalid entity types slipped through and no data was accidentally truncated during transformation.
  • Save business_rule_gate.py.

You won't see a change yet. The next chunk adds the print function and the entry point to actually run everything.

Seeing a syntax error when you save?

  • Make sure the run_business_rule_gate() function is defined at the module level (no extra indentation), not nested inside prepare_data().
  • Check that every parenthesis in the suite.add_expectation() calls is properly closed. Each call has two closing parentheses: one for the expectation constructor and one for add_expectation().

Still stuck? Help me debug a syntax error in my business_rule_gate.py file.

Run the business rule gate and review results

The gate logic is ready. Now you need a print function for human-readable output and an entry point so you can run the file directly from the terminal.

  • Add the print_business_rule_results() function and the __main__ block at the bottom of business_rule_gate.py by pasting this code:
def print_business_rule_results(result):
    """Print a human-readable summary of business rule gate results."""
    print("\n" + "=" * 60)
    print("LAYER 2: BUSINESS RULE GATE RESULTS")
    print("=" * 60)
    print(result.describe())

    if result.success:
        print("\nBUSINESS RULE GATE: PASSED")
    else:
        print("\nBUSINESS RULE GATE: FAILED")


if __name__ == "__main__":
    df = pd.read_csv("regulatory_data_raw.csv")
    result, context, df_transformed = run_business_rule_gate(df)
    print_business_rule_results(result)

What does this code do?

  • print_business_rule_results() uses result.describe() to print the full checkpoint summary, then adds a clear PASSED or FAILED verdict.
  • The __main__ block loads regulatory_data_raw.csv, runs the gate, and prints results. This lets you test the gate on its own before wiring it into the full pipeline later.
  • Save business_rule_gate.py.
  • Run the business rule gate by running this command in your terminal:
python business_rule_gate.py

You should see a checkpoint result summary ending with BUSINESS RULE GATE: FAILED. The output shows both passed and failed expectations with specific counts and percentages.

Why does the business rule gate fail?

The anomalies you injected in Step 2 include rows where tier1_ratio < cet1_ratio (violating the capital hierarchy) and rows where total_capital_ratio < 0.08 (violating the Basel III 8% minimum floor). The gate is working exactly as designed.

In a real bank, these failures would halt the regulatory filing process until the data team investigates and remediates the issues. This is BCBS 239 Principle 3 in action: catching problems at the point of origination, not after the report is filed.

Seeing a ModuleNotFoundError or FileNotFoundError?

  • If you see a FileNotFoundError for regulatory_data_raw.csv, make sure your terminal is in the regulatory-attestation-pipeline folder and that you ran generate_data.py in a previous step.
  • If you see a ModuleNotFoundError for great_expectations, confirm your virtual environment is still activated. You should see (venv) at the start of your terminal prompt.

Need more help? Help me debug an error when running business_rule_gate.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved business_rule_gate.py and that your terminal printed the checkpoint results.

ⓧ I'd like to double check the full code

Here is the complete business_rule_gate.py file. Compare it with yours line by line.

import great_expectations as gx
import pandas as pd
import numpy as np


def prepare_data(df):
    """Apply basic transformations and derive validation columns."""
    df_transformed = df.copy()
    df_transformed = df_transformed.dropna(subset=["entity_id"])
    numeric_cols = ["cet1_capital", "tier1_capital", "total_capital", "rwa", "cet1_ratio", "tier1_ratio", "total_capital_ratio", "leverage_ratio", "exposure_amount"]
    for col in numeric_cols:
        df_transformed[col] = pd.to_numeric(df_transformed[col], errors="coerce")
    df_transformed["tier1_minus_cet1"] = df_transformed["tier1_capital"] - df_transformed["cet1_capital"]
    df_transformed["total_minus_tier1"] = df_transformed["total_capital"] - df_transformed["tier1_capital"]
    df_transformed["tier1_ratio_minus_cet1_ratio"] = df_transformed["tier1_ratio"] - df_transformed["cet1_ratio"]
    return df_transformed


def run_business_rule_gate(df):
    """Run Layer 2 business rule quality gate on transformed data."""
    df_transformed = prepare_data(df)
    context = gx.get_context()
    data_source = context.data_sources.add_pandas("pandas_business")
    data_asset = data_source.add_dataframe_asset(name="transformed_regulatory_data")
    batch_definition = data_asset.add_batch_definition_whole_dataframe("business_batch")
    batch = batch_definition.get_batch(batch_parameters={"dataframe": df_transformed})

    suite = context.suites.add(gx.core.expectation_suite.ExpectationSuite(name="business_rule_gate_suite"))

    suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="tier1_minus_cet1", min_value=0))
    suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="total_minus_tier1", min_value=0))
    suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="tier1_ratio_minus_cet1_ratio", min_value=0))
    suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="cet1_ratio", min_value=0.045, mostly=0.95))
    suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="tier1_ratio", min_value=0.06, mostly=0.95))
    suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="total_capital_ratio", min_value=0.08, mostly=0.95))
    suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="leverage_ratio", min_value=0.03, mostly=0.95))
    suite.add_expectation(gx.expectations.ExpectColumnMeanToBeBetween(column="cet1_ratio", min_value=0.04, max_value=0.20))
    suite.add_expectation(gx.expectations.ExpectColumnMeanToBeBetween(column="rwa", min_value=50000, max_value=350000))
    suite.add_expectation(gx.expectations.ExpectColumnMedianToBeBetween(column="tier1_capital", min_value=1000, max_value=50000))
    suite.add_expectation(gx.expectations.ExpectColumnStdevToBeBetween(column="cet1_capital", min_value=100, max_value=30000))
    suite.add_expectation(gx.expectations.ExpectColumnDistinctValuesToBeInSet(column="entity_type", value_set=["G-SIB", "D-SIB", "Regional", "Commercial"]))
    suite.add_expectation(gx.expectations.ExpectTableRowCountToBeBetween(min_value=8000, max_value=11000))
    suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="cet1_capital", min_value=0, mostly=0.95))
    suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="tier1_capital", min_value=0, mostly=0.95))

    validation_definition = context.validation_definitions.add(
        gx.core.validation_definition.ValidationDefinition(name="business_rule_validation", data=batch_definition, suite=suite)
    )
    checkpoint = context.checkpoints.add(
        gx.checkpoint.checkpoint.Checkpoint(name="business_rule_checkpoint", validation_definitions=[validation_definition])
    )
    result = checkpoint.run(batch_parameters={"dataframe": df_transformed})
    return result, context, df_transformed


def print_business_rule_results(result):
    print("\n" + "=" * 60)
    print("LAYER 2: BUSINESS RULE GATE RESULTS")
    print("=" * 60)
    print(result.describe())
    if result.success:
        print("\nBUSINESS RULE GATE: PASSED")
    else:
        print("\nBUSINESS RULE GATE: FAILED")


if __name__ == "__main__":
    df = pd.read_csv("regulatory_data_raw.csv")
    result, context, df_transformed = run_business_rule_gate(df)
    print_business_rule_results(result)

Both quality gates are now built and catching real problems. Next up, you will wire them together into a pipeline orchestrator that runs both gates in sequence and generates an audit-ready attestation report.

Build the Pipeline Orchestrator and Generate Attestation

Your two quality gates are built and tested individually. Now it is time to wire them into a single orchestrator that runs both gates in sequence, prints a consolidated attestation verdict, and checks for a report.

In production, regulatory data teams run this exact workflow before every BCBS 239 filing. If either gate fails, the pipeline halts and the team investigates before resubmitting.

In this step, get ready to:
  • Build the pipeline orchestrator that runs both quality gates and prints a consolidated attestation summary.
  • Add the attestation report output and entry point.
  • Run the full pipeline and review the results.
Build the pipeline orchestrator

The orchestrator loads the raw data, feeds it through both quality gates in order, and produces a single APPROVED or REJECTED verdict. This is the function that would run on a schedule in a real bank's data pipeline.

  • In run_pipeline.py, add the following code:
import sys
import webbrowser
from pathlib import Path

import pandas as pd

from raw_gate import run_raw_gate, print_raw_gate_results
from business_rule_gate import run_business_rule_gate, print_business_rule_results


def run_attestation_pipeline():
    """Run the full two-layer attestation pipeline."""
    print("=" * 60)
    print("REGULATORY DATA ATTESTATION PIPELINE")
    print("Basel III / BCBS 239 Compliance Check")
    print("=" * 60)

    # Load raw data
    print("\nLoading regulatory data...")
    df = pd.read_csv("regulatory_data_raw.csv")
    print(f"Loaded {len(df)} records with {len(df.columns)} columns")

    # Layer 1: Raw Data Gate
    print("\n--- Running Layer 1: Raw Data Gate ---")
    raw_result, raw_context = run_raw_gate(df)
    print_raw_gate_results(raw_result)

    # Layer 2: Business Rule Gate
    print("\n--- Running Layer 2: Business Rule Gate ---")
    biz_result, biz_context, df_transformed = run_business_rule_gate(df)
    print_business_rule_results(biz_result)

    # Consolidated Summary
    print("\n" + "=" * 60)
    print("CONSOLIDATED ATTESTATION SUMMARY")
    print("=" * 60)

    raw_success = raw_result.success
    biz_success = biz_result.success

    print(f"Raw Data Gate:       {'PASSED' if raw_success else 'FAILED'}")
    print(f"Business Rule Gate:  {'PASSED' if biz_success else 'FAILED'}")
    print(f"Overall Attestation: {'APPROVED' if (raw_success and biz_success) else 'REJECTED'}")

    if not (raw_success and biz_success):
        print("\nATTESTATION REJECTED - Data does not meet regulatory quality standards.")
        print("Review the Data Docs report for detailed failure analysis.")
        print("Pipeline halted. Fix data issues and re-run.")

What does this code do?

  • The imports pull in run_raw_gate and run_business_rule_gate from the gate files you built in previous steps. The same DataFrame is passed to both gates so they validate the exact same data snapshot.
  • The consolidated summary checks raw_result.success and biz_result.success to determine each gate's outcome.
  • The overall attestation prints APPROVED only when both gates pass. Any single failure prints REJECTED and a halt message telling the data team to investigate.

You will not see a change yet. The next chunk completes the function and adds the entry point.

Add the attestation report output and entry point

Data Docs are HTML reports that Great Expectations generates from your validation results. The orchestrator checks for these reports and opens them in your browser automatically.

  • In run_pipeline.py, add the following code directly below the halt message block, still inside the run_attestation_pipeline function:
    # Open Data Docs if available
    data_docs_dir = Path("gx/uncommitted/data_docs/local_site")
    if data_docs_dir.exists():
        index_path = data_docs_dir / "index.html"
        if index_path.exists():
            print(f"\nData Docs report: {index_path.absolute()}")
            webbrowser.open(str(index_path.absolute()))

    return 0 if (raw_success and biz_success) else 1


if __name__ == "__main__":
    exit_code = run_attestation_pipeline()
    sys.exit(exit_code)

What does this code do?

  • The Path check looks for a Data Docs report at gx/uncommitted/data_docs/local_site/. If the HTML file exists, it opens in your default browser.
  • The function returns exit code 0 for success and 1 for failure. This makes the script compatible with CI/CD pipelines that use exit codes to gate deployments.
  • The if __name__ == "__main__" block wires the function to the terminal so you can run the pipeline with python run_pipeline.py.
  • Save run_pipeline.py.

Why does the code check for Data Docs instead of generating them?

GX generates Data Docs when a checkpoint is configured with a dedicated action. In this project, the orchestrator focuses on validation and reporting. The code handles both scenarios gracefully: it opens the report if one exists, and skips silently if it does not.

✔️ Awesome, I've got everything!

Great. Make sure you have saved run_pipeline.py before moving on.

ⓧ I'd like to double check the full code

import sys
import webbrowser
from pathlib import Path

import pandas as pd

from raw_gate import run_raw_gate, print_raw_gate_results
from business_rule_gate import run_business_rule_gate, print_business_rule_results


def run_attestation_pipeline():
    """Run the full two-layer attestation pipeline."""
    print("=" * 60)
    print("REGULATORY DATA ATTESTATION PIPELINE")
    print("Basel III / BCBS 239 Compliance Check")
    print("=" * 60)

    print("\nLoading regulatory data...")
    df = pd.read_csv("regulatory_data_raw.csv")
    print(f"Loaded {len(df)} records with {len(df.columns)} columns")

    print("\n--- Running Layer 1: Raw Data Gate ---")
    raw_result, raw_context = run_raw_gate(df)
    print_raw_gate_results(raw_result)

    print("\n--- Running Layer 2: Business Rule Gate ---")
    biz_result, biz_context, df_transformed = run_business_rule_gate(df)
    print_business_rule_results(biz_result)

    print("\n" + "=" * 60)
    print("CONSOLIDATED ATTESTATION SUMMARY")
    print("=" * 60)

    raw_success = raw_result.success
    biz_success = biz_result.success

    print(f"Raw Data Gate:       {'PASSED' if raw_success else 'FAILED'}")
    print(f"Business Rule Gate:  {'PASSED' if biz_success else 'FAILED'}")
    print(f"Overall Attestation: {'APPROVED' if (raw_success and biz_success) else 'REJECTED'}")

    if not (raw_success and biz_success):
        print("\nATTESTATION REJECTED - Data does not meet regulatory quality standards.")
        print("Review the Data Docs report for detailed failure analysis.")
        print("Pipeline halted. Fix data issues and re-run.")

    data_docs_dir = Path("gx/uncommitted/data_docs/local_site")
    if data_docs_dir.exists():
        index_path = data_docs_dir / "index.html"
        if index_path.exists():
            print(f"\nData Docs report: {index_path.absolute()}")
            webbrowser.open(str(index_path.absolute()))

    return 0 if (raw_success and biz_success) else 1


if __name__ == "__main__":
    exit_code = run_attestation_pipeline()
    sys.exit(exit_code)
Run the full pipeline and review the attestation report

With the orchestrator complete, it is time to run the full pipeline against your anomaly-injected dataset. Both gates should fail because of the injected anomalies.

  • Run the full attestation pipeline by executing:
python run_pipeline.py

You should see both gate results print in sequence, followed by the consolidated attestation summary at the bottom of the output.

The summary should show Raw Data Gate: FAILED, Business Rule Gate: FAILED, and Overall Attestation: REJECTED.

Why does the pipeline reject the data?

Your regulatory_data_raw.csv file has approximately 5% injected anomalies. These include null values in critical columns, negative capital figures, invalid country codes, and cross-column violations like tier1_ratio < cet1_ratio.

This is exactly how the pipeline should behave. In a real regulatory filing workflow, a REJECTED result means the data team must fix the source data before the bank can submit its Basel III report to regulators.

Seeing an import error or FileNotFoundError?

  • Make sure you are running the command from inside the regulatory-attestation-pipeline folder where all four Python files live. The imports expect raw_gate.py and business_rule_gate.py in the same directory.
  • Check that your virtual environment is still active. If your terminal prompt does not show (venv), reactivate it before running the pipeline.
  • If you see a FileNotFoundError for regulatory_data_raw.csv, run python generate_data.py first to regenerate the dataset.

Still stuck? Help me debug my run_pipeline.py errors.

Secret mission

Build the Anomaly Quarantine System

Your pipeline catches bad data. But what happens to the records that fail? In production, data engineers don't just flag failures. They isolate the bad records, document exactly what went wrong, and produce a clean dataset that passes every check.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • regulatory-attestation-pipeline project folder on your Desktop (Python scripts, CSV data files, Data Docs HTML reports).
  • Python virtual environment (venv/) with Great Expectations installed inside the project folder.

Keep everything running

No action needed. Choose this if you are still actively building or plan to extend the pipeline further.

  • Your project folder stays on your Desktop with all scripts, CSV files, and Data Docs reports intact.
  • The virtual environment stays ready. Reactivate it any time to run the pipeline again.
  • Everything is local, so nothing is costing you money while it sits on your machine.

Pause - I'll come back to this later

Deactivate the virtual environment to free up your terminal session. Your files stay on disk so you can pick up where you left off.

  • If your virtual environment is still active, deactivate it by running:
deactivate
  • Close your terminal window.
  • When you are ready to come back, open a terminal, navigate to ~/Desktop/regulatory-attestation-pipeline, and reactivate the virtual environment.

Delete - I don't want to use this again

Remove the entire project folder from your Desktop. This deletes all Python scripts, CSV data files, the virtual environment, and the Data Docs reports.

  • If your virtual environment is still active, deactivate it first by running:
deactivate
  • Navigate to your Desktop by running:
cd ~/Desktop
  • Delete the project folder.

macOS

  • Remove the project folder by running:
rm -rf regulatory-attestation-pipeline

Windows

  • Remove the project folder by running:
Remove-Item -Recurse -Force regulatory-attestation-pipeline
  • Confirm the folder is gone by running:
ls
  • You should no longer see regulatory-attestation-pipeline in the output.

Nice Work!

Nice Work!

You just built a production-grade regulatory data attestation pipeline using Great Expectations and Python. This is exactly the kind of data quality system banks run before submitting Basel III capital adequacy reports to regulators.

You've learned how to:

  • Generate a synthetic Basel III capital adequacy dataset with 10,000 records and 13 regulatory data columns, including deliberate anomaly injection for testing.
  • Implement a two-layer quality gate system with ~45 total expectations covering schema, completeness, domain, range, cross-column constraints, capital floors, and statistical distributions.
  • Build a pipeline orchestrator that halts on validation failure, enforcing BCBS 239 Principle 3 (completeness and accuracy at origination), and generates audit-ready Data Docs HTML reports with timestamped validation evidence.
  • Secret Mission: Built an anomaly quarantine system that isolates bad records with failure reasons, remediates the dataset, and produces a passing attestation report.

Ready to quiz yourself?