5 Practical Python Checks for Time Series, Drift, and Data Integrity

Basic checks for nulls and duplicates are necessary, but they do not catch every data-quality failure. A row can contain valid-looking values while violating a business rule, a timestamp sequence can hide gaps, and two valid tables can produce an invalid relationship.

The five patterns below return evidence about a problem rather than silently “fixing” the data. Adapt the column names, frequencies, and rules to your domain, then run the checks at ingestion and before important downstream jobs.

Setup and validation approach

python -m pip install pandas scipy networkx

Keep three outcomes separate:

• Failure: data is unsafe to publish or process, such as a broken required key.
• Warning: a change needs review, such as a distribution shift.
• Metric: a value to monitor over time, such as missing timestamps per day.

Do not hard-code one threshold for every dataset. Establish expectations from the business process, record the evidence, and version the rules with the pipeline.

1. Check time-series order, duplicates, and gaps

A timestamp column can contain invalid values, repeated timestamps, rows that move backward in arrival order, or missing intervals. The function below checks all four for a dataset expected at a fixed frequency.

import pandas as pd

def check_time_series(df, timestamp_col, expected_freq):
    timestamps = pd.to_datetime(
        df[timestamp_col], errors="coerce", utc=True
    )

    invalid_rows = df.loc[timestamps.isna()].copy()
    duplicate_rows = df.loc[
        timestamps.notna() & timestamps.duplicated(keep=False)
    ].copy()

    # Detect reversals in the original row order.
    backwards_rows = df.loc[timestamps.diff().lt(pd.Timedelta(0))].copy()

    valid = pd.DatetimeIndex(timestamps.dropna().sort_values().unique())
    if valid.empty:
        missing_timestamps = pd.DatetimeIndex([])
    else:
        expected = pd.date_range(
            start=valid.min(),
            end=valid.max(),
            freq=expected_freq,
            tz="UTC",
        )
        missing_timestamps = expected.difference(valid)

    return {
        "invalid_rows": invalid_rows,
        "duplicate_rows": duplicate_rows,
        "backwards_rows": backwards_rows,
        "missing_timestamps": missing_timestamps,
    }

issues = check_time_series(events, "event_time", "15min")
print("Missing intervals:", len(issues["missing_timestamps"]))

Convert time zones deliberately before comparing timestamps. A daylight-saving transition can look like a duplicate or gap when local times are stored without an offset. Also limit the expected date range: generating every second across several years can consume substantial memory.

For irregular event streams, a fixed date_range is the wrong test. Instead, monitor inter-arrival times, maximum allowed silence, or sequence numbers defined by the source system. The pandas DatetimeIndex documentation covers frequency-aware indexes.

2. Encode cross-field business rules

Schema validation may accept both created_at and delivered_at as valid dates even when delivery occurs first. Make each business rule explicit and return the violating rows with a rule name.

import pandas as pd

def business_rule_violations(orders):
    data = orders.copy()
    data["created_at"] = pd.to_datetime(
        data["created_at"], errors="coerce", utc=True
    )
    data["delivered_at"] = pd.to_datetime(
        data["delivered_at"], errors="coerce", utc=True
    )

    rules = {
        "delivered_before_created": (
            data["delivered_at"].notna()
            & data["created_at"].notna()
            & data["delivered_at"].lt(data["created_at"])
        ),
        "delivered_status_without_date": (
            data["status"].eq("delivered")
            & data["delivered_at"].isna()
        ),
        "negative_total": data["order_total"].lt(0),
        "refund_exceeds_total": (
            data["refund_total"].fillna(0)
            .gt(data["order_total"].fillna(0))
        ),
    }

    problems = []
    for rule_name, mask in rules.items():
        matches = data.loc[mask].copy()
        if not matches.empty:
            matches["failed_rule"] = rule_name
            problems.append(matches)

    return (
        pd.concat(problems, ignore_index=True)
        if problems
        else pd.DataFrame(columns=[*data.columns, "failed_rule"])
    )

violations = business_rule_violations(orders)

Decide how nulls affect each rule instead of letting pandas choose implicitly. For example, a missing delivery date may be valid for an open order but invalid when the status is delivered. Include stable row identifiers in every report so the source record can be traced.

3. Detect schema changes and distribution drift separately

Schema drift means the shape or type contract changed: a column was added, removed, or parsed differently. Distribution drift means the values changed while the schema remained valid. They require different checks.

Schema comparison

def schema_snapshot(df):
    return {
        "columns": list(df.columns),
        "dtypes": {name: str(dtype) for name, dtype in df.dtypes.items()},
    }

def compare_schema(reference, current):
    ref_cols = set(reference["columns"])
    cur_cols = set(current["columns"])
    common = ref_cols & cur_cols

    return {
        "missing_columns": sorted(ref_cols - cur_cols),
        "new_columns": sorted(cur_cols - ref_cols),
        "type_changes": {
            column: {
                "expected": reference["dtypes"][column],
                "current": current["dtypes"][column],
            }
            for column in sorted(common)
            if reference["dtypes"][column] != current["dtypes"][column]
        },
    }

