AI-Readiness Governance Gate

Introduction

30 Second Summary

Every time an AI model accesses enterprise data, it trusts that the data is clean, properly classified, and traceable. When that trust is misplaced, the result is hallucinations, privacy violations, and regulatory fines.

In this project, you will build a Python-based governance gate that evaluates 50,000+ financial records for AI-readiness. Your gate profiles data quality, detects sensitive fields, enforces 18 validation rules, and produces an automated pass/fail decision with a full audit trail.

What You'll Build

You will build an AI-Readiness Governance Gate that scores enterprise data across five quality dimensions, classifies fields by sensitivity, validates against business rules using Great Expectations, and generates an HTML scorecard with a gate decision.

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

• A data profiling engine that scores completeness, validity, consistency, uniqueness, and timeliness across 50,000+ records.
• A PII sensitivity classifier with regex-based detection that classifies 100% of data fields into four security tiers.
• An 18-rule validation suite powered by Great Expectations with automated Data Docs, plus a composite AI-Readiness Scorecard with weighted scoring, gate logic, and an HTML report.
• Secret Mission: Add historical governance trend tracking to monitor whether data quality improves or degrades over time.

Are there any prerequisites?

You should have Python 3.11 or higher installed, along with pip and a code editor like VS Code. Familiarity with intermediate Python and a basic understanding of JSON data structures will help you follow along.

Before We Start

Before you can build a governance gate, you need a clean project environment with the right tools installed.

In this step, you will create an isolated Python project and install Great Expectations (a data quality framework) and pandas (a data analysis library).

In this step, get ready to:

• Create a project directory and Python virtual environment.
• Install Great Expectations and pandas.
• Verify the installation works.

Create project structure and install dependencies

Every Python project benefits from its own virtual environment. This keeps the project's packages separate from your system Python, avoiding version conflicts between projects.

• Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project directory and move into it by running these commands:

mkdir ai_readiness_gate
cd ai_readiness_gate

• Confirm the folder exists by running:

ls

You should see the ai_readiness_gate folder listed on your Desktop. Your terminal prompt should now show you are inside this folder.

• Create a Python virtual environment by running this command:

macOS

python3 -m venv venv

• Activate the virtual environment by running:

source venv/bin/activate

Windows

python -m venv venv

• Activate the virtual environment by running:

venv\Scripts\activate

After activation, your terminal prompt changes to show (venv) at the beginning. This confirms you are working inside the virtual environment.

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay local to this project, so they never interfere with other projects on your machine.

Not seeing (venv) in your prompt?

Make sure you ran the activate command from inside the ai_readiness_gate directory. On Windows, if you see a permissions error, run Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser in PowerShell first, then try the activate command again.

help me activate my Python virtual environment

Now you will create a requirements.txt file that lists the exact package versions the project needs.

• In your code editor, create a new file called requirements.txt inside the ai_readiness_gate folder.
• Paste the following content into requirements.txt:

great_expectations==1.18.1
pandas==3.0.3

What does this file do?

The requirements.txt file pins the exact versions of each library. This ensures everyone working on the project gets identical behaviour, regardless of when they install.

Great Expectations (1.18.1) is a data quality framework that lets you define and enforce validation rules. pandas (3.0.3) is the standard Python library for tabular data manipulation.

• Save requirements.txt.
• Install the dependencies by running this command in your terminal:

pip install -r requirements.txt

This downloads and installs both packages along with their dependencies. It may take a minute or two.

Seeing installation errors?

Make sure (venv) appears in your terminal prompt. If it does not, you are installing into your system Python instead of the virtual environment. Run the activation command again before retrying.

If you see a message about Python version compatibility, confirm you are running Python 3.11 or higher. pandas 3.0.3 requires Python 3.11 or above.

help me fix my pip install errors for great_expectations and pandas

• Verify the installation by running this command:

python -c "import great_expectations as gx; print(gx.__version__)"

You should see 1.18.1 printed in the terminal. This confirms Great Expectations is installed correctly.

Seeing a ModuleNotFoundError?

This means the package did not install into your active environment. Check that (venv) is visible in your prompt. If it is, try running pip install -r requirements.txt again and watch for errors in the output.

help me fix ModuleNotFoundError for great_expectations

✔️ Awesome, I've got everything!

Great. Double check you have saved requirements.txt and your terminal shows 1.18.1 after running the verify command.

ⓧ I'd like to double check the full code

Here is the complete requirements.txt file. Make sure yours matches exactly:

great_expectations==1.18.1
pandas==3.0.3

Your environment is ready. Next up, you will build a sample data generator that produces 50,000 realistic financial transaction records for the governance gate to assess.

Generate Sample Financial Data

A governance gate is only useful when it has realistic data to assess. Enterprise financial systems contain messy, sensitive, and inconsistent records that AI systems must never access without quality checks.

In this step, you will build a data generator that produces 50,000 financial transaction records with intentional quality issues. These issues simulate the exact problems a governance gate needs to catch before AI access is granted.

In this step, get ready to:

• Build a data generator that produces realistic financial transaction records with PII fields.
• Inject intentional data quality issues (nulls, invalid formats, duplicates, future dates).
• Generate and verify 50,000 records as JSON output.

Build the sample data generator

The generator needs to produce records that look like real financial transactions. Each record will contain fields that mirror what a company like American Express processes daily: customer identifiers, payment details, and transaction metadata.

Why inject quality issues on purpose?

Real enterprise data is never perfectly clean. Missing values, format violations, and duplicate records are common in production systems. By injecting these issues at known rates, you can later verify that your governance gate catches them correctly.

• Create a new file called generate_data.py in your ai_readiness_gate/ directory.
• Add the imports and helper functions that generate individual field values by copying this code:

import json
import random
import string
import os
from datetime import datetime, timedelta


def generate_transaction_id(index):
    return f"TXN-{index:08d}"


def generate_card_number():
    # Simulate credit card numbers with common Amex prefixes
    prefix = random.choice(["3782", "3714", "3787"])
    remaining = "".join(random.choices(string.digits, k=11))
    return f"{prefix}{remaining}"


def generate_ssn():
    # Simulate Social Security Numbers (XXX-XX-XXXX format)
    area = random.randint(100, 999)
    group = random.randint(10, 99)
    serial = random.randint(1000, 9999)
    return f"{area}-{group}-{serial}"


def generate_email(name):
    domains = ["gmail.com", "yahoo.com", "outlook.com", "amex.com"]
    clean_name = name.lower().replace(" ", ".")
    return f"{clean_name}@{random.choice(domains)}"


def generate_phone():
    country = random.choice(["+1", "+44", "+91", "+61"])
    number = "".join(random.choices(string.digits, k=10))
    return f"{country}-{number}"


def generate_transaction_date():
    # Generate a date within the past 30 days
    days_ago = random.randint(0, 30)
    hours_ago = random.randint(0, 23)
    dt = datetime.now() - timedelta(days=days_ago, hours=hours_ago)
    return dt.strftime("%Y-%m-%d %H:%M:%S")


def generate_future_date():
    # Intentionally invalid: future dates should not exist in transaction data
    days_ahead = random.randint(1, 30)
    dt = datetime.now() + timedelta(days=days_ahead)
    return dt.strftime("%Y-%m-%d %H:%M:%S")

What do these helper functions do?

• generate_transaction_id(index) creates IDs like TXN-00000001 for easy duplicate detection later.
• generate_card_number() and generate_ssn() produce realistic PII fields that your classifier will need to flag as restricted.
• generate_future_date() produces dates in the future, which are intentionally invalid for transaction records.

• Below the helper functions, add the core generate_records function that orchestrates record creation and injects quality issues:

def generate_records(num_records=50000, quality_level=0.9):
    first_names = ["James", "Maria", "Robert", "Linda", "David", "Sarah",
                   "Michael", "Jennifer", "William", "Elizabeth", "Raj",
                   "Priya", "Chen", "Yuki", "Ahmed", "Fatima"]
    last_names = ["Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia",
                  "Miller", "Davis", "Wilson", "Anderson", "Patel", "Kumar",
                  "Wang", "Tanaka", "Hassan", "Ali"]
    merchant_categories = ["Retail", "Travel", "Dining", "Entertainment",
                           "Grocery", "Fuel", "Healthcare", "Utilities"]
    account_types = ["Platinum", "Gold", "Green", "Corporate", "Blue"]
    country_codes = ["US", "UK", "IN", "AU", "CA", "DE", "JP", "SG"]

    null_rate = 1.0 - quality_level
    invalid_email_rate = 0.03
    duplicate_rate = 0.02
    future_date_rate = 0.01

    records = []
    duplicate_ids = set()

    for i in range(num_records):
        name = f"{random.choice(first_names)} {random.choice(last_names)}"
        record = {
            "transaction_id": generate_transaction_id(i),
            "customer_name": name,
            "email": generate_email(name),
            "phone": generate_phone(),
            "card_number": generate_card_number(),
            "ssn": generate_ssn(),
            "transaction_amount": round(random.uniform(1.0, 15000.0), 2),
            "transaction_date": generate_transaction_date(),
            "merchant_category": random.choice(merchant_categories),
            "account_type": random.choice(account_types),
            "country_code": random.choice(country_codes),
        }

        # Inject null values based on quality_level
        if random.random() < null_rate:
            nullable_fields = ["email", "phone", "merchant_category"]
            field = random.choice(nullable_fields)
            record[field] = None

        # Inject invalid email formats (~3%)
        if random.random() < invalid_email_rate and record["email"]:
            record["email"] = "invalid-email-format"

        # Inject duplicate transaction IDs (~2%)
        if random.random() < duplicate_rate:
            dup_index = random.randint(0, max(0, i - 1))
            record["transaction_id"] = generate_transaction_id(dup_index)
            duplicate_ids.add(record["transaction_id"])

        # Inject future-dated transactions (~1%)
        if random.random() < future_date_rate:
            record["transaction_date"] = generate_future_date()

        # Inject negative amounts (~0.5%)
        if random.random() < 0.005:
            record["transaction_amount"] = -abs(record["transaction_amount"])

        records.append(record)

    return records

What does this function do?

The generate_records function builds each record with valid data first, then randomly corrupts some records based on probability rates. The quality_level parameter (default 0.9) controls the null injection rate. A quality level of 0.9 means 10% of records may get a null field.

Each quality issue targets a specific governance dimension. Null values test completeness. Invalid emails test validity. Duplicates test uniqueness. Future dates test consistency. Your profiler will score each dimension separately in the next step.

• Finally, add the main() function at the bottom of the file to run the generator and write output to JSON:

def main():
    os.makedirs("data", exist_ok=True)
    print("Generating 50,000 financial transaction records...")
    records = generate_records(num_records=50000)
    output_path = os.path.join("data", "transactions.json")
    with open(output_path, "w") as f:
        json.dump(records, f, indent=2)
    print(f"Generated {len(records)} records -> {output_path}")
    print(f"File size: {os.path.getsize(output_path) / (1024*1024):.1f} MB")


