src/utils/validate_data.py

import great_expectations as gx
import pandas as pd


def validate_PaySim_data(df: pd.DataFrame):
    """Validate the processed PaySim frame with Great Expectations (1.x API).

    Runs schema + business-logic checks that must pass before training.
    Returns the validation result; prints a pass/fail summary.
    """
    print('Starting data validation with Great Expectations...')

    context = gx.get_context()
    batch = (
        context.data_sources.add_pandas("paysim")
        .add_dataframe_asset("processed")
        .add_batch_definition_whole_dataframe("batch")
        .get_batch(batch_parameters={"dataframe": df})
    )

    expectations = [
        # --- schema: columns that must exist ---
        gx.expectations.ExpectColumnToExist(column="step"),
        gx.expectations.ExpectColumnToExist(column="amount"),
        gx.expectations.ExpectColumnToExist(column="oldbalanceOrg"),   # note: Org, no i
        gx.expectations.ExpectColumnToExist(column="newbalanceOrig"),
        gx.expectations.ExpectColumnToExist(column="oldbalanceDest"),
        gx.expectations.ExpectColumnToExist(column="newbalanceDest"),
        gx.expectations.ExpectColumnToExist(column="isFraud"),
        gx.expectations.ExpectColumnToExist(column="type_CASH_OUT"),
        gx.expectations.ExpectColumnToExist(column="type_TRANSFER"),
        gx.expectations.ExpectColumnToExist(column="orig_balance_missing"),
        gx.expectations.ExpectColumnToExist(column="dest_balance_missing"),
        gx.expectations.ExpectColumnToExist(column="errorBalanceOrig"),
        gx.expectations.ExpectColumnToExist(column="errorBalanceDest"),

        # --- not null ---
        gx.expectations.ExpectColumnValuesToNotBeNull(column="amount"),
        gx.expectations.ExpectColumnValuesToNotBeNull(column="errorBalanceOrig"),
        gx.expectations.ExpectColumnValuesToNotBeNull(column="errorBalanceDest"),

        # --- binary / set membership ---
        gx.expectations.ExpectColumnValuesToBeInSet(column="isFraud", value_set=[0, 1]),
        gx.expectations.ExpectColumnValuesToBeInSet(column="type_CASH_OUT", value_set=[0, 1]),
        gx.expectations.ExpectColumnValuesToBeInSet(column="type_TRANSFER", value_set=[0, 1]),
        gx.expectations.ExpectColumnValuesToBeInSet(column="orig_balance_missing", value_set=[0, 1]),
        gx.expectations.ExpectColumnValuesToBeInSet(column="dest_balance_missing", value_set=[0, 1]),

        # --- ranges ---
        gx.expectations.ExpectColumnValuesToBeBetween(column="amount", min_value=0),
        gx.expectations.ExpectColumnValuesToBeBetween(column="step", min_value=1, max_value=743),
    ]

    results = []
    for exp in expectations:
        r = batch.validate(exp)
        results.append(r)
        status = "PASS" if r.success else "FAIL"
        print(f"[{status}] {exp.__class__.__name__} — {getattr(exp, 'column', '')}")

    all_passed = all(r.success for r in results)
    print(f"\nValidation {'PASSED' if all_passed else 'FAILED'} "
          f"({sum(r.success for r in results)}/{len(results)} checks)")
    return results