baseline = schema_snapshot(training_data)
schema_issues = compare_schema(baseline, new_batch)

A dtype string is only a first line of defense. A column of numeric identifiers may still parse as integers while losing leading zeros. Add domain checks for formats, allowed values, ranges, units, and nullable fields.

Numeric distribution comparison

from scipy.stats import ks_2samp

def compare_numeric_distribution(reference, current):
    ref = pd.Series(reference).dropna()
    cur = pd.Series(current).dropna()

    if len(ref) < 2 or len(cur) < 2:
        return {"status": "insufficient_data"}

    result = ks_2samp(ref, cur)
    return {
        "status": "ok",
        "reference_count": len(ref),
        "current_count": len(cur),
        "ks_statistic": float(result.statistic),
        "p_value": float(result.pvalue),
        "reference_median": float(ref.median()),
        "current_median": float(cur.median()),
    }

drift = compare_numeric_distribution(
    training_data["order_total"],
    new_batch["order_total"],
)

The two-sample Kolmogorov–Smirnov test compares continuous distributions. A small p-value is not automatically a production alert: with large samples, a small and harmless difference can be statistically detectable. Review effect size, sample size, seasonality, data-collection changes, and business impact. Use category-frequency checks for categorical fields and domain-specific metrics for model features. See SciPy’s KS test documentation for assumptions and return values.

4. Find hierarchy cycles and missing parents

Product categories, reporting lines, and folder trees are usually expected to be directed acyclic graphs. A cycle such as A ? B ? C ? A can break recursion and aggregation. A parent ID that never appears as a node creates an orphan.

import pandas as pd
import networkx as nx

def check_hierarchy(nodes, id_col="node_id", parent_col="parent_id"):
    ids = set(nodes[id_col].dropna())
    parent_ids = set(nodes[parent_col].dropna())
    missing_parents = sorted(parent_ids - ids)

    edges = [
        (parent, child)
        for child, parent in nodes[[id_col, parent_col]].itertuples(
            index=False, name=None
        )
        if pd.notna(parent) and parent in ids
    ]

    graph = nx.DiGraph()
    graph.add_nodes_from(ids)
    graph.add_edges_from(edges)

    cycle = None
    if not nx.is_directed_acyclic_graph(graph):
        cycle = nx.find_cycle(graph, orientation="original")

    roots = nodes.loc[nodes[parent_col].isna(), id_col].tolist()

    return {
        "missing_parents": missing_parents,
        "cycle": cycle,
        "root_ids": roots,
        "isolated_nodes": list(nx.isolates(graph)),
    }

hierarchy_issues = check_hierarchy(categories)

A cycle report should include the actual path, not only a Boolean. Define whether multiple roots or isolated nodes are valid for your dataset. NetworkX documents the relevant directed acyclic graph algorithms.

5. Check foreign keys and relationship cardinality

When data comes from files or APIs rather than a database with enforced constraints, verify both key existence and uniqueness. Pandas merge can label unmatched rows and validate expected cardinality.

def check_order_customers(orders, customers):
    duplicate_customers = customers.loc[
        customers["customer_id"].duplicated(keep=False)
    ].sort_values("customer_id")

    joined = orders.merge(
        customers[["customer_id"]],
        on="customer_id",
        how="left",
        indicator=True,
        validate="many_to_one",
    )

    orphan_orders = joined.loc[
        joined["_merge"].eq("left_only")
    ].drop(columns="_merge")

    return {
        "duplicate_customer_keys": duplicate_customers,
        "orphan_orders": orphan_orders,
    }

relationship_issues = check_order_customers(orders, customers)

The validate="many_to_one" argument raises an error if customer IDs are not unique on the parent side. That prevents an accidental many-to-many join from multiplying rows. The pandas merge reference lists the supported cardinality checks.

For a composite key, pass all columns to on and test duplicates on the same set:

key = ["account_id", "invoice_number"]

duplicate_invoices = invoices.loc[
    invoices.duplicated(key, keep=False)
]

payments_with_invoice = payments.merge(
    invoices[key],
    on=key,
    how="left",
    indicator=True,
    validate="many_to_one",
)

Put the checks into the pipeline

A validation script is useful only when its result changes the workflow. Run cheap schema and key checks first, quarantine invalid batches, and publish a report that includes the rule, affected row count, sample IDs, source file, and run time. Avoid logging sensitive field values unnecessarily.

• Unit-test each rule with one valid and one invalid fixture.
• Store a versioned schema and business-rule configuration with the code.
• Track warning metrics over time rather than treating every fluctuation as a failure.
• Do not automatically delete or repair records unless the correction rule is deterministic and audited.
• Reconcile row counts before and after joins, filters, and deduplication.

TipsMake’s guide to Python decorators for logging and reusable checks shows one way to wrap repeated pipeline behavior. If you need an interactive environment for exploring failed rows, the comparison of Python IDEs and notebooks includes tools commonly used for data work.

These five patterns are deliberately small. In a larger system, a validation framework can manage schemas, checkpoints, and reports, but the underlying rules still need clear ownership, domain knowledge, and tests.