Enter Platform
Resources
Back to Notebooks
Open in Colab
MLOps·Feature Engineering Pipeline·Advanced

Feature Pipeline Builder

Build an automated end-to-end feature engineering pipeline that extracts raw market data, computes derived features, validates feature quality and completeness, and stores processed features to the feature store on a production schedule with full dependency graph management and historical backfill support.

feature-engineeringmlops

Automated Feature Pipeline

This notebook outlines the architecture and implementation of an automated feature pipeline. A feature pipeline is a critical component in machine learning systems, responsible for transforming raw data into features suitable for model training and inference. Automation ensures consistency, reliability, and scalability.

Key Concepts:

ConceptDescriptionWhy it's Important
Raw Data IngestionCollecting data from various sources (databases, APIs, files)Foundation for all features; needs to be reliable and timely.
Feature EngineeringTransforming raw data into predictive features (e.g., aggregations, lags, rolling statistics)Directly impacts model performance; tailored to specific problems.
Feature StoreCentralized repository for storing and serving curated featuresEnsures feature consistency across training/inference; reduces redundancy.
Feature ValidationChecking features for quality, consistency, and expected distributionsPrevents data drift, anomalies, and errors from impacting models.
Pipeline OrchestrationManaging the execution, scheduling, and monitoring of the entire feature generation processEnsures timely, efficient, and robust feature delivery.
Monitoring & AlertingObserving pipeline health, feature drift, and resource utilizationProactive identification of issues to maintain pipeline and model performance.

Dependency Installation

We will install the necessary libraries for data manipulation, visualization, and logging.

[ ]
pip install pandas numpy matplotlib seaborn scikit-learn loguru
Requirement already satisfied: pandas in /usr/local/lib/python3.12/dist-packages (2.2.2)
Requirement already satisfied: numpy in /usr/local/lib/python3.12/dist-packages (2.0.2)
Requirement already satisfied: matplotlib in /usr/local/lib/python3.12/dist-packages (3.10.0)
Requirement already satisfied: seaborn in /usr/local/lib/python3.12/dist-packages (0.13.2)
Requirement already satisfied: scikit-learn in /usr/local/lib/python3.12/dist-packages (1.6.1)
Requirement already satisfied: loguru in /usr/local/lib/python3.12/dist-packages (0.7.3)
Requirement already satisfied: python-dateutil>=2.8.2 in /usr/local/lib/python3.12/dist-packages (from pandas) (2.9.0.post0)
Requirement already satisfied: pytz>=2020.1 in /usr/local/lib/python3.12/dist-packages (from pandas) (2025.2)
Requirement already satisfied: tzdata>=2022.7 in /usr/local/lib/python3.12/dist-packages (from pandas) (2026.2)
Requirement already satisfied: contourpy>=1.0.1 in /usr/local/lib/python3.12/dist-packages (from matplotlib) (1.3.3)
Requirement already satisfied: cycler>=0.10 in /usr/local/lib/python3.12/dist-packages (from matplotlib) (0.12.1)
Requirement already satisfied: fonttools>=4.22.0 in /usr/local/lib/python3.12/dist-packages (from matplotlib) (4.63.0)
Requirement already satisfied: kiwisolver>=1.3.1 in /usr/local/lib/python3.12/dist-packages (from matplotlib) (1.5.0)
Requirement already satisfied: packaging>=20.0 in /usr/local/lib/python3.12/dist-packages (from matplotlib) (26.2)
Requirement already satisfied: pillow>=8 in /usr/local/lib/python3.12/dist-packages (from matplotlib) (11.3.0)
Requirement already satisfied: pyparsing>=2.3.1 in /usr/local/lib/python3.12/dist-packages (from matplotlib) (3.3.2)
Requirement already satisfied: scipy>=1.6.0 in /usr/local/lib/python3.12/dist-packages (from scikit-learn) (1.16.3)
Requirement already satisfied: joblib>=1.2.0 in /usr/local/lib/python3.12/dist-packages (from scikit-learn) (1.5.3)
Requirement already satisfied: threadpoolctl>=3.1.0 in /usr/local/lib/python3.12/dist-packages (from scikit-learn) (3.6.0)
Requirement already satisfied: six>=1.5 in /usr/local/lib/python3.12/dist-packages (from python-dateutil>=2.8.2->pandas) (1.17.0)

Library Imports

This section imports all required libraries. Standard libraries are imported first, followed by third-party libraries.

[ ]
import datetime
import logging
import random
import time
from collections import deque
from typing import Dict, Any, List, Union

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
from loguru import logger
from sklearn.preprocessing import StandardScaler

