Skip to content

Synthetic Data with Error Injection

This section demonstrates deterministic error injection — seeded, reproducible corruption applied as the last pipeline stage, for testing data-quality and reconciliation pipelines: spelled-out numbers in numeric columns (1200 → "one thousand, two hundred"), missing values, broken mappings, duplicated rows, and a corrupted derived summary that no longer reconciles with its detail.

Example Code

"""
Example: deterministic error injection with the FakerSynthesizer.

Perfect data doesn't test anything. This example generates a clean supermarket
daily-sales table plus a derived department summary, then injects seeded,
reproducible errors as the LAST pipeline stage — identities and aggregates are
computed on clean data, corruption exists only in the delivered output:

  - format_violation "words":  1200.0 -> "one thousand, two hundred"
    (the column is deliberately delivered as a string column — the untouched
    numbers rendered as text next to the corrupted cells — downstream parsers
    must survive exactly that)
  - missing values at a controlled rate
  - mapping_break: the strict sku -> department mapping violated in a
    fraction of rows
  - duplicate_rows: duplicated identifiers appended
  - out_of_range ON THE DERIVED SUMMARY: the detail no longer reconciles to
    the aggregate — the "unresolved mismatch" a data-quality or reconciliation
    engine must alert on.

Everything is reproducible under random_state: the same run always corrupts
the same cells, so a DQ test suite can assert exactly which rows get flagged.
"""
import pandas as pd

from ydata.metadata import Metadata
from ydata.metadata.builder import MetadataConfigurationBuilder
from ydata.synthesizers import FakerSynthesizer

products = pd.DataFrame({
    "sku": ["SKU-0001", "SKU-0002", "SKU-0003", "SKU-0004", "SKU-0005"],
    "product_name": ["Semi-Skimmed Milk 2L", "Sourdough Loaf", "Bananas 1kg",
                     "Frozen Peas 750g", "Orange Juice 1L"],
    "department": ["dairy", "bakery", "produce", "frozen", "beverages"],
})

columns = {
    "units_sold": {"datatype": "numerical", "vartype": "int", "min": 1, "max": 500},
    "revenue": {"datatype": "numerical", "vartype": "float", "min": 1.0, "max": 2000.0},
}
metadata = Metadata(configuration_builder=MetadataConfigurationBuilder(columns))

synth = FakerSynthesizer(locale="en")
synth.fit(
    metadata,
    table_name="daily_sales",
    scaffold={
        "dimensions": [
            {"column": "sku", "from_table": "products",
             "from_column": "sku", "include": ["product_name", "department"]},
            {"column": "sales_date",
             "date_range": {"start": "2026-03-01", "end": "2026-03-31", "freq": "D"}},
        ]
    },
    reference_data={"products": products},
    derived_tables=[
        {"name": "department_summary", "type": "aggregate",
         "group_by": ["sales_date", "department"],
         "aggregations": {
             "total_revenue": {"column": "revenue", "agg": "sum"},
             "total_units": {"column": "units_sold", "agg": "sum"},
         }},
    ],
    errors={
        "daily_sales": {
            "columns": {
                "revenue": {
                    "format_violation": {"rate": 0.05, "styles": ["words", "currency"]},
                    "missing": 0.03,
                },
                "units_sold": {"out_of_range": {"rate": 0.02, "factor": 1000}},
            },
            "table": {
                "duplicate_rows": 0.02,
                "mapping_break": {"columns": ["sku", "department"], "rate": 0.03},
            },
        },
        "department_summary": {
            "columns": {"total_revenue": {"out_of_range": {"rate": 0.1, "factor": 100}}},
        },
    },
)

data = synth.sample(random_state=42)
sales = data["daily_sales"].to_pandas()
summary = data["department_summary"].to_pandas()

# ---------------------------------------------------------------------------
# What a data-quality pipeline should now find
# ---------------------------------------------------------------------------
# the whole column is text now; the corrupted cells are the ones that no
# longer parse as a plain number
text_cells = sales["revenue"].notna() & ~sales["revenue"].astype(str).str.fullmatch(r"-?\d+(\.\d+)?")
print("Representation errors (e.g. spelled-out numbers):")
print(sales.loc[text_cells, ["sku", "sales_date", "revenue"]].head(5).to_string())
assert text_cells.any() and pd.api.types.is_string_dtype(sales["revenue"])

print(f"\nMissing revenues injected: {sales['revenue'].isna().sum()}")
assert sales["revenue"].isna().any()

truth = dict(zip(products["sku"], products["department"]))
broken = sales[[truth[s] != d for s, d in zip(sales["sku"], sales["department"])]]
print(f"Broken sku->department mappings: {len(broken)}")
assert len(broken) >= 1

duplicates = sales.duplicated().sum()
print(f"Duplicated rows: {duplicates}")
assert duplicates >= 1

# absurd-but-numeric outliers (units x1000, far beyond the 1-500 domain)
outliers = pd.to_numeric(sales["units_sold"], errors="coerce") > 1000
print(f"Out-of-range unit counts: {outliers.sum()}")
assert outliers.any()

# reconciliation now fails on both fronts: the summary was corrupted AFTER
# derivation, and the corrupted detail (broken mappings, duplicates, text
# cells) no longer rolls up to the clean groups either — align on the
# summary's groups to count the disagreements
parseable_revenue = pd.to_numeric(sales["revenue"], errors="coerce")
recomputed = parseable_revenue.groupby([sales["sales_date"], sales["department"]]).sum()
reported = summary.set_index(["sales_date", "department"])["total_revenue"]
recomputed = recomputed.reindex(reported.index)
mismatches = (reported.round(2) != recomputed.round(2)).sum()
print(f"Date/department groups where the summary does NOT reconcile: {mismatches}")
assert mismatches >= 1

print("\nAll injected error classes are present and reproducible under random_state=42.")