if __name__ == "__main__":
    main()

What does this code do?

• os.makedirs("data", exist_ok=True) creates the data/ subdirectory if it does not already exist.
• json.dump(records, f, indent=2) writes the records as formatted JSON so you can inspect individual entries later.
• The if __name__ == "__main__": guard ensures main() only runs when you execute this file directly, not when it is imported.

• Save generate_data.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before moving on.

ⓧ I'd like to double check the full code

import json
import random
import string
import os
from datetime import datetime, timedelta


def generate_transaction_id(index):
    return f"TXN-{index:08d}"


def generate_card_number():
    prefix = random.choice(["3782", "3714", "3787"])
    remaining = "".join(random.choices(string.digits, k=11))
    return f"{prefix}{remaining}"


def generate_ssn():
    area = random.randint(100, 999)
    group = random.randint(10, 99)
    serial = random.randint(1000, 9999)
    return f"{area}-{group}-{serial}"


def generate_email(name):
    domains = ["gmail.com", "yahoo.com", "outlook.com", "amex.com"]
    clean_name = name.lower().replace(" ", ".")
    return f"{clean_name}@{random.choice(domains)}"


def generate_phone():
    country = random.choice(["+1", "+44", "+91", "+61"])
    number = "".join(random.choices(string.digits, k=10))
    return f"{country}-{number}"


def generate_transaction_date():
    days_ago = random.randint(0, 30)
    hours_ago = random.randint(0, 23)
    dt = datetime.now() - timedelta(days=days_ago, hours=hours_ago)
    return dt.strftime("%Y-%m-%d %H:%M:%S")


def generate_future_date():
    days_ahead = random.randint(1, 30)
    dt = datetime.now() + timedelta(days=days_ahead)
    return dt.strftime("%Y-%m-%d %H:%M:%S")


def generate_records(num_records=50000, quality_level=0.9):
    first_names = ["James", "Maria", "Robert", "Linda", "David", "Sarah",
                   "Michael", "Jennifer", "William", "Elizabeth", "Raj",
                   "Priya", "Chen", "Yuki", "Ahmed", "Fatima"]
    last_names = ["Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia",
                  "Miller", "Davis", "Wilson", "Anderson", "Patel", "Kumar",
                  "Wang", "Tanaka", "Hassan", "Ali"]
    merchant_categories = ["Retail", "Travel", "Dining", "Entertainment",
                           "Grocery", "Fuel", "Healthcare", "Utilities"]
    account_types = ["Platinum", "Gold", "Green", "Corporate", "Blue"]
    country_codes = ["US", "UK", "IN", "AU", "CA", "DE", "JP", "SG"]

    null_rate = 1.0 - quality_level
    invalid_email_rate = 0.03
    duplicate_rate = 0.02
    future_date_rate = 0.01

    records = []
    duplicate_ids = set()

    for i in range(num_records):
        name = f"{random.choice(first_names)} {random.choice(last_names)}"
        record = {
            "transaction_id": generate_transaction_id(i),
            "customer_name": name,
            "email": generate_email(name),
            "phone": generate_phone(),
            "card_number": generate_card_number(),
            "ssn": generate_ssn(),
            "transaction_amount": round(random.uniform(1.0, 15000.0), 2),
            "transaction_date": generate_transaction_date(),
            "merchant_category": random.choice(merchant_categories),
            "account_type": random.choice(account_types),
            "country_code": random.choice(country_codes),
        }

        if random.random() < null_rate:
            nullable_fields = ["email", "phone", "merchant_category"]
            field = random.choice(nullable_fields)
            record[field] = None

        if random.random() < invalid_email_rate and record["email"]:
            record["email"] = "invalid-email-format"

        if random.random() < duplicate_rate:
            dup_index = random.randint(0, max(0, i - 1))
            record["transaction_id"] = generate_transaction_id(dup_index)
            duplicate_ids.add(record["transaction_id"])

        if random.random() < future_date_rate:
            record["transaction_date"] = generate_future_date()

        if random.random() < 0.005:
            record["transaction_amount"] = -abs(record["transaction_amount"])

        records.append(record)

    return records


def main():
    os.makedirs("data", exist_ok=True)
    print("Generating 50,000 financial transaction records...")
    records = generate_records(num_records=50000)
    output_path = os.path.join("data", "transactions.json")
    with open(output_path, "w") as f:
        json.dump(records, f, indent=2)
    print(f"Generated {len(records)} records -> {output_path}")
    print(f"File size: {os.path.getsize(output_path) / (1024*1024):.1f} MB")


if __name__ == "__main__":
    main()

Generate and verify sample data

With the generator built, it is time to produce the dataset and confirm it has the expected structure. You will also do a quick sanity check using pandas to verify the record count and field names.

• Generate the sample data by running this command in your terminal:

python generate_data.py

• You should see output similar to:

Generating 50,000 financial transaction records...
Generated 50000 records -> data/transactions.json
File size: 128.4 MB

File size may vary slightly

Your file size will be in the range of 120-135 MB depending on the randomly generated names and numbers. The exact value does not matter as long as you see 50,000 records confirmed.

• Verify the record count and structure by running this quick pandas check:

python -c "import pandas as pd, json; df = pd.DataFrame(json.load(open('data/transactions.json'))); print(f'Records: {len(df)}'); print(f'Fields: {list(df.columns)}')"

• You should see 50,000 records and 11 fields listed: transaction_id, customer_name, email, phone, card_number, ssn, transaction_amount, transaction_date, merchant_category, account_type, country_code.

Seeing a FileNotFoundError?

• Make sure you are running the commands from inside the ai_readiness_gate/ directory, not a parent folder.
• Confirm that python generate_data.py completed successfully before running the verification command.
• Check that a data/ folder appeared in your project directory by running ls.

Why can't I find my data/transactions.json file?

• Confirm the data/ folder was created by running:

ls data/

• You should see transactions.json listed in the output.

You now have 50,000 records of realistic financial transaction data loaded with intentional quality problems. Next up, you will build a profiling engine that scores this data across five quality dimensions.

Build the Data Profiling Engine

Your sample data is generated and waiting in data/transactions.json. Now you need a way to measure how healthy that data actually is before any AI system touches it.

In this step, you will build a profiling engine that scores your dataset across 5 quality dimensions. Enterprise governance teams use these exact dimensions to determine whether data is safe for AI consumption.

In this step, get ready to:

• Build a DataProfiler class with 5 scoring methods.
• Aggregate dimension scores into a profile report.
• Run the profiler against your 50,000 transaction records.

Build the profiling module

Data profiling measures how complete, valid, consistent, unique, and timely your data is. Each dimension gets a score from 0.0 to 1.0, giving you a quantifiable picture of data health.

What are the 5 quality dimensions?

Completeness measures how many cells have actual values (not null). Validity checks whether values follow the correct format (like email addresses matching a regex pattern).

Consistency verifies cross-field logic (e.g., transaction dates should not be in the future). Uniqueness detects duplicates in fields that should be unique. Timeliness confirms data is fresh enough for AI consumption.

• Create a new file called profiler.py in your ai_readiness_gate/ folder.
• Add the imports, class definition, and the first scoring method by pasting the following code:

import pandas as pd
from datetime import datetime, timedelta
import re


class DataProfiler:
    def __init__(self, df):
        self.df = df
        self.results = {}

    def score_completeness(self):
        total_cells = self.df.size
        non_null_cells = self.df.count().sum()
        score = non_null_cells / total_cells
        null_by_column = self.df.isnull().sum()
        fields_with_nulls = null_by_column[null_by_column > 0].to_dict()
        self.results["completeness"] = {
            "score": round(score, 4),
            "fields_with_nulls": fields_with_nulls,
        }
        return score

What does this code do?

The DataProfiler class accepts a pandas DataFrame and stores results in a dictionary as each scoring method runs.

The score_completeness() method compares total cells to non-null cells. It also records which specific fields have null values, so you know exactly where the gaps are.

You won't see a change yet. The next chunk adds the remaining four scoring methods.

• Add the remaining four scoring methods directly below score_completeness() in profiler.py:

    def score_validity(self):
        checks = 0
        passes = 0

        if "email" in self.df.columns:
            email_mask = self.df["email"].dropna()
            email_pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
            valid_emails = email_mask.str.match(email_pattern).sum()
            checks += len(email_mask)
            passes += valid_emails

        if "transaction_amount" in self.df.columns:
            amounts = self.df["transaction_amount"].dropna()
            valid_amounts = (amounts > 0).sum()
            checks += len(amounts)
            passes += valid_amounts

        if "country_code" in self.df.columns:
            codes = self.df["country_code"].dropna()
            valid_codes = (codes.str.len() == 2).sum()
            checks += len(codes)
            passes += valid_codes

        score = passes / checks if checks > 0 else 0.0
        self.results["validity"] = {"score": round(score, 4)}
        return score

    def score_consistency(self):
        checks = 0
        passes = 0

        if "transaction_date" in self.df.columns:
            dates = pd.to_datetime(self.df["transaction_date"], errors="coerce")
            now = pd.Timestamp.now()
            valid_dates = (dates <= now).sum()
            checks += dates.notna().sum()
            passes += valid_dates

        if "transaction_amount" in self.df.columns and "account_type" in self.df.columns:
            corporate = self.df[self.df["account_type"] == "Corporate"]
            valid_corporate = (corporate["transaction_amount"] > 0).sum()
            checks += len(corporate)
            passes += valid_corporate

        score = passes / checks if checks > 0 else 0.0
        self.results["consistency"] = {"score": round(score, 4)}
        return score

    def score_uniqueness(self):
        if "transaction_id" not in self.df.columns:
            self.results["uniqueness"] = {"score": 1.0}
            return 1.0

        total = len(self.df)
        unique = self.df["transaction_id"].nunique()
        duplicates = total - unique
        score = unique / total
        self.results["uniqueness"] = {
            "score": round(score, 4),
            "duplicate_count": duplicates,
        }
        return score

    def score_timeliness(self, threshold_hours=24):
        if "transaction_date" in self.df.columns:
            dates = pd.to_datetime(self.df["transaction_date"], errors="coerce")
            max_date = dates.max()
            now = pd.Timestamp.now()
            hours_since_latest = (now - max_date).total_seconds() / 3600
            is_timely = hours_since_latest <= threshold_hours
            self.results["timeliness"] = {
                "score": 1.0 if is_timely else 0.0,
                "hours_since_latest": round(hours_since_latest, 1),
                "threshold_hours": threshold_hours,
                "status": "PASS" if is_timely else "FAIL",
            }
            return 1.0 if is_timely else 0.0
        self.results["timeliness"] = {"score": 0.0, "status": "NO_DATE_COLUMN"}
        return 0.0

