Test and Run a Reproducible Data Science Pipeline

Scenario

In financial services, production models (such as credit risk assessment or fraud detection) must be strictly reproducible, auditable, and resilient to malformed upstream data. You are tasked with implementing the execution engine and testing framework for a batch scoring and model evaluation pipeline.

Task

1. Implement a class BatchPipelineRunner that coordinates configuration loading, data validation, feature engineering, and model inference.
2. Design unit tests and an end-to-end integration test (using standard Python unittest or pytest patterns) that verify schema compliance, artifact generation, and error handling.

Pipeline Interface

from typing import Dict, Any, Tuple
import pandas as pd

class DataValidationError(Exception):
    """Raised when input data violates expected schema or bounds."""
    pass

class BatchPipelineRunner:
    def __init__(self, config: Dict[str, Any]):
        """
        config contains:
          - 'schema': dict mapping column_name -> expected_type (e.g., {'account_age': int, 'balance': float})
          - 'feature_cols': list of str
          - 'output_dir': str
          - 'model_version': str
        """
        self.config = config

    def validate_input(self, df: pd.DataFrame) -> bool:
        """
        Validates that:
          - All required schema columns exist.
          - Data types are compatible with config['schema'].
          - No negative values in numeric columns specified in config['schema'].
        Raises DataValidationError if invalid.
        """
        pass

    def transform_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Applies deterministic transformations:
          - Fills missing values in feature_cols with median values derived from the batch.
          - Creates a log-transformed feature 'log_balance' if 'balance' exists (using log1p).
        Returns a new DataFrame containing transformed features.
        """
        pass

    def run(self, input_df: pd.DataFrame, model_mock: Any) -> Dict[str, Any]:
        """
        Orchestrates:
          1. Validation
          2. Feature transformation
          3. Model prediction (calls model_mock.predict(features))
        Returns a metadata dict containing:
          - 'status': 'SUCCESS' or 'FAILED'
          - 'record_count': int
          - 'model_version': str
          - 'predictions': list or np.ndarray
        """
        pass

Testing Requirements

Provide test cases testing:

• Schema failure handling (missing column or negative balance) raising DataValidationError.
• Feature transformation correctness (deterministic imputation and log transformation).
• An integration test showing end-to-end execution returning the expected metadata structure.

Constraints

• The input DataFrame will have between 0 and 1,000,000 rows.
• Transformations must not mutate the original input_df in-place.
• Ensure structured logs or return status indicate exact reasons on failure.