# Configure logging
logger.remove()
logger.add(lambda msg: print(msg, end=''), colorize=True, format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>")

2

Core Functions

This section defines the core functions for our automated feature pipeline. Each function is presented in its own code block, preceded by a markdown header explaining its purpose, algorithm, and parameters. Docstrings, type hints, and logger statements are included for clarity and maintainability.

Function Name: create_pipeline_state

This function initializes the global state dictionary for the feature pipeline. It sets up various components like the feature store (simulated as an empty DataFrame), logging, and configuration parameters.

Parameters: config (dict): A dictionary containing configuration parameters for the pipeline, such as data generation settings.

Returns: (dict): An initialized state dictionary for the feature pipeline.

[ ]
def create_pipeline_state(config: Dict[str, Any]) -> Dict[str, Any]:
    """
    Initializes the global state dictionary for the feature pipeline.

    Parameters
    ----------
    config : Dict[str, Any]
        Configuration parameters for the pipeline, e.g., data generation settings.

    Returns
    -------
    Dict[str, Any]
        An initialized state dictionary for the feature pipeline.

    Examples
    --------
    >>> config = {'start_date': '2023-01-01', 'num_days': 100, 'num_entities': 5}
    >>> state = create_pipeline_state(config)
    >>> 'feature_store' in state
    True
    """
    logger.info("Initializing pipeline state...")
    state = {
        'config': config,
        'raw_data': pd.DataFrame(),
        'processed_data': pd.DataFrame(),
        'feature_store': pd.DataFrame(columns=['entity_id', 'timestamp', 'value', 'lag_1', 'rolling_mean_3', 'rolling_std_3']),
        'metrics': {},
        'last_run_timestamp': None
    }
    logger.info("Pipeline state initialized successfully.")
    return state

Function Name: simulate_data_ingestion

This function simulates the ingestion of raw time-series data. It generates synthetic data for multiple entities over a specified period, including a trend and some seasonality, making it suitable for feature engineering demonstrations. It uses pd.date_range to create a time index and np.random.randn for noise.

Parameters: state (dict): The current pipeline state dictionary.

Returns: (dict): The updated pipeline state dictionary with the raw_data field populated.

[ ]
def simulate_data_ingestion(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Simulates the ingestion of raw time-series data.

    Generates synthetic data for multiple entities over a specified period.
    Includes a trend and some seasonality.

    Parameters
    ----------
    state : Dict[str, Any]
        The current pipeline state dictionary.

    Returns
    -------
    Dict[str, Any]
        The updated pipeline state dictionary with the `raw_data` field populated.

    Examples
    --------
    >>> config = {'start_date': '2023-01-01', 'num_days': 10, 'num_entities': 2}
    >>> state = create_pipeline_state(config)
    >>> state = simulate_data_ingestion(state)
    >>> isinstance(state['raw_data'], pd.DataFrame)
    True
    """
    logger.info("Simulating data ingestion...")
    config = state['config']
    start_date = pd.to_datetime(config['start_date'])
    num_days = config['num_days']
    num_entities = config['num_entities']

    dates = pd.date_range(start=start_date, periods=num_days, freq='D')
    all_data = []

    for i in range(num_entities):
        entity_id = f'entity_{i+1}'
        # Generate a base value with a trend and some seasonality
        base_values = 50 + np.arange(num_days) * 0.1 + 10 * np.sin(np.arange(num_days) * 2 * np.pi / 30)
        # Add random noise
        values = base_values + np.random.randn(num_days) * 5

        entity_df = pd.DataFrame({
            'entity_id': entity_id,
            'timestamp': dates,
            'value': values
        })
        all_data.append(entity_df)

    state['raw_data'] = pd.concat(all_data).reset_index(drop=True)
    logger.info(f"Ingested {len(state['raw_data'])} rows of raw data for {num_entities} entities.")
    return state

Function Name: generate_lag_features

This function creates lagged features for a given time-series dataset. Lagged features are previous values of a time series, which are crucial for forecasting and understanding temporal dependencies. The function groups data by entity_id and applies shift to create the lagged columns.

Parameters: state (dict): The current pipeline state dictionary, expected to contain raw_data. lag_periods (List[int]): A list of integers representing the number of periods to lag the 'value' column.

Returns: (dict): The updated pipeline state dictionary with lagged features added to processed_data.

[ ]
def generate_lag_features(state: Dict[str, Any], lag_periods: List[int]) -> Dict[str, Any]:
    """
    Generates lagged features for the 'value' column in the raw data.

    Parameters
    ----------
    state : Dict[str, Any]
        The current pipeline state dictionary, expected to contain 'raw_data'.
    lag_periods : List[int]
        A list of integers representing the number of periods to lag the 'value' column.

    Returns
    -------
    Dict[str, Any]
        The updated pipeline state dictionary with lagged features added to 'processed_data'.

    Examples
    --------
    >>> config = {'start_date': '2023-01-01', 'num_days': 10, 'num_entities': 1}
    >>> state = create_pipeline_state(config)
    >>> state = simulate_data_ingestion(state)
    >>> state = generate_lag_features(state, lag_periods=[1, 2])
    >>> 'value_lag_1' in state['processed_data'].columns
    True
    """
    logger.info(f"Generating lag features for periods: {lag_periods}...")
    data = state['raw_data'].copy()
    data = data.sort_values(by=['entity_id', 'timestamp'])

    for lag in lag_periods:
        data[f'value_lag_{lag}'] = data.groupby('entity_id')['value'].shift(lag)

    state['processed_data'] = data
    logger.info(f"Generated {len(lag_periods)} lag features. First 5 rows:\n{state['processed_data'].head()}")
    return state

Function Name: generate_rolling_features

This function computes rolling window statistics (e.g., mean, standard deviation) for the 'value' column. Rolling features capture recent trends and variability, which are important for capturing short-term dynamics in time series. It uses groupby and rolling methods from pandas.

Parameters: state (dict): The current pipeline state dictionary, expected to contain processed_data. window_sizes (List[int]): A list of integers representing the window sizes for rolling calculations.

Returns: (dict): The updated pipeline state dictionary with rolling features added to processed_data.

[ ]
def generate_rolling_features(state: Dict[str, Any], window_sizes: List[int]) -> Dict[str, Any]:
    """
    Generates rolling window features (mean and std dev) for the 'value' column.

    Parameters
    ----------
    state : Dict[str, Any]
        The current pipeline state dictionary, expected to contain 'processed_data'.
    window_sizes : List[int]
        A list of integers representing the window sizes for rolling calculations.

    Returns
    -------
    Dict[str, Any]
        The updated pipeline state dictionary with rolling features added to 'processed_data'.

    Examples
    --------
    >>> config = {'start_date': '2023-01-01', 'num_days': 10, 'num_entities': 1}
    >>> state = create_pipeline_state(config)
    >>> state = simulate_data_ingestion(state)
    >>> state = generate_lag_features(state, lag_periods=[1])
    >>> state = generate_rolling_features(state, window_sizes=[3])
    >>> 'value_rolling_mean_3' in state['processed_data'].columns
    True
    """
    logger.info(f"Generating rolling features for window sizes: {window_sizes}...")
    data = state['processed_data'].copy()
    data = data.sort_values(by=['entity_id', 'timestamp'])

    for window in window_sizes:
        data[f'value_rolling_mean_{window}'] = data.groupby('entity_id')['value'].rolling(window=window).mean().reset_index(level=0, drop=True)
        data[f'value_rolling_std_{window}'] = data.groupby('entity_id')['value'].rolling(window=window).std().reset_index(level=0, drop=True)

    state['processed_data'] = data
    logger.info(f"Generated {len(window_sizes) * 2} rolling features. First 5 rows:\n{state['processed_data'].head()}")
    return state

Function Name: handle_missing_values

This function addresses missing values that often arise after generating time-series features (e.g., initial lags or rolling windows). It uses a configurable strategy, such as filling with zero, mean, or median, or dropping rows. For simplicity, we'll implement dropping NaN values here, which is a common approach for initial feature engineering.

Parameters: state (dict): The current pipeline state dictionary, expected to contain processed_data. strategy (str): The strategy for handling missing values ('drop', 'fill_zero', 'fill_mean').

Returns: (dict): The updated pipeline state dictionary with missing values handled in processed_data.

[ ]
def handle_missing_values(state: Dict[str, Any], strategy: str = 'drop') -> Dict[str, Any]:
    """
    Handles missing values (NaNs) in the processed data, typically introduced by feature engineering.

    Parameters
    ----------
    state : Dict[str, Any]
        The current pipeline state dictionary, expected to contain 'processed_data'.
    strategy : str, optional
        The strategy for handling missing values ('drop', 'fill_zero', 'fill_mean'),
        defaults to 'drop'.

    Returns
    -------
    Dict[str, Any]
        The updated pipeline state dictionary with missing values handled in 'processed_data'.

    Examples
    --------
    >>> config = {'start_date': '2023-01-01', 'num_days': 10, 'num_entities': 1}
    >>> state = create_pipeline_state(config)
    >>> state = simulate_data_ingestion(state)
    >>> state = generate_lag_features(state, lag_periods=[1])
    >>> state['processed_data'].iloc[0, state['processed_data'].columns.get_loc('value_lag_1')] = np.nan # Simulate NaN
    >>> initial_rows = len(state['processed_data'])
    >>> state = handle_missing_values(state, strategy='drop')
    >>> len(state['processed_data']) < initial_rows
    True
    """
    logger.info(f"Handling missing values with strategy: '{strategy}'...")
    data = state['processed_data'].copy()
    initial_rows = len(data)

    if strategy == 'drop':
        data.dropna(inplace=True)
        logger.info(f"Dropped {initial_rows - len(data)} rows with missing values.")
    elif strategy == 'fill_zero':
        data.fillna(0, inplace=True)
        logger.info("Filled missing values with 0.")
    elif strategy == 'fill_mean':
        for col in data.select_dtypes(include=np.number).columns:
            data[col].fillna(data[col].mean(), inplace=True)
        logger.info("Filled missing values with column means.")
    else:
        logger.warning(f"Unknown missing value strategy: '{strategy}'. No action taken.")

    state['processed_data'] = data
    logger.info(f"Missing values handled. Current data shape: {state['processed_data'].shape}")
    return state

Function Name: validate_features

This function performs basic validation checks on the engineered features. It checks for NaN values, infinite values, and performs an outlier detection using IQR. This step is crucial for maintaining data quality and preventing corrupted features from entering the model training process. It logs any issues found.

Parameters: state (dict): The current pipeline state dictionary, expected to contain processed_data.

Returns: (dict): The updated pipeline state dictionary with validation results added to metrics.

[ ]
def validate_features(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Performs basic validation checks on the engineered features.

    Checks for NaNs, infinite values, and performs outlier detection.

    Parameters
    ----------
    state : Dict[str, Any]
        The current pipeline state dictionary, expected to contain 'processed_data'.

    Returns
    -------
    Dict[str, Any]
        The updated pipeline state dictionary with validation results added to 'metrics'.

    Examples
    --------
    >>> config = {'start_date': '2023-01-01', 'num_days': 10, 'num_entities': 1}
    >>> state = create_pipeline_state(config)
    >>> state = simulate_data_ingestion(state)
    >>> state = generate_lag_features(state, lag_periods=[1])
    >>> state = handle_missing_values(state, strategy='drop')
    >>> state = validate_features(state)
    >>> 'validation_summary' in state['metrics']
    True
    """
    logger.info("Validating engineered features...")
    data = state['processed_data']
    validation_issues = []

    feature_cols = [col for col in data.columns if col not in ['entity_id', 'timestamp']]

    # Check for NaNs
    nan_counts = data[feature_cols].isnull().sum()
    if nan_counts.sum() > 0:
        logger.warning(f"NaN values found in features:\n{nan_counts[nan_counts > 0]}")
        validation_issues.append({'type': 'NaN', 'details': nan_counts[nan_counts > 0].to_dict()})

    # Check for Infinite values
    inf_counts = data[feature_cols].apply(lambda x: np.isinf(x).sum())
    if inf_counts.sum() > 0:
        logger.warning(f"Infinite values found in features:\n{inf_counts[inf_counts > 0]}")
        validation_issues.append({'type': 'Infinite', 'details': inf_counts[inf_counts > 0].to_dict()})

    # Check for outliers using IQR for numerical features
    outlier_summary = {}
    for col in feature_cols:
        if pd.api.types.is_numeric_dtype(data[col]):
            Q1 = data[col].quantile(0.25)
            Q3 = data[col].quantile(0.75)
            IQR = Q3 - Q1
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR
            num_outliers = ((data[col] < lower_bound) | (data[col] > upper_bound)).sum()
            if num_outliers > 0:
                outlier_summary[col] = num_outliers
    if outlier_summary:
        logger.warning(f"Outliers detected in features:\n{outlier_summary}")
        validation_issues.append({'type': 'Outliers', 'details': outlier_summary})

    if not validation_issues:
        logger.info("Feature validation passed: No major issues detected.")
    else:
        logger.error("Feature validation found issues. See details in logs.")

    state['metrics']['validation_summary'] = validation_issues
    return state

Function Name: store_features

This function simulates storing the engineered features into a feature store. In a real-world scenario, this would involve writing to a dedicated database (e.g., Feast, Google Cloud Feature Store). Here, we append the new features to a pandas DataFrame representing our in-memory feature store. It handles potential duplicates by concatenating and dropping if a timestamp is already present for an entity_id.

Parameters: state (dict): The current pipeline state dictionary, containing processed_data and feature_store.

Returns: (dict): The updated pipeline state dictionary with features stored in feature_store.

[ ]
def store_features(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Simulates storing the engineered features into a feature store.

    Parameters
    ----------
    state : Dict[str, Any]
        The current pipeline state dictionary, containing 'processed_data' and 'feature_store'.

    Returns
    -------
    Dict[str, Any]
        The updated pipeline state dictionary with features stored in 'feature_store'.

    Examples
    --------
    >>> config = {'start_date': '2023-01-01', 'num_days': 5, 'num_entities': 1}
    >>> state = create_pipeline_state(config)
    >>> state = simulate_data_ingestion(state)
    >>> state = generate_lag_features(state, lag_periods=[1])
    >>> state = handle_missing_values(state, strategy='drop')
    >>> state = store_features(state)
    >>> len(state['feature_store']) > 0
    True
    """
    logger.info("Storing features into the feature store...")
    features_to_store = state['processed_data'].copy()

    # Ensure only feature columns are stored, along with identifiers
    feature_cols = [col for col in features_to_store.columns if col not in ['raw_value']]
    features_to_store = features_to_store[feature_cols]

    # Simulate appending to feature store, handling potential duplicates
    current_store = state['feature_store']
    combined_store = pd.concat([current_store, features_to_store])

    # Drop duplicates based on entity_id and timestamp
    combined_store.drop_duplicates(subset=['entity_id', 'timestamp'], keep='last', inplace=True)
    state['feature_store'] = combined_store.sort_values(by=['entity_id', 'timestamp']).reset_index(drop=True)

    logger.info(f"Features stored. Feature store now contains {len(state['feature_store'])} records.")
    return state

Function Name: run_pipeline_step_with_jitter

This helper function executes a pipeline step (another function) with a simulated delay and exponential backoff for retries. This pattern is useful for robust interactions with external services (APIs, databases) where transient errors or rate limits might occur. It introduces random jitter to backoff times to prevent thundering herd problems.

Parameters: func (Callable): The function to execute as a pipeline step. args (tuple): Positional arguments to pass to the function. kwargs (dict): Keyword arguments to pass to the function. max_retries (int): Maximum number of retries if the function fails. initial_delay (float): Initial delay in seconds before retrying.

Returns: (Any): The result of the executed function.

Raises: Exception: If the function fails after all retries.

[ ]
def run_pipeline_step_with_jitter(func: Any, *args: Any, max_retries: int = 3, initial_delay: float = 1.0, **kwargs: Any) -> Any:
    """
    Executes a pipeline step with a simulated delay and exponential backoff for retries.

    Parameters
    ----------
    func : Callable
        The function to execute as a pipeline step.
    *args : Any
        Positional arguments to pass to the function.
    max_retries : int, optional
        Maximum number of retries if the function fails, defaults to 3.
    initial_delay : float, optional
        Initial delay in seconds before retrying, defaults to 1.0.
    **kwargs : Any
        Keyword arguments to pass to the function.

    Returns
    -------
    Any
        The result of the executed function.

    Raises
    ------
    Exception
        If the function fails after all retries.

    Examples
    --------
    >>> def mock_failure_func(attempt):
    ...     if attempt < 2:
    ...         raise ValueError("Simulated transient error")
    ...     return "Success"
    >>> # Example of a function that eventually succeeds:
    >>> # state = {'attempt': 0}
    >>> # def resilient_func(s): s['attempt'] += 1; return mock_failure_func(s['attempt'])
    >>> # run_pipeline_step_with_jitter(resilient_func, state)
    'Success'
    """
    delay = initial_delay
    for i in range(max_retries + 1):
        try:
            logger.debug(f"Attempt {i+1} for function '{func.__name__}'.")
            result = func(*args, **kwargs)
            logger.debug(f"Function '{func.__name__}' succeeded on attempt {i+1}.")
            return result
        except Exception as e:
            logger.error(f"Function '{func.__name__}' failed on attempt {i+1} with error: {e}")
            if i < max_retries:
                jitter = random.uniform(0.5, 1.5) # Add random jitter
                sleep_time = delay * jitter
                logger.warning(f"Retrying '{func.__name__}' in {sleep_time:.2f} seconds...")
                time.sleep(sleep_time)
                delay *= 2  # Exponential backoff
            else:
                logger.critical(f"Function '{func.__name__}' failed after {max_retries} retries.")
                raise

Function Name: update_metrics

This function updates the pipeline's monitoring metrics. It can track various aspects such as data volume, feature completeness, processing time, or custom validation scores. Here, it will update a simple dictionary of metrics. In a real system, this would push metrics to a monitoring system like Prometheus or DataDog.

Parameters: state (dict): The current pipeline state dictionary. metric_name (str): The name of the metric to update. metric_value (Union[int, float, Dict[str, Any]]): The value of the metric. Can be a simple number or a dictionary for complex metrics.

Returns: (dict): The updated pipeline state dictionary with new metrics.

[ ]
def update_metrics(state: Dict[str, Any], metric_name: str, metric_value: Union[int, float, Dict[str, Any]]) -> Dict[str, Any]:
    """
    Updates the pipeline's monitoring metrics.

    Parameters
    ----------
    state : Dict[str, Any]
        The current pipeline state dictionary.
    metric_name : str
        The name of the metric to update.
    metric_value : Union[int, float, Dict[str, Any]]
        The value of the metric. Can be a simple number or a dictionary for complex metrics.

    Returns
    -------
    Dict[str, Any]
        The updated pipeline state dictionary with new metrics.

    Examples
    --------
    >>> config = {'start_date': '2023-01-01', 'num_days': 1, 'num_entities': 1}
    >>> state = create_pipeline_state(config)
    >>> state = update_metrics(state, 'ingested_rows', 100)
    >>> state['metrics']['ingested_rows']
    100
    """
    logger.info(f"Updating metric '{metric_name}' with value: {metric_value}")
    state['metrics'][metric_name] = metric_value
    return state

Demonstration/Visualization

This section demonstrates the end-to-end execution of the automated feature pipeline using the defined core functions. It will show the raw data, the engineered features, and provide visualizations to illustrate the transformations and the state of the feature store. We will simulate a continuous pipeline run over several iterations.

[ ]
# 1. Initialize Pipeline Configuration
config = {
    'start_date': '2023-01-01',
    'num_days': 50, # Initial days for raw data generation
    'num_entities': 3,
    'lag_periods': [1, 2, 3],
    'rolling_window_sizes': [3, 7],
    'missing_value_strategy': 'drop'
}
pipeline_state = run_pipeline_step_with_jitter(create_pipeline_state, config)

# 2. Simulate Initial Data Ingestion and Feature Generation
logger.info("--- Initial Pipeline Run ---")
pipeline_state = run_pipeline_step_with_jitter(simulate_data_ingestion, pipeline_state)
pipeline_state = run_pipeline_step_with_jitter(generate_lag_features, pipeline_state, lag_periods=config['lag_periods'])
pipeline_state = run_pipeline_step_with_jitter(generate_rolling_features, pipeline_state, window_sizes=config['rolling_window_sizes'])
pipeline_state = run_pipeline_step_with_jitter(handle_missing_values, pipeline_state, strategy=config['missing_value_strategy'])
pipeline_state = run_pipeline_step_with_jitter(validate_features, pipeline_state)
pipeline_state = run_pipeline_step_with_jitter(store_features, pipeline_state)
pipeline_state = run_pipeline_step_with_jitter(update_metrics, pipeline_state, 'initial_feature_count', len(pipeline_state['feature_store']))

logger.info("Initial feature store snapshot:")
display(pipeline_state['feature_store'].head())

# 3. Visualize Raw Data vs. Engineered Features for one entity
entity_to_plot = pipeline_state['config']['num_entities'] // 2 + 1 # Pick a middle entity
entity_id_str = f'entity_{entity_to_plot}'
plot_df = pipeline_state['feature_store'][pipeline_state['feature_store']['entity_id'] == entity_id_str].copy()

plt.figure(figsize=(15, 7))
plt.plot(plot_df['timestamp'], plot_df['value'], label='Original Value', color='blue', alpha=0.7)
plt.plot(plot_df['timestamp'], plot_df['value_lag_1'], label='Lag 1 Value', color='green', linestyle='--', alpha=0.7)
plt.plot(plot_df['timestamp'], plot_df['value_rolling_mean_3'], label='Rolling Mean (3-day)', color='red', linestyle=':', alpha=0.7)
plt.title(f'Raw Value vs. Engineered Features for {entity_id_str}')
plt.xlabel('Timestamp')
plt.ylabel('Value')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()

# 4. Simulate Subsequent Pipeline Runs (e.g., daily updates)
logger.info("--- Simulating Daily Pipeline Updates ---")
num_updates = 5 # Simulate 5 more days of updates
for i in range(num_updates):
    logger.info(f"--- Daily Update {i+1}/{num_updates} ---")
    # Increment start date for new data generation to simulate new data arrival
    pipeline_state['config']['start_date'] = (pd.to_datetime(pipeline_state['config']['start_date']) + datetime.timedelta(days=config['num_days']) + datetime.timedelta(days=i)).strftime('%Y-%m-%d')
    pipeline_state['config']['num_days'] = 1 # Process one new day of data

    # Simulate ingestion of new data
    pipeline_state = run_pipeline_step_with_jitter(simulate_data_ingestion, pipeline_state)

    # Re-run feature engineering on the *entire* raw data (or a window) to ensure correct lags/rolls
    # For simplicity, we re-run on the cumulative raw data. In production, this would be incremental.
    current_raw_data = pd.concat([pipeline_state['raw_data'], pipeline_state['feature_store'][['entity_id', 'timestamp', 'value']]])
    current_raw_data.drop_duplicates(subset=['entity_id', 'timestamp'], keep='last', inplace=True)
    current_raw_data.sort_values(by=['entity_id', 'timestamp'], inplace=True)
    pipeline_state['raw_data'] = current_raw_data

    pipeline_state = run_pipeline_step_with_jitter(generate_lag_features, pipeline_state, lag_periods=config['lag_periods'])
    pipeline_state = run_pipeline_step_with_jitter(generate_rolling_features, pipeline_state, window_sizes=config['rolling_window_sizes'])
    pipeline_state = run_pipeline_step_with_jitter(handle_missing_values, pipeline_state, strategy=config['missing_value_strategy'])
    pipeline_state = run_pipeline_step_with_jitter(validate_features, pipeline_state)
    pipeline_state = run_pipeline_step_with_jitter(store_features, pipeline_state)
    pipeline_state = run_pipeline_step_with_jitter(update_metrics, pipeline_state, f'feature_count_update_{i+1}', len(pipeline_state['feature_store']))

logger.info("Pipeline updates complete.")
logger.info(f"Final feature store size: {len(pipeline_state['feature_store'])} records.")
logger.info("Final feature store head:")
display(pipeline_state['feature_store'].tail())

# 5. Visualize Feature Distribution (example: rolling mean)
plt.figure(figsize=(10, 6))
sns.histplot(pipeline_state['feature_store']['value_rolling_mean_3'].dropna(), kde=True)
plt.title('Distribution of Rolling Mean (3-day) Feature')
plt.xlabel('Rolling Mean Value')
plt.ylabel('Frequency')
plt.grid(True)
plt.tight_layout()
plt.show()

# 6. Display pipeline metrics
logger.info("Pipeline Metrics Summary:")
metrics_df = pd.DataFrame([pipeline_state['metrics']]).T
metrics_df.columns = ['Value']
display(metrics_df)

2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'create_pipeline_state'.
2026-06-10 07:49:34 | INFO     | __main__:create_pipeline_state:22 - Initializing pipeline state...
2026-06-10 07:49:34 | INFO     | __main__:create_pipeline_state:31 - Pipeline state initialized successfully.
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'create_pipeline_state' succeeded on attempt 1.
2026-06-10 07:49:34 | INFO     | __main__:<cell line: 0>:13 - --- Initial Pipeline Run ---
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'simulate_data_ingestion'.
2026-06-10 07:49:34 | INFO     | __main__:simulate_data_ingestion:26 - Simulating data ingestion...
2026-06-10 07:49:34 | INFO     | __main__:simulate_data_ingestion:50 - Ingested 150 rows of raw data for 3 entities.
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'simulate_data_ingestion' succeeded on attempt 1.
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_lag_features'.
2026-06-10 07:49:34 | INFO     | __main__:generate_lag_features:26 - Generating lag features for periods: [1, 2, 3]...
2026-06-10 07:49:34 | INFO     | __main__:generate_lag_features:34 - Generated 3 lag features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3
0  entity_1 2023-01-01  46.577949          NaN          NaN          NaN
1  entity_1 2023-01-02  46.489019    46.577949          NaN          NaN
2  entity_1 2023-01-03  62.355653    46.489019    46.577949          NaN
3  entity_1 2023-01-04  60.938055    62.355653    46.489019    46.577949
4  entity_1 2023-01-05  55.321050    60.938055    62.355653    46.489019
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_lag_features' succeeded on attempt 1.
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_rolling_features'.
2026-06-10 07:49:34 | INFO     | __main__:generate_rolling_features:27 - Generating rolling features for window sizes: [3, 7]...
2026-06-10 07:49:34 | INFO     | __main__:generate_rolling_features:36 - Generated 4 rolling features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3  \
0  entity_1 2023-01-01  46.577949          NaN          NaN          NaN   
1  entity_1 2023-01-02  46.489019    46.577949          NaN          NaN   
2  entity_1 2023-01-03  62.355653    46.489019    46.577949          NaN   
3  entity_1 2023-01-04  60.938055    62.355653    46.489019    46.577949   
4  entity_1 2023-01-05  55.321050    60.938055    62.355653    46.489019   

   value_rolling_mean_3  value_rolling_std_3  value_rolling_mean_7  \
0                   NaN                  NaN                   NaN   
1                   NaN                  NaN                   NaN   
2             51.807540             9.135042                   NaN   
3             56.594242             8.780037                   NaN   
4             59.538253             3.720349                   NaN   

   value_rolling_std_7  
0                  NaN  
1                  NaN  
2                  NaN  
3                  NaN  
4                  NaN  
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_rolling_features' succeeded on attempt 1.
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'handle_missing_values'.
2026-06-10 07:49:34 | INFO     | __main__:handle_missing_values:30 - Handling missing values with strategy: 'drop'...
2026-06-10 07:49:34 | INFO     | __main__:handle_missing_values:36 - Dropped 18 rows with missing values.
2026-06-10 07:49:34 | INFO     | __main__:handle_missing_values:48 - Missing values handled. Current data shape: (132, 10)
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'handle_missing_values' succeeded on attempt 1.
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'validate_features'.
2026-06-10 07:49:34 | INFO     | __main__:validate_features:28 - Validating engineered features...
2026-06-10 07:49:34 | WARNING  | __main__:validate_features:59 - Outliers detected in features:
{'value_rolling_std_7': np.int64(1)}
2026-06-10 07:49:34 | ERROR    | __main__:validate_features:65 - Feature validation found issues. See details in logs.
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'validate_features' succeeded on attempt 1.
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'store_features'.
2026-06-10 07:49:34 | INFO     | __main__:store_features:26 - Storing features into the feature store...
2026-06-10 07:49:34 | INFO     | __main__:store_features:41 - Features stored. Feature store now contains 132 records.
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'store_features' succeeded on attempt 1.
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'update_metrics'.
2026-06-10 07:49:34 | INFO     | __main__:update_metrics:27 - Updating metric 'initial_feature_count' with value: 132
2026-06-10 07:49:34 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'update_metrics' succeeded on attempt 1.
2026-06-10 07:49:34 | INFO     | __main__:<cell line: 0>:22 - Initial feature store snapshot:

/tmp/ipykernel_4822/3386017649.py:35: FutureWarning: The behavior of DataFrame concatenation with empty or all-NA entries is deprecated. In a future version, this will no longer exclude empty or all-NA columns when determining the result dtypes. To retain the old behavior, exclude the relevant entries before the concat operation.
  combined_store = pd.concat([current_store, features_to_store])
entity_id timestamp value lag_1 rolling_mean_3 rolling_std_3 value_lag_1 value_lag_2 value_lag_3 value_rolling_mean_3 value_rolling_std_3 value_rolling_mean_7 value_rolling_std_7
0 entity_1 2023-01-07 55.194904 NaN NaN NaN 63.440282 55.321050 60.938055 57.985412 4.724477 55.759559 7.071949
1 entity_1 2023-01-08 60.884464 NaN NaN NaN 55.194904 63.440282 55.321050 59.839883 4.220773 57.803347 5.955362
2 entity_1 2023-01-09 56.127975 NaN NaN NaN 60.884464 55.194904 63.440282 57.402448 3.051390 59.180340 3.519424
3 entity_1 2023-01-10 63.471766 NaN NaN NaN 56.127975 60.884464 55.194904 60.161402 3.724907 59.339785 3.707511
4 entity_1 2023-01-11 56.198290 NaN NaN NaN 63.471766 56.127975 60.884464 58.599344 4.219788 58.662676 3.798662
cell output
2026-06-10 07:49:35 | INFO     | __main__:<cell line: 0>:43 - --- Simulating Daily Pipeline Updates ---
2026-06-10 07:49:35 | INFO     | __main__:<cell line: 0>:46 - --- Daily Update 1/5 ---
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'simulate_data_ingestion'.
2026-06-10 07:49:35 | INFO     | __main__:simulate_data_ingestion:26 - Simulating data ingestion...
2026-06-10 07:49:35 | INFO     | __main__:simulate_data_ingestion:50 - Ingested 3 rows of raw data for 3 entities.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'simulate_data_ingestion' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_lag_features'.
2026-06-10 07:49:35 | INFO     | __main__:generate_lag_features:26 - Generating lag features for periods: [1, 2, 3]...
2026-06-10 07:49:35 | INFO     | __main__:generate_lag_features:34 - Generated 3 lag features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3
0  entity_1 2023-01-07  55.194904          NaN          NaN          NaN
1  entity_1 2023-01-08  60.884464    55.194904          NaN          NaN
2  entity_1 2023-01-09  56.127975    60.884464    55.194904          NaN
3  entity_1 2023-01-10  63.471766    56.127975    60.884464    55.194904
4  entity_1 2023-01-11  56.198290    63.471766    56.127975    60.884464
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_lag_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_rolling_features'.
2026-06-10 07:49:35 | INFO     | __main__:generate_rolling_features:27 - Generating rolling features for window sizes: [3, 7]...
2026-06-10 07:49:35 | INFO     | __main__:generate_rolling_features:36 - Generated 4 rolling features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3  \
0  entity_1 2023-01-07  55.194904          NaN          NaN          NaN   
1  entity_1 2023-01-08  60.884464    55.194904          NaN          NaN   
2  entity_1 2023-01-09  56.127975    60.884464    55.194904          NaN   
3  entity_1 2023-01-10  63.471766    56.127975    60.884464    55.194904   
4  entity_1 2023-01-11  56.198290    63.471766    56.127975    60.884464   

   value_rolling_mean_3  value_rolling_std_3  value_rolling_mean_7  \
0                   NaN                  NaN                   NaN   
1                   NaN                  NaN                   NaN   
2             57.402448             3.051390                   NaN   
3             60.161402             3.724907                   NaN   
4             58.599344             4.219788                   NaN   

   value_rolling_std_7  
0                  NaN  
1                  NaN  
2                  NaN  
3                  NaN  
4                  NaN  
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_rolling_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'handle_missing_values'.
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:30 - Handling missing values with strategy: 'drop'...
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:36 - Dropped 18 rows with missing values.
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:48 - Missing values handled. Current data shape: (117, 10)
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'handle_missing_values' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'validate_features'.
2026-06-10 07:49:35 | INFO     | __main__:validate_features:28 - Validating engineered features...
2026-06-10 07:49:35 | WARNING  | __main__:validate_features:59 - Outliers detected in features:
{'value_rolling_std_7': np.int64(1)}
2026-06-10 07:49:35 | ERROR    | __main__:validate_features:65 - Feature validation found issues. See details in logs.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'validate_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'store_features'.
2026-06-10 07:49:35 | INFO     | __main__:store_features:26 - Storing features into the feature store...
2026-06-10 07:49:35 | INFO     | __main__:store_features:41 - Features stored. Feature store now contains 135 records.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'store_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'update_metrics'.
2026-06-10 07:49:35 | INFO     | __main__:update_metrics:27 - Updating metric 'feature_count_update_1' with value: 135
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'update_metrics' succeeded on attempt 1.
2026-06-10 07:49:35 | INFO     | __main__:<cell line: 0>:46 - --- Daily Update 2/5 ---
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'simulate_data_ingestion'.
2026-06-10 07:49:35 | INFO     | __main__:simulate_data_ingestion:26 - Simulating data ingestion...
2026-06-10 07:49:35 | INFO     | __main__:simulate_data_ingestion:50 - Ingested 3 rows of raw data for 3 entities.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'simulate_data_ingestion' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_lag_features'.
2026-06-10 07:49:35 | INFO     | __main__:generate_lag_features:26 - Generating lag features for periods: [1, 2, 3]...
2026-06-10 07:49:35 | INFO     | __main__:generate_lag_features:34 - Generated 3 lag features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3
0  entity_1 2023-01-07  55.194904          NaN          NaN          NaN
1  entity_1 2023-01-08  60.884464    55.194904          NaN          NaN
2  entity_1 2023-01-09  56.127975    60.884464    55.194904          NaN
3  entity_1 2023-01-10  63.471766    56.127975    60.884464    55.194904
4  entity_1 2023-01-11  56.198290    63.471766    56.127975    60.884464
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_lag_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_rolling_features'.
2026-06-10 07:49:35 | INFO     | __main__:generate_rolling_features:27 - Generating rolling features for window sizes: [3, 7]...
2026-06-10 07:49:35 | INFO     | __main__:generate_rolling_features:36 - Generated 4 rolling features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3  \
0  entity_1 2023-01-07  55.194904          NaN          NaN          NaN   
1  entity_1 2023-01-08  60.884464    55.194904          NaN          NaN   
2  entity_1 2023-01-09  56.127975    60.884464    55.194904          NaN   
3  entity_1 2023-01-10  63.471766    56.127975    60.884464    55.194904   
4  entity_1 2023-01-11  56.198290    63.471766    56.127975    60.884464   

   value_rolling_mean_3  value_rolling_std_3  value_rolling_mean_7  \
0                   NaN                  NaN                   NaN   
1                   NaN                  NaN                   NaN   
2             57.402448             3.051390                   NaN   
3             60.161402             3.724907                   NaN   
4             58.599344             4.219788                   NaN   

   value_rolling_std_7  
0                  NaN  
1                  NaN  
2                  NaN  
3                  NaN  
4                  NaN  
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_rolling_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'handle_missing_values'.
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:30 - Handling missing values with strategy: 'drop'...
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:36 - Dropped 18 rows with missing values.
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:48 - Missing values handled. Current data shape: (120, 10)
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'handle_missing_values' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'validate_features'.
2026-06-10 07:49:35 | INFO     | __main__:validate_features:28 - Validating engineered features...
2026-06-10 07:49:35 | WARNING  | __main__:validate_features:59 - Outliers detected in features:
{'value_rolling_std_7': np.int64(1)}
2026-06-10 07:49:35 | ERROR    | __main__:validate_features:65 - Feature validation found issues. See details in logs.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'validate_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'store_features'.
2026-06-10 07:49:35 | INFO     | __main__:store_features:26 - Storing features into the feature store...
2026-06-10 07:49:35 | INFO     | __main__:store_features:41 - Features stored. Feature store now contains 138 records.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'store_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'update_metrics'.
2026-06-10 07:49:35 | INFO     | __main__:update_metrics:27 - Updating metric 'feature_count_update_2' with value: 138
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'update_metrics' succeeded on attempt 1.
2026-06-10 07:49:35 | INFO     | __main__:<cell line: 0>:46 - --- Daily Update 3/5 ---
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'simulate_data_ingestion'.
2026-06-10 07:49:35 | INFO     | __main__:simulate_data_ingestion:26 - Simulating data ingestion...
2026-06-10 07:49:35 | INFO     | __main__:simulate_data_ingestion:50 - Ingested 3 rows of raw data for 3 entities.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'simulate_data_ingestion' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_lag_features'.
2026-06-10 07:49:35 | INFO     | __main__:generate_lag_features:26 - Generating lag features for periods: [1, 2, 3]...
2026-06-10 07:49:35 | INFO     | __main__:generate_lag_features:34 - Generated 3 lag features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3
0  entity_1 2023-01-07  55.194904          NaN          NaN          NaN
1  entity_1 2023-01-08  60.884464    55.194904          NaN          NaN
2  entity_1 2023-01-09  56.127975    60.884464    55.194904          NaN
3  entity_1 2023-01-10  63.471766    56.127975    60.884464    55.194904
4  entity_1 2023-01-11  56.198290    63.471766    56.127975    60.884464
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_lag_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_rolling_features'.
2026-06-10 07:49:35 | INFO     | __main__:generate_rolling_features:27 - Generating rolling features for window sizes: [3, 7]...
2026-06-10 07:49:35 | INFO     | __main__:generate_rolling_features:36 - Generated 4 rolling features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3  \
0  entity_1 2023-01-07  55.194904          NaN          NaN          NaN   
1  entity_1 2023-01-08  60.884464    55.194904          NaN          NaN   
2  entity_1 2023-01-09  56.127975    60.884464    55.194904          NaN   
3  entity_1 2023-01-10  63.471766    56.127975    60.884464    55.194904   
4  entity_1 2023-01-11  56.198290    63.471766    56.127975    60.884464   

   value_rolling_mean_3  value_rolling_std_3  value_rolling_mean_7  \
0                   NaN                  NaN                   NaN   
1                   NaN                  NaN                   NaN   
2             57.402448             3.051390                   NaN   
3             60.161402             3.724907                   NaN   
4             58.599344             4.219788                   NaN   

   value_rolling_std_7  
0                  NaN  
1                  NaN  
2                  NaN  
3                  NaN  
4                  NaN  
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_rolling_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'handle_missing_values'.
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:30 - Handling missing values with strategy: 'drop'...
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:36 - Dropped 18 rows with missing values.
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:48 - Missing values handled. Current data shape: (123, 10)
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'handle_missing_values' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'validate_features'.
2026-06-10 07:49:35 | INFO     | __main__:validate_features:28 - Validating engineered features...
2026-06-10 07:49:35 | WARNING  | __main__:validate_features:59 - Outliers detected in features:
{'value_rolling_std_7': np.int64(1)}
2026-06-10 07:49:35 | ERROR    | __main__:validate_features:65 - Feature validation found issues. See details in logs.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'validate_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'store_features'.
2026-06-10 07:49:35 | INFO     | __main__:store_features:26 - Storing features into the feature store...
2026-06-10 07:49:35 | INFO     | __main__:store_features:41 - Features stored. Feature store now contains 141 records.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'store_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'update_metrics'.
2026-06-10 07:49:35 | INFO     | __main__:update_metrics:27 - Updating metric 'feature_count_update_3' with value: 141
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'update_metrics' succeeded on attempt 1.
2026-06-10 07:49:35 | INFO     | __main__:<cell line: 0>:46 - --- Daily Update 4/5 ---
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'simulate_data_ingestion'.
2026-06-10 07:49:35 | INFO     | __main__:simulate_data_ingestion:26 - Simulating data ingestion...
2026-06-10 07:49:35 | INFO     | __main__:simulate_data_ingestion:50 - Ingested 3 rows of raw data for 3 entities.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'simulate_data_ingestion' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_lag_features'.
2026-06-10 07:49:35 | INFO     | __main__:generate_lag_features:26 - Generating lag features for periods: [1, 2, 3]...
2026-06-10 07:49:35 | INFO     | __main__:generate_lag_features:34 - Generated 3 lag features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3
0  entity_1 2023-01-07  55.194904          NaN          NaN          NaN
1  entity_1 2023-01-08  60.884464    55.194904          NaN          NaN
2  entity_1 2023-01-09  56.127975    60.884464    55.194904          NaN
3  entity_1 2023-01-10  63.471766    56.127975    60.884464    55.194904
4  entity_1 2023-01-11  56.198290    63.471766    56.127975    60.884464
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_lag_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_rolling_features'.
2026-06-10 07:49:35 | INFO     | __main__:generate_rolling_features:27 - Generating rolling features for window sizes: [3, 7]...
2026-06-10 07:49:35 | INFO     | __main__:generate_rolling_features:36 - Generated 4 rolling features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3  \
0  entity_1 2023-01-07  55.194904          NaN          NaN          NaN   
1  entity_1 2023-01-08  60.884464    55.194904          NaN          NaN   
2  entity_1 2023-01-09  56.127975    60.884464    55.194904          NaN   
3  entity_1 2023-01-10  63.471766    56.127975    60.884464    55.194904   
4  entity_1 2023-01-11  56.198290    63.471766    56.127975    60.884464   

   value_rolling_mean_3  value_rolling_std_3  value_rolling_mean_7  \
0                   NaN                  NaN                   NaN   
1                   NaN                  NaN                   NaN   
2             57.402448             3.051390                   NaN   
3             60.161402             3.724907                   NaN   
4             58.599344             4.219788                   NaN   

   value_rolling_std_7  
0                  NaN  
1                  NaN  
2                  NaN  
3                  NaN  
4                  NaN  
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_rolling_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'handle_missing_values'.
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:30 - Handling missing values with strategy: 'drop'...
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:36 - Dropped 18 rows with missing values.
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:48 - Missing values handled. Current data shape: (126, 10)
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'handle_missing_values' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'validate_features'.
2026-06-10 07:49:35 | INFO     | __main__:validate_features:28 - Validating engineered features...
2026-06-10 07:49:35 | WARNING  | __main__:validate_features:59 - Outliers detected in features:
{'value_rolling_std_7': np.int64(1)}
2026-06-10 07:49:35 | ERROR    | __main__:validate_features:65 - Feature validation found issues. See details in logs.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'validate_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'store_features'.
2026-06-10 07:49:35 | INFO     | __main__:store_features:26 - Storing features into the feature store...
2026-06-10 07:49:35 | INFO     | __main__:store_features:41 - Features stored. Feature store now contains 144 records.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'store_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'update_metrics'.
2026-06-10 07:49:35 | INFO     | __main__:update_metrics:27 - Updating metric 'feature_count_update_4' with value: 144
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'update_metrics' succeeded on attempt 1.
2026-06-10 07:49:35 | INFO     | __main__:<cell line: 0>:46 - --- Daily Update 5/5 ---
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'simulate_data_ingestion'.
2026-06-10 07:49:35 | INFO     | __main__:simulate_data_ingestion:26 - Simulating data ingestion...
2026-06-10 07:49:35 | INFO     | __main__:simulate_data_ingestion:50 - Ingested 3 rows of raw data for 3 entities.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'simulate_data_ingestion' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_lag_features'.
2026-06-10 07:49:35 | INFO     | __main__:generate_lag_features:26 - Generating lag features for periods: [1, 2, 3]...
2026-06-10 07:49:35 | INFO     | __main__:generate_lag_features:34 - Generated 3 lag features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3
0  entity_1 2023-01-07  55.194904          NaN          NaN          NaN
1  entity_1 2023-01-08  60.884464    55.194904          NaN          NaN
2  entity_1 2023-01-09  56.127975    60.884464    55.194904          NaN
3  entity_1 2023-01-10  63.471766    56.127975    60.884464    55.194904
4  entity_1 2023-01-11  56.198290    63.471766    56.127975    60.884464
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_lag_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'generate_rolling_features'.
2026-06-10 07:49:35 | INFO     | __main__:generate_rolling_features:27 - Generating rolling features for window sizes: [3, 7]...
2026-06-10 07:49:35 | INFO     | __main__:generate_rolling_features:36 - Generated 4 rolling features. First 5 rows:
  entity_id  timestamp      value  value_lag_1  value_lag_2  value_lag_3  \
0  entity_1 2023-01-07  55.194904          NaN          NaN          NaN   
1  entity_1 2023-01-08  60.884464    55.194904          NaN          NaN   
2  entity_1 2023-01-09  56.127975    60.884464    55.194904          NaN   
3  entity_1 2023-01-10  63.471766    56.127975    60.884464    55.194904   
4  entity_1 2023-01-11  56.198290    63.471766    56.127975    60.884464   

   value_rolling_mean_3  value_rolling_std_3  value_rolling_mean_7  \
0                   NaN                  NaN                   NaN   
1                   NaN                  NaN                   NaN   
2             57.402448             3.051390                   NaN   
3             60.161402             3.724907                   NaN   
4             58.599344             4.219788                   NaN   

   value_rolling_std_7  
0                  NaN  
1                  NaN  
2                  NaN  
3                  NaN  
4                  NaN  
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'generate_rolling_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'handle_missing_values'.
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:30 - Handling missing values with strategy: 'drop'...
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:36 - Dropped 18 rows with missing values.
2026-06-10 07:49:35 | INFO     | __main__:handle_missing_values:48 - Missing values handled. Current data shape: (129, 10)
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'handle_missing_values' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'validate_features'.
2026-06-10 07:49:35 | INFO     | __main__:validate_features:28 - Validating engineered features...
2026-06-10 07:49:35 | WARNING  | __main__:validate_features:59 - Outliers detected in features:
{'value_rolling_std_7': np.int64(1)}
2026-06-10 07:49:35 | ERROR    | __main__:validate_features:65 - Feature validation found issues. See details in logs.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'validate_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'store_features'.
2026-06-10 07:49:35 | INFO     | __main__:store_features:26 - Storing features into the feature store...
2026-06-10 07:49:35 | INFO     | __main__:store_features:41 - Features stored. Feature store now contains 147 records.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'store_features' succeeded on attempt 1.
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:43 - Attempt 1 for function 'update_metrics'.
2026-06-10 07:49:35 | INFO     | __main__:update_metrics:27 - Updating metric 'feature_count_update_5' with value: 147
2026-06-10 07:49:35 | DEBUG    | __main__:run_pipeline_step_with_jitter:45 - Function 'update_metrics' succeeded on attempt 1.
2026-06-10 07:49:35 | INFO     | __main__:<cell line: 0>:68 - Pipeline updates complete.
2026-06-10 07:49:35 | INFO     | __main__:<cell line: 0>:69 - Final feature store size: 147 records.
2026-06-10 07:49:35 | INFO     | __main__:<cell line: 0>:70 - Final feature store head:
entity_id timestamp value lag_1 rolling_mean_3 rolling_std_3 value_lag_1 value_lag_2 value_lag_3 value_rolling_mean_3 value_rolling_std_3 value_rolling_mean_7 value_rolling_std_7
142 entity_3 2023-02-20 50.426293 NaN NaN NaN 52.890507 58.128749 58.622324 53.815183 3.933602 54.510346 3.805781
143 entity_3 2023-02-22 48.257034 NaN NaN NaN 50.426293 52.890507 58.128749 50.524611 2.318300 53.350776 4.342085
144 entity_3 2023-02-25 44.981275 NaN NaN NaN 48.257034 50.426293 52.890507 47.888201 2.741183 51.746761 5.115158
145 entity_3 2023-03-01 44.223119 NaN NaN NaN 44.981275 48.257034 50.426293 45.820476 2.143901 51.075615 5.808859
146 entity_3 2023-03-06 50.580346 NaN NaN NaN 44.223119 44.981275 48.257034 46.594914 3.472241 49.926761 4.769879
cell output
2026-06-10 07:49:36 | INFO     | __main__:<cell line: 0>:84 - Pipeline Metrics Summary:
Value
validation_summary [{'type': 'Outliers', 'details': {'value_rolli...
initial_feature_count 132
feature_count_update_1 135
feature_count_update_2 138
feature_count_update_3 141
feature_count_update_4 144
feature_count_update_5 147

Production Considerations

Building an automated feature pipeline for production requires attention to several best practices to ensure reliability, scalability, and maintainability. Below is a table summarizing key considerations.

AspectBest PracticesImpact
ModularityBreak down the pipeline into small, testable, and reusable functions/components.Easier to debug, maintain, and scale; promotes team collaboration.
IdempotencyDesign pipeline steps so that running them multiple times produces the same result as running once.Enables safe retries and recovery from failures without data corruption.
Version ControlStore all pipeline code, feature definitions, and configurations in a version control system (e.g., Git).Tracks changes, facilitates collaboration, and allows rollback to previous versions.
TestingImplement unit tests for individual functions and integration tests for end-to-end pipeline flows.Catches bugs early, ensures correctness, and builds confidence in changes.
Monitoring & AlertingTrack key metrics (e.g., data freshness, feature distribution, processing times) and set up alerts for anomalies.Proactive identification of issues (data drift, pipeline failures) before they impact models.
ScalabilityUse distributed computing frameworks (e.g., Spark, Dask) for large datasets and parallel processing.Handles growing data volumes and complex feature computations efficiently.
Data GovernanceDocument data sources, transformations, and feature definitions; manage access control.Ensures data quality, compliance, and understanding across teams.
Error Handling & RetriesImplement robust try/except blocks with exponential backoff and jitter for transient failures.Improves pipeline resilience against temporary issues in external systems.
Feature StoreUtilize a dedicated feature store for centralized feature management and serving.Ensures consistency between training and inference, reduces feature re-computation.
ObservabilityImplement detailed logging, tracing, and metrics collection for each pipeline step.Provides deep insights into pipeline execution and performance for debugging and optimization.
SecuritySecure access to data sources, feature stores, and pipeline infrastructure; encrypt data in transit and at rest.Protects sensitive data and prevents unauthorized access or tampering.

Conclusion

This notebook has provided a structured approach to building an automated feature pipeline. We've covered the essential components from data ingestion and feature engineering (lagged and rolling statistics) to handling missing values, validating features, and storing them in a simulated feature store. The demonstration highlighted how these functions can be orchestrated to process data and generate valuable features for machine learning models. We also discussed critical production considerations, emphasizing the importance of modularity, robustness, scalability, and observability for a production-grade feature pipeline. Implementing these practices ensures a reliable and efficient system for delivering high-quality features.