What do these methods check?

• score_validity() runs three format checks: emails match a regex pattern, transaction amounts are positive, and country codes are exactly 2 characters.
• score_consistency() applies cross-field logic. Transaction dates must not be in the future, and Corporate account transactions must have positive amounts.
• score_uniqueness() counts duplicate transaction_id values. Any duplicates lower the score.
• score_timeliness() checks whether the most recent record falls within a 24-hour threshold. If data is stale, AI models may produce outdated insights.

You won't see a change yet. The next chunk ties all 5 methods together.

Calculate dimension scores and generate the profile report

Each scoring method stores its result in the self.results dictionary. Now you need an orchestration method that runs all 5 and a summary printer that formats the output for human review.

• Add the run_full_profile() and print_summary() methods below score_timeliness() in profiler.py:

    def run_full_profile(self):
        self.score_completeness()
        self.score_validity()
        self.score_consistency()
        self.score_uniqueness()
        self.score_timeliness()
        return self.results

    def print_summary(self):
        print("\n===== DATA PROFILE SUMMARY =====")
        for dimension, data in self.results.items():
            score_pct = data["score"] * 100
            print(f"  {dimension.capitalize():15s}: {score_pct:.1f}%")
        print("================================\n")

What does this code do?

run_full_profile() calls each scoring method in sequence and returns the complete results dictionary. This single method triggers the entire profiling pipeline.

print_summary() loops through the stored results and formats each dimension as a percentage. This is the human-readable output a governance analyst would review.

• Save profiler.py.
• Run the profiler against your sample data by running this command:

python -c "
import json
import pandas as pd
from profiler import DataProfiler

df = pd.DataFrame(json.load(open('data/transactions.json')))
profiler = DataProfiler(df)
profiler.run_full_profile()
profiler.print_summary()
"

You should see a formatted table of dimension scores in your terminal, similar to:

===== DATA PROFILE SUMMARY =====
  Completeness   : 96.8%
  Validity       : 98.4%
  Consistency    : 99.0%
  Uniqueness     : 98.1%
  Timeliness     : 100.0%
================================

Why might my numbers differ?

Your exact percentages will vary because the data generator uses random values. The important thing is that every dimension produces a score between 0% and 100%, and at least one dimension shows a score below 100% (reflecting the intentional quality issues in your dataset).

Not seeing output or getting an error?

• Make sure your virtual environment is still active. You should see (venv) at the start of your terminal prompt.
• Confirm you are running the command from inside the ai_readiness_gate/ directory where profiler.py and the data/ folder both live.
• Check for indentation errors in profiler.py. Every method must be indented one level inside the DataProfiler class.

Help me debug my DataProfiler import error

✔️ Awesome, I've got everything!

Great. Double check you have saved profiler.py before moving on.

ⓧ I'd like to double check the full code

import pandas as pd
from datetime import datetime, timedelta
import re


class DataProfiler:
    def __init__(self, df):
        self.df = df
        self.results = {}

    def score_completeness(self):
        total_cells = self.df.size
        non_null_cells = self.df.count().sum()
        score = non_null_cells / total_cells
        null_by_column = self.df.isnull().sum()
        fields_with_nulls = null_by_column[null_by_column > 0].to_dict()
        self.results["completeness"] = {
            "score": round(score, 4),
            "fields_with_nulls": fields_with_nulls,
        }
        return score

    def score_validity(self):
        checks = 0
        passes = 0

        if "email" in self.df.columns:
            email_mask = self.df["email"].dropna()
            email_pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
            valid_emails = email_mask.str.match(email_pattern).sum()
            checks += len(email_mask)
            passes += valid_emails

        if "transaction_amount" in self.df.columns:
            amounts = self.df["transaction_amount"].dropna()
            valid_amounts = (amounts > 0).sum()
            checks += len(amounts)
            passes += valid_amounts

        if "country_code" in self.df.columns:
            codes = self.df["country_code"].dropna()
            valid_codes = (codes.str.len() == 2).sum()
            checks += len(codes)
            passes += valid_codes

        score = passes / checks if checks > 0 else 0.0
        self.results["validity"] = {"score": round(score, 4)}
        return score

    def score_consistency(self):
        checks = 0
        passes = 0

        if "transaction_date" in self.df.columns:
            dates = pd.to_datetime(self.df["transaction_date"], errors="coerce")
            now = pd.Timestamp.now()
            valid_dates = (dates <= now).sum()
            checks += dates.notna().sum()
            passes += valid_dates

        if "transaction_amount" in self.df.columns and "account_type" in self.df.columns:
            corporate = self.df[self.df["account_type"] == "Corporate"]
            valid_corporate = (corporate["transaction_amount"] > 0).sum()
            checks += len(corporate)
            passes += valid_corporate

        score = passes / checks if checks > 0 else 0.0
        self.results["consistency"] = {"score": round(score, 4)}
        return score

    def score_uniqueness(self):
        if "transaction_id" not in self.df.columns:
            self.results["uniqueness"] = {"score": 1.0}
            return 1.0

        total = len(self.df)
        unique = self.df["transaction_id"].nunique()
        duplicates = total - unique
        score = unique / total
        self.results["uniqueness"] = {
            "score": round(score, 4),
            "duplicate_count": duplicates,
        }
        return score

    def score_timeliness(self, threshold_hours=24):
        if "transaction_date" in self.df.columns:
            dates = pd.to_datetime(self.df["transaction_date"], errors="coerce")
            max_date = dates.max()
            now = pd.Timestamp.now()
            hours_since_latest = (now - max_date).total_seconds() / 3600
            is_timely = hours_since_latest <= threshold_hours
            self.results["timeliness"] = {
                "score": 1.0 if is_timely else 0.0,
                "hours_since_latest": round(hours_since_latest, 1),
                "threshold_hours": threshold_hours,
                "status": "PASS" if is_timely else "FAIL",
            }
            return 1.0 if is_timely else 0.0
        self.results["timeliness"] = {"score": 0.0, "status": "NO_DATE_COLUMN"}
        return 0.0

    def run_full_profile(self):
        self.score_completeness()
        self.score_validity()
        self.score_consistency()
        self.score_uniqueness()
        self.score_timeliness()
        return self.results

    def print_summary(self):
        print("\n===== DATA PROFILE SUMMARY =====")
        for dimension, data in self.results.items():
            score_pct = data["score"] * 100
            print(f"  {dimension.capitalize():15s}: {score_pct:.1f}%")
        print("================================\n")

Your profiling engine can now quantify data health across 5 dimensions in a single call. Next up, you will build a sensitivity classifier that detects which fields contain PII and assigns them to security tiers.

Build the PII Sensitivity Classifier

Your data profiler can now score datasets across five quality dimensions. Knowing the quality is important, but there's another critical question before AI touches enterprise data: which fields contain personally identifiable information?

In this step, you'll build a sensitivity classifier that scans every field in your dataset, detects PII using regex pattern matching, and assigns each field to one of four security tiers. This is the layer that prevents an AI agent from accidentally ingesting credit card numbers or Social Security numbers.

In this step, get ready to:

• Build a SensitivityClassifier class with regex-based PII detection and a 4-tier classification system.
• Generate a classification report with exposure metrics showing which fields need masking before AI access.

Build the classifier module

Enterprise data governance requires classifying every field by sensitivity before granting AI access. Your classifier will use a tiered system: Restricted (must be masked), Confidential (limit exposure), Internal (business use only), and Public (safe for broad access).

• Create a new file called classifier.py in your ai_readiness_gate folder.
• Add the class skeleton with tier constants, PII regex patterns, and the field-to-tier mapping by copying this code:

import re
import pandas as pd


class SensitivityClassifier:
    TIER_RESTRICTED = "Restricted"
    TIER_CONFIDENTIAL = "Confidential"
    TIER_INTERNAL = "Internal"
    TIER_PUBLIC = "Public"

    PII_PATTERNS = {
        "credit_card": re.compile(r"^[3-6]\d{14,15}$"),
        "ssn": re.compile(r"^\d{3}-\d{2}-\d{4}$"),
        "email": re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"),
        "phone": re.compile(r"^\+\d{1,3}-\d{7,12}$"),
    }

    FIELD_TIER_MAP = {
        "card_number": TIER_RESTRICTED,
        "ssn": TIER_RESTRICTED,
        "email": TIER_CONFIDENTIAL,
        "phone": TIER_CONFIDENTIAL,
        "customer_name": TIER_CONFIDENTIAL,
        "transaction_id": TIER_INTERNAL,
        "transaction_amount": TIER_INTERNAL,
        "transaction_date": TIER_INTERNAL,
        "account_type": TIER_INTERNAL,
        "merchant_category": TIER_PUBLIC,
        "country_code": TIER_PUBLIC,
    }

    def __init__(self, df):
        self.df = df
        self.classification_map = {}
        self.pii_fields = []
        self.fields_requiring_masking = []

What does this code do?

PII_PATTERNS contains pre-compiled regex patterns for detecting sensitive data formats. For example, credit_card matches 15-16 digit numbers starting with 3-6 (covering Visa, Mastercard, Amex prefixes).

FIELD_TIER_MAP is a lookup table that maps every known field name to its sensitivity tier. Fields like card_number and ssn are Restricted because they must never reach an AI system unmasked.

• Save classifier.py.
• You won't see output yet. The next chunk adds the classification logic that uses these patterns and maps.

Now you need the method that actually classifies a field. It checks the known field map first, then falls back to regex pattern matching on sample data for unknown fields.

• In classifier.py, add the classify_field method below the __init__ method:

    def classify_field(self, column_name):
        # Check the known field mapping first
        if column_name in self.FIELD_TIER_MAP:
            return self.FIELD_TIER_MAP[column_name]

        # For unknown fields, sample values and test against PII regex patterns
        sample = self.df[column_name].dropna().head(100).astype(str)
        for pattern_name, pattern in self.PII_PATTERNS.items():
            match_rate = sample.str.match(pattern).mean()
            if match_rate > 0.5:
                if pattern_name in ["credit_card", "ssn"]:
                    return self.TIER_RESTRICTED
                return self.TIER_CONFIDENTIAL

        return self.TIER_INTERNAL

What does this method do?

classify_field uses a two-pass approach. First, it checks if the column name exists in the known FIELD_TIER_MAP. This handles the 11 fields you know about upfront.

For unknown fields (like if new columns appear in future data), it samples 100 values and tests them against each PII regex. If over 50% match a sensitive pattern, the field gets flagged. This makes the classifier extensible to new datasets without code changes.

• Save classifier.py.
• You won't see output yet. The next substep adds reporting methods and runs the full classifier.

Seeing a red underline on the indentation?

Make sure classify_field is indented at the same level as __init__ (one level inside the class). Python uses indentation to determine what belongs to the class.

help me fix indentation errors in my Python class

Generate classification report and exposure metrics

With the core classification logic in place, you need methods to classify all fields at once, calculate exposure metrics, and output a formatted report.

• In classifier.py, add the following three methods below classify_field:

    def classify_all_fields(self):
        # Iterate through every column and assign a tier
        for column in self.df.columns:
            tier = self.classify_field(column)
            self.classification_map[column] = tier
            if tier in [self.TIER_RESTRICTED, self.TIER_CONFIDENTIAL]:
                self.pii_fields.append(column)
            if tier == self.TIER_RESTRICTED:
                self.fields_requiring_masking.append(column)
        return self.classification_map

    def get_exposure_metrics(self):
        # Calculate summary statistics about PII exposure
        total_fields = len(self.classification_map)
        pii_count = len(self.pii_fields)
        exposure_rate = pii_count / total_fields if total_fields > 0 else 0.0
        tier_counts = {}
        for tier in self.classification_map.values():
            tier_counts[tier] = tier_counts.get(tier, 0) + 1
        return {
            "total_fields": total_fields,
            "pii_field_count": pii_count,
            "exposure_rate": round(exposure_rate, 4),
            "fields_requiring_masking": self.fields_requiring_masking,
            "tier_distribution": tier_counts,
            "classification_coverage": 1.0,
        }

    def print_report(self):
        # Output a formatted classification summary to the console
        metrics = self.get_exposure_metrics()
        print("\n===== SENSITIVITY CLASSIFICATION REPORT =====")
        print(f"  Total fields analyzed: {metrics['total_fields']}")
        print(f"  PII fields detected: {metrics['pii_field_count']}")
        print(f"  Exposure rate: {metrics['exposure_rate']*100:.1f}%")
        print(f"  Classification coverage: 100%")
        print(f"\n  Tier Distribution:")
        for tier, count in metrics["tier_distribution"].items():
            print(f"    {tier}: {count} fields")
        print(f"\n  Fields requiring masking before AI access:")
        for field in self.fields_requiring_masking:
            print(f"    - {field}")
        print("=============================================\n")

What do these methods do?

• classify_all_fields() loops through every column in the DataFrame, classifies it, and populates three instance lists: the full classification map, detected PII fields, and fields requiring masking.
• get_exposure_metrics() calculates the PII exposure rate (sensitive fields divided by total fields) and builds a tier distribution summary. This data feeds into the governance scorecard later.
• print_report() outputs a human-readable console summary showing exactly which fields are flagged and which need masking before AI access.

• Save classifier.py.

✔️ Awesome, I've got everything!

Great. Make sure you've saved classifier.py before running the test below.

ⓧ I'd like to double check the full code

import re
import pandas as pd


class SensitivityClassifier:
    TIER_RESTRICTED = "Restricted"
    TIER_CONFIDENTIAL = "Confidential"
    TIER_INTERNAL = "Internal"
    TIER_PUBLIC = "Public"

    PII_PATTERNS = {
        "credit_card": re.compile(r"^[3-6]\d{14,15}$"),
        "ssn": re.compile(r"^\d{3}-\d{2}-\d{4}$"),
        "email": re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"),
        "phone": re.compile(r"^\+\d{1,3}-\d{7,12}$"),
    }

    FIELD_TIER_MAP = {
        "card_number": TIER_RESTRICTED,
        "ssn": TIER_RESTRICTED,
        "email": TIER_CONFIDENTIAL,
        "phone": TIER_CONFIDENTIAL,
        "customer_name": TIER_CONFIDENTIAL,
        "transaction_id": TIER_INTERNAL,
        "transaction_amount": TIER_INTERNAL,
        "transaction_date": TIER_INTERNAL,
        "account_type": TIER_INTERNAL,
        "merchant_category": TIER_PUBLIC,
        "country_code": TIER_PUBLIC,
    }

    def __init__(self, df):
        self.df = df
        self.classification_map = {}
        self.pii_fields = []
        self.fields_requiring_masking = []

    def classify_field(self, column_name):
        if column_name in self.FIELD_TIER_MAP:
            return self.FIELD_TIER_MAP[column_name]

        sample = self.df[column_name].dropna().head(100).astype(str)
        for pattern_name, pattern in self.PII_PATTERNS.items():
            match_rate = sample.str.match(pattern).mean()
            if match_rate > 0.5:
                if pattern_name in ["credit_card", "ssn"]:
                    return self.TIER_RESTRICTED
                return self.TIER_CONFIDENTIAL

        return self.TIER_INTERNAL

    def classify_all_fields(self):
        for column in self.df.columns:
            tier = self.classify_field(column)
            self.classification_map[column] = tier
            if tier in [self.TIER_RESTRICTED, self.TIER_CONFIDENTIAL]:
                self.pii_fields.append(column)
            if tier == self.TIER_RESTRICTED:
                self.fields_requiring_masking.append(column)
        return self.classification_map

    def get_exposure_metrics(self):
        total_fields = len(self.classification_map)
        pii_count = len(self.pii_fields)
        exposure_rate = pii_count / total_fields if total_fields > 0 else 0.0
        tier_counts = {}
        for tier in self.classification_map.values():
            tier_counts[tier] = tier_counts.get(tier, 0) + 1
        return {
            "total_fields": total_fields,
            "pii_field_count": pii_count,
            "exposure_rate": round(exposure_rate, 4),
            "fields_requiring_masking": self.fields_requiring_masking,
            "tier_distribution": tier_counts,
            "classification_coverage": 1.0,
        }

    def print_report(self):
        metrics = self.get_exposure_metrics()
        print("\n===== SENSITIVITY CLASSIFICATION REPORT =====")
        print(f"  Total fields analyzed: {metrics['total_fields']}")
        print(f"  PII fields detected: {metrics['pii_field_count']}")
        print(f"  Exposure rate: {metrics['exposure_rate']*100:.1f}%")
        print(f"  Classification coverage: 100%")
        print(f"\n  Tier Distribution:")
        for tier, count in metrics["tier_distribution"].items():
            print(f"    {tier}: {count} fields")
        print(f"\n  Fields requiring masking before AI access:")
        for field in self.fields_requiring_masking:
            print(f"    - {field}")
        print("=============================================\n")

Now run the classifier against your sample data to verify it detects all sensitive fields.

• Test the classifier by running this command in your terminal:

python -c "
import json, pandas as pd
from classifier import SensitivityClassifier

with open('data/transactions.json') as f:
    df = pd.DataFrame(json.load(f))

classifier = SensitivityClassifier(df)
classifier.classify_all_fields()
classifier.print_report()
"

• You should see a formatted report showing 11 total fields analyzed, 5 PII fields detected, and card_number and ssn listed under "Fields requiring masking before AI access."

What should I see?

The report shows 100% classification coverage, meaning every field received a tier assignment with zero unclassified fields. The tier distribution breaks down as: 2 Restricted fields (card_number, ssn), 3 Confidential fields (email, phone, customer_name), 4 Internal fields, and 2 Public fields.

Seeing a ModuleNotFoundError?

Make sure you're running the command from inside the ai_readiness_gate directory where classifier.py lives. Python imports look for modules in the current working directory.

Also confirm your virtual environment is still active. You should see the environment name in parentheses at the start of your terminal prompt.

help me debug a ModuleNotFoundError when importing classifier

Your classifier now detects and categorizes every sensitive field in the dataset. Next up, you'll use Great Expectations to enforce 18 business rules as automated data quality checks.

Run Great Expectations Validation

Your profiler and classifier have measured the data's quality and identified sensitive fields. The next question is: does the data comply with specific business rules that your organization enforces?

Custom profiling catches broad patterns, but enterprise governance requires explicit, auditable rules. In this step, you will use Great Expectations (GX) to define a validation suite of business rules and run them against your transaction data. GX produces a compliance rate and auto-generated HTML Data Docs that serve as audit evidence.

In this step, get ready to:

• Connect GX to your pandas DataFrame as an in-memory data source.
• Define business rule expectations covering schema, nulls, ranges, formats, and allowed values.
• Run a GX Checkpoint and extract a compliance rate from the results.

Set up the GX Data Context and connect to data

Every GX workflow starts with a Data Context. This is the central Python object that manages your expectations, validation definitions, and results. You will create an ephemeral (in-memory) context so the validation runs without needing any persistent configuration files.

• Create a new file called gx_validation.py in your ai_readiness_gate/ folder and add the following imports and data connection setup:

import great_expectations as gx
import great_expectations.expectations as gxe
import pandas as pd


def run_validation(df):
    # Initialize an ephemeral (in-memory) Data Context
    context = gx.get_context()

    # Register a pandas data source with the context
    data_source = context.data_sources.add_pandas("transaction_data")
    # Create a dataframe asset representing our transaction records
    data_asset = data_source.add_dataframe_asset(name="transactions")
    # Define a batch that encompasses the entire dataframe
    batch_definition = data_asset.add_batch_definition_whole_dataframe(
        "full_dataset"
    )

What does this code do?

gx.get_context() creates an ephemeral Data Context that lives in memory for the duration of your script. No config files are written to disk.

The three chained calls register your pandas DataFrame as a data source, wrap it as a named asset, and define a batch covering all rows. GX needs this pipeline so it knows how to retrieve data when the Checkpoint runs.

Why Great Expectations instead of custom pandas checks?

Custom pandas validation works, but it produces ad-hoc scripts that auditors cannot easily review. GX produces standardised, human-readable HTML reports (Data Docs) that serve as audit evidence. It also separates rule definitions from execution, which mirrors how enterprise policy frameworks like AEMP70 separate rules from enforcement.

GX is open-source (Apache 2.0) and referenced in governance analyst job postings alongside tools like Collibra. Knowing purpose-built governance tooling demonstrates deeper domain expertise than raw pandas checks.

• Save gx_validation.py.

You won't see output yet. The next chunk adds the business rules that GX will evaluate.

Define business rule expectations

An Expectation Suite is a named collection of data quality rules. Each rule (called an Expectation) tests one property of the data. You will define expectations covering schema validation, null checks, value ranges, format patterns, uniqueness, and allowed-value sets.

• In gx_validation.py, add the following code inside the run_validation function, directly below the batch_definition line:

    # Create an Expectation Suite to hold all business rules
    suite = context.suites.add(
        gx.ExpectationSuite(name="ai_readiness_rules")
    )

    # Rule: all 11 expected columns must be present
    expected_columns = [
        "transaction_id", "customer_name", "email", "phone",
        "card_number", "ssn", "transaction_amount", "transaction_date",
        "merchant_category", "account_type", "country_code",
    ]
    suite.add_expectation(
        gxe.ExpectTableColumnsToMatchSet(
            column_set=expected_columns, exact_match=True
        )
    )

    # Rule: table must have exactly 11 columns
    suite.add_expectation(
        gxe.ExpectTableColumnCountToEqual(value=11)
    )

    # Rules: critical fields must never be null
    for col in ["transaction_id", "customer_name", "transaction_amount",
                "transaction_date", "account_type"]:
        suite.add_expectation(
            gxe.ExpectColumnValuesToNotBeNull(column=col)
        )

    # Rule: transaction amounts must be positive and within range
    suite.add_expectation(
        gxe.ExpectColumnValuesToBeBetween(
            column="transaction_amount", min_value=0.01, max_value=100000.0
        )
    )

    # Rule: at least 90% of emails must match a valid format
    suite.add_expectation(
        gxe.ExpectColumnValuesToMatchRegex(
            column="email",
            regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
            mostly=0.9,
        )
    )

    # Rule: transaction IDs must be unique
    suite.add_expectation(
        gxe.ExpectColumnValuesToBeUnique(column="transaction_id")
    )

    # Rule: account_type must be one of the known values
    suite.add_expectation(
        gxe.ExpectColumnValuesToBeInSet(
            column="account_type",
            value_set=["Platinum", "Gold", "Green", "Corporate", "Blue"],
        )
    )

    # Rule: country codes must be in the allowed set
    suite.add_expectation(
        gxe.ExpectColumnValuesToBeInSet(
            column="country_code",
            value_set=["US", "UK", "IN", "AU", "CA", "DE", "JP", "SG"],
        )
    )

    # Rule: merchant categories must match allowed values (95% threshold)
    suite.add_expectation(
        gxe.ExpectColumnValuesToBeInSet(
            column="merchant_category",
            value_set=["Retail", "Travel", "Dining", "Entertainment",
                       "Grocery", "Fuel", "Healthcare", "Utilities"],
            mostly=0.95,
        )
    )

    # Rule: country codes must be exactly 2 characters
    suite.add_expectation(
        gxe.ExpectColumnValueLengthsToBeBetween(
            column="country_code", min_value=2, max_value=2
        )
    )

    # Rule: transaction IDs must follow the TXN-XXXXXXXX format
    suite.add_expectation(
        gxe.ExpectColumnValuesToMatchRegex(
            column="transaction_id", regex=r"^TXN-\d{8}$"
        )
    )

    # Rule: SSNs must follow the XXX-XX-XXXX format (99% threshold)
    suite.add_expectation(
        gxe.ExpectColumnValuesToMatchRegex(
            column="ssn", regex=r"^\d{3}-\d{2}-\d{4}$", mostly=0.99
        )
    )

What does this code do?

Each suite.add_expectation() call registers one business rule. The rules cover six categories:

• Schema: columns must match the expected set and count.
• Completeness: critical fields cannot be null.
• Ranges: transaction amounts must fall within 0.01 to 100,000.
• Format: emails, transaction IDs, and SSNs must match regex patterns.
• Uniqueness: transaction IDs cannot repeat.
• Allowed values: account types, country codes, and merchant categories must come from known sets.

The mostly parameter on some expectations sets a tolerance threshold. For example, mostly=0.9 means the rule passes if at least 90% of values match, allowing for some known data quality issues without failing the entire check.

• Save gx_validation.py.

You won't see a change yet. The next chunk wires these rules to a Checkpoint and runs them.

Run the validation and extract results

A Checkpoint is GX's execution unit. It links your batch definition (the data) to your expectation suite (the rules) through a Validation Definition, then runs everything and produces results with pass/fail statistics.

• In gx_validation.py, add the following code at the bottom of the run_validation function, below the last suite.add_expectation() call:

    # Link the batch definition and suite into a Validation Definition
    validation_definition = context.validation_definitions.add(
        gx.ValidationDefinition(
            name="ai_readiness_validation",
            data=batch_definition,
            suite=suite,
        )
    )

    # Create a Checkpoint to orchestrate the validation run
    checkpoint = context.checkpoints.add(
        gx.Checkpoint(
            name="ai_readiness_checkpoint",
            validation_definitions=[validation_definition],
        )
    )

    # Execute the checkpoint, passing the DataFrame at runtime
    checkpoint_result = checkpoint.run(
        batch_parameters={"dataframe": df}
    )

    # Extract validation statistics from the results
    run_results = checkpoint_result.run_results
    first_key = list(run_results.keys())[0]
    result = run_results[first_key]

    stats = {
        "evaluated_expectations": result["statistics"]["evaluated_expectations"],
        "successful_expectations": result["statistics"]["successful_expectations"],
        "unsuccessful_expectations": result["statistics"]["unsuccessful_expectations"],
        "success": result["success"],
    }

    # Calculate the compliance rate as a ratio of passed to total
    compliance_rate = (
        stats["successful_expectations"] / stats["evaluated_expectations"]
    )
    stats["compliance_rate"] = round(compliance_rate, 4)

    # Print a summary of the validation run
    print(f"\n===== GX VALIDATION RESULTS =====")
    print(f"  Expectations evaluated: {stats['evaluated_expectations']}")
    print(f"  Passed: {stats['successful_expectations']}")
    print(f"  Failed: {stats['unsuccessful_expectations']}")
    print(f"  Compliance rate: {compliance_rate*100:.1f}%")
    print(f"=================================\n")

    return stats

What does this code do?

The ValidationDefinition binds the suite (rules) to the batch_definition (data). The Checkpoint wraps one or more ValidationDefinitions into a single runnable unit.

checkpoint.run(batch_parameters={"dataframe": df}) passes your pandas DataFrame at runtime. GX evaluates every expectation and stores the results. You then pull out the statistics dictionary to calculate how many rules passed versus failed.

The returned stats dictionary gives the scorecard module everything it needs to compute the Policy Compliance dimension of the final AI-Readiness Score.

• Save gx_validation.py.
• Test the validation by running this command in your terminal:

python -c "import json, pandas as pd; from gx_validation import run_validation; df = pd.DataFrame(json.load(open('data/transactions.json'))); run_validation(df)"

You should see output showing the number of expectations evaluated, how many passed, how many failed, and a compliance rate percentage.

Seeing an import error or unexpected failure?

• Make sure your virtual environment is activated. GX must be importable from the current Python session.
• Confirm you are running the command from inside the ai_readiness_gate/ directory where gx_validation.py and data/transactions.json both exist.
• If you see a KeyError on statistics, double check that the run_results extraction code matches the snippet above exactly.

Help me debug my GX validation run

✔️ Awesome, I've got everything!

Great. Double check you have saved gx_validation.py and that the validation ran successfully with a printed compliance rate.

ⓧ I'd like to double check the full code

import great_expectations as gx
import great_expectations.expectations as gxe
import pandas as pd


def run_validation(df):
    context = gx.get_context()

    data_source = context.data_sources.add_pandas("transaction_data")
    data_asset = data_source.add_dataframe_asset(name="transactions")
    batch_definition = data_asset.add_batch_definition_whole_dataframe(
        "full_dataset"
    )

    suite = context.suites.add(
        gx.ExpectationSuite(name="ai_readiness_rules")
    )

    expected_columns = [
        "transaction_id", "customer_name", "email", "phone",
        "card_number", "ssn", "transaction_amount", "transaction_date",
        "merchant_category", "account_type", "country_code",
    ]
    suite.add_expectation(
        gxe.ExpectTableColumnsToMatchSet(
            column_set=expected_columns, exact_match=True
        )
    )

    suite.add_expectation(
        gxe.ExpectTableColumnCountToEqual(value=11)
    )

    for col in ["transaction_id", "customer_name", "transaction_amount",
                "transaction_date", "account_type"]:
        suite.add_expectation(
            gxe.ExpectColumnValuesToNotBeNull(column=col)
        )

    suite.add_expectation(
        gxe.ExpectColumnValuesToBeBetween(
            column="transaction_amount", min_value=0.01, max_value=100000.0
        )
    )

    suite.add_expectation(
        gxe.ExpectColumnValuesToMatchRegex(
            column="email",
            regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
            mostly=0.9,
        )
    )

    suite.add_expectation(
        gxe.ExpectColumnValuesToBeUnique(column="transaction_id")
    )

    suite.add_expectation(
        gxe.ExpectColumnValuesToBeInSet(
            column="account_type",
            value_set=["Platinum", "Gold", "Green", "Corporate", "Blue"],
        )
    )

    suite.add_expectation(
        gxe.ExpectColumnValuesToBeInSet(
            column="country_code",
            value_set=["US", "UK", "IN", "AU", "CA", "DE", "JP", "SG"],
        )
    )

    suite.add_expectation(
        gxe.ExpectColumnValuesToBeInSet(
            column="merchant_category",
            value_set=["Retail", "Travel", "Dining", "Entertainment",
                       "Grocery", "Fuel", "Healthcare", "Utilities"],
            mostly=0.95,
        )
    )

    suite.add_expectation(
        gxe.ExpectColumnValueLengthsToBeBetween(
            column="country_code", min_value=2, max_value=2
        )
    )

    suite.add_expectation(
        gxe.ExpectColumnValuesToMatchRegex(
            column="transaction_id", regex=r"^TXN-\d{8}$"
        )
    )

    suite.add_expectation(
        gxe.ExpectColumnValuesToMatchRegex(
            column="ssn", regex=r"^\d{3}-\d{2}-\d{4}$", mostly=0.99
        )
    )

    validation_definition = context.validation_definitions.add(
        gx.ValidationDefinition(
            name="ai_readiness_validation",
            data=batch_definition,
            suite=suite,
        )
    )

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

    checkpoint_result = checkpoint.run(
        batch_parameters={"dataframe": df}
    )

    run_results = checkpoint_result.run_results
    first_key = list(run_results.keys())[0]
    result = run_results[first_key]

    stats = {
        "evaluated_expectations": result["statistics"]["evaluated_expectations"],
        "successful_expectations": result["statistics"]["successful_expectations"],
        "unsuccessful_expectations": result["statistics"]["unsuccessful_expectations"],
        "success": result["success"],
    }

    compliance_rate = (
        stats["successful_expectations"] / stats["evaluated_expectations"]
    )
    stats["compliance_rate"] = round(compliance_rate, 4)

    print(f"\n===== GX VALIDATION RESULTS =====")
    print(f"  Expectations evaluated: {stats['evaluated_expectations']}")
    print(f"  Passed: {stats['successful_expectations']}")
    print(f"  Failed: {stats['unsuccessful_expectations']}")
    print(f"  Compliance rate: {compliance_rate*100:.1f}%")
    print(f"=================================\n")

    return stats

Your governance gate now has automated policy enforcement. Next up, you will build the composite AI-Readiness Scorecard that combines profiling, classification, and validation results into a single pass/fail gate decision.

Build the Scorecard and Orchestrate the Pipeline

You now have three independent modules that each assess a different facet of data governance: profiling scores quality, the classifier maps sensitivity, and Great Expectations enforces business rules. Individually, none of them can answer the real question: is this data safe for AI to consume?

In this step, you will build a composite scoring algorithm that combines all three assessments into a single AI-Readiness Score (0-100). You will also implement gate logic that makes an automated pass/fail decision, and create an orchestrator that runs the full pipeline end-to-end.

In this step, get ready to:

• Build a composite scoring algorithm with weighted governance dimensions.
• Implement gate logic with PASS, CONDITIONAL, and BLOCKED decisions plus remediation guidance.
• Create a pipeline orchestrator that produces a timestamped JSON audit trail and HTML scorecard.

Build the scoring algorithm

The scorecard needs to weigh each governance dimension differently. Data quality is the most important factor (40%), while classification coverage, lineage completeness, and policy compliance each contribute 20%.

• Create a new file called scorecard.py in your ai_readiness_gate/ folder.
• Add the class foundation with scoring weights, initialization, and lineage metadata by pasting this code:

import json
import os
from datetime import datetime


class GovernanceScorecard:
    WEIGHT_QUALITY = 0.40
    WEIGHT_CLASSIFICATION = 0.20
    WEIGHT_LINEAGE = 0.20
    WEIGHT_COMPLIANCE = 0.20

    GATE_PASS_THRESHOLD = 80
    GATE_CONDITIONAL_THRESHOLD = 60

    def __init__(self, profile_results, classification_metrics, gx_stats):
        self.profile_results = profile_results
        self.classification_metrics = classification_metrics
        self.gx_stats = gx_stats
        self.lineage_metadata = self._build_lineage()
        self.score = 0
        self.gate_decision = ""
        self.remediation_steps = []

    def _build_lineage(self):
        return {
            "source_system": "Enterprise Data Intelligence Platform",
            "extraction_timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "transformation_steps": [
                "JSON ingestion from source pipeline",
                "Data profiling across 5 quality dimensions",
                "PII sensitivity classification (4-tier)",
                "Great Expectations validation (18 rules)",
                "Composite AI-readiness scoring",
            ],
            "data_owner": "Finance Data Governance Team",
            "field_count_with_lineage": 9,
            "total_field_count": 11,
            "lineage_coverage": round(9 / 11, 4),
        }

What does this code do?

The class constants define the weighted formula: data quality contributes 40% of the final score, while classification, lineage, and compliance each contribute 20%. The gate thresholds set the automated decision boundaries.

The _build_lineage() method creates a data lineage record that documents where the data came from, what transformations were applied, and who owns it. This is critical for enterprise audit trails. The coverage ratio (9/11 fields) reflects that 2 fields lack documented lineage.

• Now add the calculate_score() method below the _build_lineage() method:

    def calculate_score(self):
        quality_scores = [
            self.profile_results.get("completeness", {}).get("score", 0),
            self.profile_results.get("validity", {}).get("score", 0),
            self.profile_results.get("consistency", {}).get("score", 0),
            self.profile_results.get("uniqueness", {}).get("score", 0),
            self.profile_results.get("timeliness", {}).get("score", 0),
        ]
        avg_quality = sum(quality_scores) / len(quality_scores)

        classification_score = self.classification_metrics.get(
            "classification_coverage", 0
        )
        lineage_score = self.lineage_metadata["lineage_coverage"]
        compliance_score = self.gx_stats.get("compliance_rate", 0)

        raw_score = (
            avg_quality * self.WEIGHT_QUALITY
            + classification_score * self.WEIGHT_CLASSIFICATION
            + lineage_score * self.WEIGHT_LINEAGE
            + compliance_score * self.WEIGHT_COMPLIANCE
        )
        self.score = round(raw_score * 100, 1)
        return self.score

What does this code do?

The method averages all 5 quality dimension scores (completeness, validity, consistency, uniqueness, timeliness) into a single data quality figure. It then combines that with the three other governance dimensions using the weighted formula.

The final score is scaled to 0-100 for human readability. A score of 85 means the data passes 85% of governance criteria across all weighted dimensions.

• Save scorecard.py.

You won't see output yet. The next chunk adds the gate decision logic that uses this score.

Implement gate logic and remediation guidance

A raw score is not enough for an enterprise governance gate. The system needs to make a definitive decision and tell the data team exactly what to fix if the data does not pass.

• In scorecard.py, add the gate decision and remediation methods below calculate_score():

    def determine_gate_decision(self):
        if self.score >= self.GATE_PASS_THRESHOLD:
            self.gate_decision = "PASS"
        elif self.score >= self.GATE_CONDITIONAL_THRESHOLD:
            self.gate_decision = "CONDITIONAL"
        else:
            self.gate_decision = "BLOCKED"
        return self.gate_decision

    def generate_remediation(self):
        self.remediation_steps = []

        completeness = self.profile_results.get("completeness", {})
        if completeness.get("score", 1) < 0.95:
            nulls = completeness.get("fields_with_nulls", {})
            self.remediation_steps.append(
                f"Fix null values in {len(nulls)} field(s): {list(nulls.keys())}"
            )

        if self.classification_metrics.get("fields_requiring_masking"):
            fields = self.classification_metrics["fields_requiring_masking"]
            self.remediation_steps.append(
                f"Mask {len(fields)} restricted field(s) before AI access: {fields}"
            )

        if self.gx_stats.get("unsuccessful_expectations", 0) > 0:
            failed = self.gx_stats["unsuccessful_expectations"]
            self.remediation_steps.append(
                f"Resolve {failed} failed data quality expectation(s)"
            )

        lineage_coverage = self.lineage_metadata["lineage_coverage"]
        if lineage_coverage < 1.0:
            missing = (
                self.lineage_metadata["total_field_count"]
                - self.lineage_metadata["field_count_with_lineage"]
            )
            self.remediation_steps.append(
                f"Document lineage for {missing} undocumented field(s)"
            )

        return self.remediation_steps

What does this code do?

The determine_gate_decision() method implements three-tier gate logic. A score of 80 or above means the data is cleared for AI consumption. Scores between 60 and 79 require specific fixes before proceeding. Below 60, AI access is completely blocked.

The generate_remediation() method inspects each governance dimension and generates actionable fix instructions. It checks for null values, PII fields that need masking, failed GX expectations, and undocumented lineage.

• Now add the report generation and file-saving methods below generate_remediation():

    def generate_report(self):
        self.calculate_score()
        self.determine_gate_decision()
        self.generate_remediation()

        report = {
            "report_metadata": {
                "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "report_type": "AI-Readiness Governance Scorecard",
                "version": "1.0",
            },
            "ai_readiness_score": self.score,
            "gate_decision": self.gate_decision,
            "dimension_breakdown": {
                "data_quality": {
                    "weight": "40%",
                    "completeness": self.profile_results.get("completeness", {}).get("score", 0),
                    "validity": self.profile_results.get("validity", {}).get("score", 0),
                    "consistency": self.profile_results.get("consistency", {}).get("score", 0),
                    "uniqueness": self.profile_results.get("uniqueness", {}).get("score", 0),
                    "timeliness": self.profile_results.get("timeliness", {}).get("status", "N/A"),
                },
                "sensitivity_classification": {
                    "weight": "20%",
                    "coverage": self.classification_metrics.get("classification_coverage", 0),
                    "pii_exposure_rate": self.classification_metrics.get("exposure_rate", 0),
                    "fields_requiring_masking": self.classification_metrics.get("fields_requiring_masking", []),
                },
                "lineage_completeness": {
                    "weight": "20%",
                    "coverage": self.lineage_metadata["lineage_coverage"],
                    "documented_fields": self.lineage_metadata["field_count_with_lineage"],
                    "total_fields": self.lineage_metadata["total_field_count"],
                },
                "policy_compliance": {
                    "weight": "20%",
                    "compliance_rate": self.gx_stats.get("compliance_rate", 0),
                    "passed": self.gx_stats.get("successful_expectations", 0),
                    "total": self.gx_stats.get("evaluated_expectations", 0),
                },
            },
            "lineage_metadata": self.lineage_metadata,
            "remediation_guidance": self.remediation_steps,
        }
        return report

    def save_json_report(self, output_dir="reports"):
        os.makedirs(output_dir, exist_ok=True)
        report = self.generate_report()
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filepath = os.path.join(output_dir, f"scorecard_{timestamp}.json")
        with open(filepath, "w") as f:
            json.dump(report, f, indent=2)
        print(f"  JSON scorecard saved: {filepath}")
        return filepath, report

    def save_html_report(self, output_dir="reports"):
        os.makedirs(output_dir, exist_ok=True)
        report = self.generate_report()
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filepath = os.path.join(output_dir, f"scorecard_{timestamp}.html")

        gate_color = {"PASS": "#28a745", "CONDITIONAL": "#ffc107", "BLOCKED": "#dc3545"}
        color = gate_color.get(report["gate_decision"], "#6c757d")

        remediation_html = "".join(
            f"<li>{step}</li>" for step in report["remediation_guidance"]
        )

        html = f"""<!DOCTYPE html>
<html>
<head><title>AI-Readiness Governance Scorecard</title>
<style>
body {{ font-family: Arial, sans-serif; max-width: 900px; margin: 0 auto; padding: 20px; }}
.score-box {{ text-align: center; padding: 30px; border-radius: 10px; background: {color}; color: white; }}
.score-box h1 {{ font-size: 3em; margin: 0; }}
.section {{ margin: 20px 0; padding: 15px; border: 1px solid #ddd; border-radius: 5px; }}
.metric {{ display: flex; justify-content: space-between; padding: 8px 0; border-bottom: 1px solid #eee; }}
</style></head>
<body>
<h1>AI-Readiness Governance Scorecard</h1>
<p>Generated: {report["report_metadata"]["generated_at"]}</p>
<div class="score-box">
<h1>{report["ai_readiness_score"]}/100</h1>
<h2>Gate Decision: {report["gate_decision"]}</h2>
</div>
<div class="section">
<h3>Data Quality (40% weight)</h3>
<div class="metric"><span>Completeness</span><span>{report["dimension_breakdown"]["data_quality"]["completeness"]*100:.1f}%</span></div>
<div class="metric"><span>Validity</span><span>{report["dimension_breakdown"]["data_quality"]["validity"]*100:.1f}%</span></div>
<div class="metric"><span>Consistency</span><span>{report["dimension_breakdown"]["data_quality"]["consistency"]*100:.1f}%</span></div>
<div class="metric"><span>Uniqueness</span><span>{report["dimension_breakdown"]["data_quality"]["uniqueness"]*100:.1f}%</span></div>
<div class="metric"><span>Timeliness</span><span>{report["dimension_breakdown"]["data_quality"]["timeliness"]}</span></div>
</div>
<div class="section">
<h3>Sensitivity Classification (20% weight)</h3>
<div class="metric"><span>Coverage</span><span>100%</span></div>
<div class="metric"><span>PII Exposure Rate</span><span>{report["dimension_breakdown"]["sensitivity_classification"]["pii_exposure_rate"]*100:.1f}%</span></div>
<div class="metric"><span>Fields Requiring Masking</span><span>{report["dimension_breakdown"]["sensitivity_classification"]["fields_requiring_masking"]}</span></div>
</div>
<div class="section">
<h3>Lineage Completeness (20% weight)</h3>
<div class="metric"><span>Coverage</span><span>{report["dimension_breakdown"]["lineage_completeness"]["coverage"]*100:.1f}%</span></div>
<div class="metric"><span>Documented Fields</span><span>{report["dimension_breakdown"]["lineage_completeness"]["documented_fields"]} / {report["dimension_breakdown"]["lineage_completeness"]["total_fields"]}</span></div>
</div>
<div class="section">
<h3>Policy Compliance (20% weight)</h3>
<div class="metric"><span>Compliance Rate</span><span>{report["dimension_breakdown"]["policy_compliance"]["compliance_rate"]*100:.1f}%</span></div>
<div class="metric"><span>Rules Passed</span><span>{report["dimension_breakdown"]["policy_compliance"]["passed"]} / {report["dimension_breakdown"]["policy_compliance"]["total"]}</span></div>
</div>
<div class="section">
<h3>Remediation Guidance</h3>
<ol>{remediation_html}</ol>
</div>
<div class="section">
<h3>Data Lineage</h3>
<div class="metric"><span>Source System</span><span>{report["lineage_metadata"]["source_system"]}</span></div>
<div class="metric"><span>Data Owner</span><span>{report["lineage_metadata"]["data_owner"]}</span></div>
<div class="metric"><span>Extraction Time</span><span>{report["lineage_metadata"]["extraction_timestamp"]}</span></div>
</div>
</body></html>"""

        with open(filepath, "w") as f:
            f.write(html)
        print(f"  HTML scorecard saved: {filepath}")
        return filepath

What does this code do?

The generate_report() method orchestrates the full scoring pipeline internally: calculate the score, make the gate decision, then generate remediation steps. It assembles everything into a structured report dictionary.

The save_json_report() method produces a timestamped JSON file that serves as the audit trail. The save_html_report() method generates a color-coded HTML scorecard where green means PASS, yellow means CONDITIONAL, and red means BLOCKED.

• Save scorecard.py.

You won't see output yet. The next substep builds the orchestrator that wires all modules together.

Seeing a syntax error?

Make sure each method is indented with exactly 4 spaces inside the class body. The f"""...""" triple-quoted string in save_html_report() must use double-brace {{ and }} for literal CSS braces.

Help me debug my scorecard.py file

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

import json
import os
from datetime import datetime


class GovernanceScorecard:
    WEIGHT_QUALITY = 0.40
    WEIGHT_CLASSIFICATION = 0.20
    WEIGHT_LINEAGE = 0.20
    WEIGHT_COMPLIANCE = 0.20

    GATE_PASS_THRESHOLD = 80
    GATE_CONDITIONAL_THRESHOLD = 60

    def __init__(self, profile_results, classification_metrics, gx_stats):
        self.profile_results = profile_results
        self.classification_metrics = classification_metrics
        self.gx_stats = gx_stats
        self.lineage_metadata = self._build_lineage()
        self.score = 0
        self.gate_decision = ""
        self.remediation_steps = []

    def _build_lineage(self):
        return {
            "source_system": "Enterprise Data Intelligence Platform",
            "extraction_timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "transformation_steps": [
                "JSON ingestion from source pipeline",
                "Data profiling across 5 quality dimensions",
                "PII sensitivity classification (4-tier)",
                "Great Expectations validation (18 rules)",
                "Composite AI-readiness scoring",
            ],
            "data_owner": "Finance Data Governance Team",
            "field_count_with_lineage": 9,
            "total_field_count": 11,
            "lineage_coverage": round(9 / 11, 4),
        }

    def calculate_score(self):
        quality_scores = [
            self.profile_results.get("completeness", {}).get("score", 0),
            self.profile_results.get("validity", {}).get("score", 0),
            self.profile_results.get("consistency", {}).get("score", 0),
            self.profile_results.get("uniqueness", {}).get("score", 0),
            self.profile_results.get("timeliness", {}).get("score", 0),
        ]
        avg_quality = sum(quality_scores) / len(quality_scores)

        classification_score = self.classification_metrics.get(
            "classification_coverage", 0
        )
        lineage_score = self.lineage_metadata["lineage_coverage"]
        compliance_score = self.gx_stats.get("compliance_rate", 0)

        raw_score = (
            avg_quality * self.WEIGHT_QUALITY
            + classification_score * self.WEIGHT_CLASSIFICATION
            + lineage_score * self.WEIGHT_LINEAGE
            + compliance_score * self.WEIGHT_COMPLIANCE
        )
        self.score = round(raw_score * 100, 1)
        return self.score

    def determine_gate_decision(self):
        if self.score >= self.GATE_PASS_THRESHOLD:
            self.gate_decision = "PASS"
        elif self.score >= self.GATE_CONDITIONAL_THRESHOLD:
            self.gate_decision = "CONDITIONAL"
        else:
            self.gate_decision = "BLOCKED"
        return self.gate_decision

    def generate_remediation(self):
        self.remediation_steps = []

        completeness = self.profile_results.get("completeness", {})
        if completeness.get("score", 1) < 0.95:
            nulls = completeness.get("fields_with_nulls", {})
            self.remediation_steps.append(
                f"Fix null values in {len(nulls)} field(s): {list(nulls.keys())}"
            )

        if self.classification_metrics.get("fields_requiring_masking"):
            fields = self.classification_metrics["fields_requiring_masking"]
            self.remediation_steps.append(
                f"Mask {len(fields)} restricted field(s) before AI access: {fields}"
            )

        if self.gx_stats.get("unsuccessful_expectations", 0) > 0:
            failed = self.gx_stats["unsuccessful_expectations"]
            self.remediation_steps.append(
                f"Resolve {failed} failed data quality expectation(s)"
            )

        lineage_coverage = self.lineage_metadata["lineage_coverage"]
        if lineage_coverage < 1.0:
            missing = (
                self.lineage_metadata["total_field_count"]
                - self.lineage_metadata["field_count_with_lineage"]
            )
            self.remediation_steps.append(
                f"Document lineage for {missing} undocumented field(s)"
            )

        return self.remediation_steps

    def generate_report(self):
        self.calculate_score()
        self.determine_gate_decision()
        self.generate_remediation()

        report = {
            "report_metadata": {
                "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "report_type": "AI-Readiness Governance Scorecard",
                "version": "1.0",
            },
            "ai_readiness_score": self.score,
            "gate_decision": self.gate_decision,
            "dimension_breakdown": {
                "data_quality": {
                    "weight": "40%",
                    "completeness": self.profile_results.get("completeness", {}).get("score", 0),
                    "validity": self.profile_results.get("validity", {}).get("score", 0),
                    "consistency": self.profile_results.get("consistency", {}).get("score", 0),
                    "uniqueness": self.profile_results.get("uniqueness", {}).get("score", 0),
                    "timeliness": self.profile_results.get("timeliness", {}).get("status", "N/A"),
                },
                "sensitivity_classification": {
                    "weight": "20%",
                    "coverage": self.classification_metrics.get("classification_coverage", 0),
                    "pii_exposure_rate": self.classification_metrics.get("exposure_rate", 0),
                    "fields_requiring_masking": self.classification_metrics.get("fields_requiring_masking", []),
                },
                "lineage_completeness": {
                    "weight": "20%",
                    "coverage": self.lineage_metadata["lineage_coverage"],
                    "documented_fields": self.lineage_metadata["field_count_with_lineage"],
                    "total_fields": self.lineage_metadata["total_field_count"],
                },
                "policy_compliance": {
                    "weight": "20%",
                    "compliance_rate": self.gx_stats.get("compliance_rate", 0),
                    "passed": self.gx_stats.get("successful_expectations", 0),
                    "total": self.gx_stats.get("evaluated_expectations", 0),
                },
            },
            "lineage_metadata": self.lineage_metadata,
            "remediation_guidance": self.remediation_steps,
        }
        return report

    def save_json_report(self, output_dir="reports"):
        os.makedirs(output_dir, exist_ok=True)
        report = self.generate_report()
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filepath = os.path.join(output_dir, f"scorecard_{timestamp}.json")
        with open(filepath, "w") as f:
            json.dump(report, f, indent=2)
        print(f"  JSON scorecard saved: {filepath}")
        return filepath, report

    def save_html_report(self, output_dir="reports"):
        os.makedirs(output_dir, exist_ok=True)
        report = self.generate_report()
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filepath = os.path.join(output_dir, f"scorecard_{timestamp}.html")

        gate_color = {"PASS": "#28a745", "CONDITIONAL": "#ffc107", "BLOCKED": "#dc3545"}
        color = gate_color.get(report["gate_decision"], "#6c757d")

        remediation_html = "".join(
            f"<li>{step}</li>" for step in report["remediation_guidance"]
        )

        html = f"""<!DOCTYPE html>
<html>
<head><title>AI-Readiness Governance Scorecard</title>
<style>
body {{ font-family: Arial, sans-serif; max-width: 900px; margin: 0 auto; padding: 20px; }}
.score-box {{ text-align: center; padding: 30px; border-radius: 10px; background: {color}; color: white; }}
.score-box h1 {{ font-size: 3em; margin: 0; }}
.section {{ margin: 20px 0; padding: 15px; border: 1px solid #ddd; border-radius: 5px; }}
.metric {{ display: flex; justify-content: space-between; padding: 8px 0; border-bottom: 1px solid #eee; }}
</style></head>
<body>
<h1>AI-Readiness Governance Scorecard</h1>
<p>Generated: {report["report_metadata"]["generated_at"]}</p>
<div class="score-box">
<h1>{report["ai_readiness_score"]}/100</h1>
<h2>Gate Decision: {report["gate_decision"]}</h2>
</div>
<div class="section">
<h3>Data Quality (40% weight)</h3>
<div class="metric"><span>Completeness</span><span>{report["dimension_breakdown"]["data_quality"]["completeness"]*100:.1f}%</span></div>
<div class="metric"><span>Validity</span><span>{report["dimension_breakdown"]["data_quality"]["validity"]*100:.1f}%</span></div>
<div class="metric"><span>Consistency</span><span>{report["dimension_breakdown"]["data_quality"]["consistency"]*100:.1f}%</span></div>
<div class="metric"><span>Uniqueness</span><span>{report["dimension_breakdown"]["data_quality"]["uniqueness"]*100:.1f}%</span></div>
<div class="metric"><span>Timeliness</span><span>{report["dimension_breakdown"]["data_quality"]["timeliness"]}</span></div>
</div>
<div class="section">
<h3>Sensitivity Classification (20% weight)</h3>
<div class="metric"><span>Coverage</span><span>100%</span></div>
<div class="metric"><span>PII Exposure Rate</span><span>{report["dimension_breakdown"]["sensitivity_classification"]["pii_exposure_rate"]*100:.1f}%</span></div>
<div class="metric"><span>Fields Requiring Masking</span><span>{report["dimension_breakdown"]["sensitivity_classification"]["fields_requiring_masking"]}</span></div>
</div>
<div class="section">
<h3>Lineage Completeness (20% weight)</h3>
<div class="metric"><span>Coverage</span><span>{report["dimension_breakdown"]["lineage_completeness"]["coverage"]*100:.1f}%</span></div>
<div class="metric"><span>Documented Fields</span><span>{report["dimension_breakdown"]["lineage_completeness"]["documented_fields"]} / {report["dimension_breakdown"]["lineage_completeness"]["total_fields"]}</span></div>
</div>
<div class="section">
<h3>Policy Compliance (20% weight)</h3>
<div class="metric"><span>Compliance Rate</span><span>{report["dimension_breakdown"]["policy_compliance"]["compliance_rate"]*100:.1f}%</span></div>
<div class="metric"><span>Rules Passed</span><span>{report["dimension_breakdown"]["policy_compliance"]["passed"]} / {report["dimension_breakdown"]["policy_compliance"]["total"]}</span></div>
</div>
<div class="section">
<h3>Remediation Guidance</h3>
<ol>{remediation_html}</ol>
</div>
<div class="section">
<h3>Data Lineage</h3>
<div class="metric"><span>Source System</span><span>{report["lineage_metadata"]["source_system"]}</span></div>
<div class="metric"><span>Data Owner</span><span>{report["lineage_metadata"]["data_owner"]}</span></div>
<div class="metric"><span>Extraction Time</span><span>{report["lineage_metadata"]["extraction_timestamp"]}</span></div>
</div>
</body></html>"""

        with open(filepath, "w") as f:
            f.write(html)
        print(f"  HTML scorecard saved: {filepath}")
        return filepath

Create the pipeline orchestrator

The individual modules are ready. Now you need a single entry point that loads the data, runs every assessment in sequence, and produces the final scorecard.

• Create a new file called main.py in your ai_readiness_gate/ folder.
• Add the full orchestrator pipeline by pasting this code:

import json
import time
import pandas as pd
from profiler import DataProfiler
from classifier import SensitivityClassifier
from gx_validation import run_validation
from scorecard import GovernanceScorecard


def main():
    print("=" * 60)
    print("  AI-READINESS GOVERNANCE GATE")
    print("  Enterprise Data Assessment Pipeline")
    print("=" * 60)

    start_time = time.time()

    print("\n[1/5] Loading data...")
    with open("data/transactions.json", "r") as f:
        records = json.load(f)
    df = pd.DataFrame(records)
    print(f"  Loaded {len(df)} records with {len(df.columns)} fields")

    print("\n[2/5] Running data profiler...")
    profiler = DataProfiler(df)
    profile_results = profiler.run_full_profile()
    profiler.print_summary()

    print("\n[3/5] Running sensitivity classifier...")
    classifier = SensitivityClassifier(df)
    classifier.classify_all_fields()
    classification_metrics = classifier.get_exposure_metrics()
    classifier.print_report()

    print("\n[4/5] Running Great Expectations validation...")
    gx_stats = run_validation(df)

    print("\n[5/5] Generating AI-Readiness Scorecard...")
    scorecard = GovernanceScorecard(
        profile_results, classification_metrics, gx_stats
    )
    json_path, report = scorecard.save_json_report()
    html_path = scorecard.save_html_report()

    elapsed = time.time() - start_time

    print("\n" + "=" * 60)
    print("  GOVERNANCE GATE RESULT")
    print("=" * 60)
    print(f"  AI-Readiness Score: {report['ai_readiness_score']}/100")
    print(f"  Gate Decision: {report['gate_decision']}")
    print(f"  Records processed: {len(df)}")
    print(f"  Processing time: {elapsed:.1f} seconds")
    print(f"  Throughput: {len(df)/elapsed:,.0f} records/second")
    print(f"\n  Issues found:")
    for step in report["remediation_guidance"]:
        print(f"    - {step}")
    print("=" * 60)


if __name__ == "__main__":
    main()

What does this code do?

The main() function runs the 5-stage governance pipeline in sequence. It loads the JSON data into a DataFrame, profiles it for quality, classifies field sensitivity, validates against 18 GX expectations, and finally generates the composite scorecard.

The timer tracks total processing time so you can verify the gate processes 50,000+ records in under 15 seconds. The throughput metric gives you a records-per-second figure to demonstrate the pipeline's efficiency.

• Save main.py.
• Run the full governance pipeline by running this command:

python main.py

You should see output for all 5 pipeline stages, ending with a summary that shows the AI-Readiness Score, gate decision, processing time, and a list of issues found.

What should I see?

The final output prints an AI-Readiness Score between 70-85 (depending on random data generation), a gate decision of CONDITIONAL or PASS, and remediation items like masking restricted fields and resolving failed expectations. Processing time should be well under 15 seconds.

Two new files appear in the reports/ directory: a timestamped JSON audit file and an HTML scorecard. Open the HTML file in your browser to see the color-coded governance report.

• Confirm the report files were created by running:

ls reports/

You should see two files: scorecard_<timestamp>.json and scorecard_<timestamp>.html.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated and that you are running python main.py from inside the ai_readiness_gate/ directory where all your .py files live. The imports rely on the modules being in the same folder.

Help me fix a ModuleNotFoundError when running main.py

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and that both report files exist in reports/.

ⓧ I'd like to double check the full code

import json
import time
import pandas as pd
from profiler import DataProfiler
from classifier import SensitivityClassifier
from gx_validation import run_validation
from scorecard import GovernanceScorecard


def main():
    print("=" * 60)
    print("  AI-READINESS GOVERNANCE GATE")
    print("  Enterprise Data Assessment Pipeline")
    print("=" * 60)

    start_time = time.time()

    print("\n[1/5] Loading data...")
    with open("data/transactions.json", "r") as f:
        records = json.load(f)
    df = pd.DataFrame(records)
    print(f"  Loaded {len(df)} records with {len(df.columns)} fields")

    print("\n[2/5] Running data profiler...")
    profiler = DataProfiler(df)
    profile_results = profiler.run_full_profile()
    profiler.print_summary()

    print("\n[3/5] Running sensitivity classifier...")
    classifier = SensitivityClassifier(df)
    classifier.classify_all_fields()
    classification_metrics = classifier.get_exposure_metrics()
    classifier.print_report()

    print("\n[4/5] Running Great Expectations validation...")
    gx_stats = run_validation(df)

    print("\n[5/5] Generating AI-Readiness Scorecard...")
    scorecard = GovernanceScorecard(
        profile_results, classification_metrics, gx_stats
    )
    json_path, report = scorecard.save_json_report()
    html_path = scorecard.save_html_report()

    elapsed = time.time() - start_time

    print("\n" + "=" * 60)
    print("  GOVERNANCE GATE RESULT")
    print("=" * 60)
    print(f"  AI-Readiness Score: {report['ai_readiness_score']}/100")
    print(f"  Gate Decision: {report['gate_decision']}")
    print(f"  Records processed: {len(df)}")
    print(f"  Processing time: {elapsed:.1f} seconds")
    print(f"  Throughput: {len(df)/elapsed:,.0f} records/second")
    print(f"\n  Issues found:")
    for step in report["remediation_guidance"]:
        print(f"    - {step}")
    print("=" * 60)


if __name__ == "__main__":
    main()

Your governance gate is fully operational. You have a production-ready pipeline that profiles, classifies, validates, scores, and reports on enterprise data in seconds. Next up, you will extend it with historical trend tracking in the Secret Mission.

Secret mission

Add Historical Governance Trend Tracking

Your governance gate gives a point-in-time snapshot, but enterprise governance programs track posture over time. Can you build a trend tracker that compares each scan to previous runs and reports whether your data governance is improving or degrading?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs or cloud resources to worry about. Decide whether to keep your files for future reference, pause your environment, or delete everything.

Resources you used:

• Project directory ai_readiness_gate/ with all Python source files.
• Python virtual environment (inside the project directory).
• Generated sample data in data/transactions.json (approximately 50,000 records).
• Governance reports in reports/ (HTML scorecards, JSON audit trails, governance history).

Keep everything running

No action needed. Choose this if you plan to keep iterating on your governance gate or want to reference the code later.

• Your project files stay on your Desktop in the ai_readiness_gate/ folder.
• You can rerun python main.py anytime to add new entries to your governance history.
• Re-activate the virtual environment with source .venv/bin/activate (macOS/Linux) or .venv\Scripts\activate (Windows) before running scripts.

Pause - I'll come back to this later

Deactivate your virtual environment to free up your terminal session. Your files remain on disk for when you return.

• Deactivate the virtual environment by running:

deactivate

• When you come back, navigate to ai_readiness_gate/ and re-activate the environment with source .venv/bin/activate (macOS/Linux) or .venv\Scripts\activate (Windows).

Delete - I don't want to use this again

Remove all project files, the virtual environment, and generated data. This cannot be undone.

• Deactivate the virtual environment first (if still active) by running:

deactivate

• Navigate to your Desktop by running:

cd ~/Desktop

• Delete the entire project directory by running this command (macOS/Linux):

rm -rf ai_readiness_gate

• On Windows, run this instead:

rmdir /s /q ai_readiness_gate

• Confirm the folder is gone by running:

ls

• You should no longer see ai_readiness_gate in the output.

Well Done!

Well Done!

You just built a production-grade AI-Readiness Governance Gate that profiles, classifies, validates, and scores enterprise data before any AI system can touch it.

You've learned how to:

• Build a data profiling engine that scores 50,000+ records across 5 quality dimensions and a PII sensitivity classifier that detects and classifies 100% of fields into 4 security tiers using regex pattern matching.
• Implement 18 Great Expectations validation rules as automated business-rule enforcement, producing Data Docs for audit readiness.
• Build a composite AI-Readiness scoring algorithm with weighted governance dimensions, automated pass/fail gate logic, data lineage metadata, and production-ready HTML scorecards with timestamped audit trails and remediation guidance.
• Secret Mission: Added historical governance trend tracking to monitor governance posture over time across multiple scan runs.

Ready to quiz yourself?