KETI Time Series Preprocessing

A comprehensive time series data preprocessing pipeline designed for robust data cleaning, transformation, and preparation. This library provides a modular approach to handle common time series preprocessing tasks including data refinement, outlier detection, missing value imputation, scaling, and smoothing.

🎯 Overview

KETI Time Series Preprocessing offers a pipeline-based architecture that processes time series data through configurable modules. Each module handles specific preprocessing tasks while maintaining data integrity and ensuring compatibility with downstream analysis.

✨ Key Features

🔧 Modular Pipeline Architecture

Flexible Module Composition: Combine preprocessing steps in custom sequences
Pipeline Validation: Automatic compatibility checking between modules
Type Safety: Enforced input/output type validation (DataFrame ↔ DataFrameSet)
Error Handling: Robust error detection and recovery mechanisms

🧹 Data Refinement

Duplicate Removal: Eliminate redundant timestamps and values
Frequency Standardization: Automatic resampling to consistent intervals
Index Alignment: Proper datetime indexing and sorting
Data Type Enforcement: Configurable column type validation

🔍 Outlier & Error Detection

Certain Error Detection: Rule-based removal of known invalid values
Uncertain Error Detection: Statistical anomaly detection (IQR, Z-score, etc.)
Range Validation: Min/max boundary checking
Abnormal Value Filtering: Predefined error code removal

🩹 Missing Data Imputation

Multiple Methods: Linear, KNN, MICE, forward/backward fill
Sequential Application: Apply different methods based on gap size
Adaptive Strategies: Method selection based on missing data patterns
Quality Control: Data availability thresholds

📊 Data Transformation

Scaling Methods: MinMax, Standard, Robust, MaxAbs scalers
Smoothing: Exponentially Weighted Moving Average (EWMA)
Frequency Inference: Automatic detection of sampling rates
Batch Processing: Handle single DataFrames or dataset collections

🚀 Quick Start

Installation

git clone https://github.com/your-repo/keti-ts-preprocessing.git
cd keti-ts-preprocessing
pip install -r requirements.txt

Basic Usage

1. Simple Pipeline Example

import pandas as pd
from data_pipeline import pipeline

# Load time series data
data = pd.read_csv('sensor_data.csv', parse_dates=['time'], index_col='time')

# Define preprocessing pipeline
pipeline_steps = [
    ['data_refinement', {
        'remove_duplication': {'flag': True}, 
        'static_frequency': {'flag': True, 'frequency': None}  # Auto-detect frequency
    }],
    ['data_outlier', {
        'certain_error_to_NaN': {
            'flag': True,
            'data_min_max_limit': {
                'max_num': {'temperature': 50, 'humidity': 100},
                'min_num': {'temperature': -20, 'humidity': 0}
            }
        },
        'uncertain_error_to_NaN': {
            'flag': True,
            'param': {'outlierDetectorConfig': [{
                'algorithm': 'IQR',
                'percentile': 95,
                'alg_parameter': {'weight': 80}
            }]}
        }
    }],
    ['data_imputation', {
        'flag': True,
        'imputation_method': [
            {'min': 0, 'max': 5, 'method': 'linear', 'parameter': {}},
            {'min': 6, 'max': 10, 'method': 'KNN', 'parameter': {}}
        ],
        'total_non_NaN_ratio': 70
    }],
    ['data_scaling', {
        'flag': True, 
        'method': 'minmax'
    }]
]

# Execute pipeline
processed_data = pipeline(data, pipeline_steps)

# Save results
processed_data.to_csv('cleaned_data.csv')

2. Advanced Configuration with Type Validation

import pandas as pd
from data_pipeline import pipeline, pipeline_connection_check

# Load data
data = pd.read_csv('environmental_data.csv', parse_dates=['time'], index_col='time')

# Define expected data types
expected_types = {
    'temperature': float,
    'humidity': float,
    'co2_level': int,
    'pressure': float
}

# Define range limits
range_limits = {
    'max_num': {
        'temperature': 40, 'humidity': 100, 
        'co2_level': 5000, 'pressure': 1100
    },
    'min_num': {
        'temperature': -10, 'humidity': 0, 
        'co2_level': 300, 'pressure': 900
    }
}

# Configure comprehensive pipeline
comprehensive_pipeline = [
    ['data_refinement', {
        'remove_duplication': {'flag': True},
        'static_frequency': {'flag': True, 'frequency': pd.Timedelta('10min')}
    }],
    ['data_outlier', {
        'certain_error_to_NaN': {
            'flag': True,
            'data_min_max_limit': range_limits,
            'abnormal_value_list': {'all': [9999, -9999, 99.9, -99.9]}
        },
        'uncertain_error_to_NaN': {
            'flag': True,
            'param': {'outlierDetectorConfig': [{
                'algorithm': 'IQR',
                'percentile': 99,
                'alg_parameter': {'weight': 85}
            }]}
        }
    }],
    ['data_imputation', {
        'flag': True,
        'imputation_method': [
            {'min': 0, 'max': 3, 'method': 'linear', 'parameter': {}},
            {'min': 4, 'max': 10, 'method': 'KNN', 'parameter': {'n_neighbors': 5}},
            {'min': 11, 'max': 20, 'method': 'mean', 'parameter': {}}
        ],
        'total_non_NaN_ratio': 80
    }],
    ['data_smoothing', {
        'flag': True,
        'emw_param': 0.3  # EWMA smoothing parameter
    }],
    ['data_scaling', {
        'flag': True,
        'method': 'standard'  # Z-score normalization
    }]
]

# Validate pipeline compatibility
data_type = 'DF'  # Single DataFrame input
is_valid = pipeline_connection_check(comprehensive_pipeline, data_type)

if is_valid:
    # Execute pipeline with type validation
    result = pipeline(
        data, 
        comprehensive_pipeline, 
        expected_types=expected_types
    )
    print("✅ Pipeline executed successfully!")
    print(f"📊 Output shape: {result.shape}")
    print(f"🧹 Missing values: {result.isnull().sum().sum()}")
else:
    print("❌ Pipeline validation failed!")

3. Dataset Collection Processing

import pandas as pd
from pathlib import Path
from data_pipeline import pipeline

# Load multiple CSV files
data_dir = Path('./sensor_data/')
dataset = {}

for csv_file in data_dir.glob('*.csv'):
    df = pd.read_csv(csv_file, parse_dates=['time'], index_col='time')
    dataset[csv_file.stem] = df

print(f"📁 Loaded {len(dataset)} datasets")

# Pipeline for dataset collection (DFSet)
dataset_pipeline = [
    ['data_refinement', {
        'remove_duplication': {'flag': True},
        'static_frequency': {'flag': True, 'frequency': '15min'}
    }],
    ['data_imputation', {
        'flag': True,
        'imputation_method': [
            {'min': 0, 'max': 10, 'method': 'linear', 'parameter': {}}
        ],
        'total_non_NaN_ratio': 60
    }]
]

# Process all datasets
processed_dataset = pipeline(dataset, dataset_pipeline)

# Save processed datasets
output_dir = Path('./processed_data/')
output_dir.mkdir(exist_ok=True)

for name, df in processed_dataset.items():
    df.to_csv(output_dir / f"{name}_processed.csv")
    print(f"💾 Saved: {name}_processed.csv")

📚 Core Modules

🔧 `data_pipeline.py`

Main pipeline orchestrator and module coordinator.

Key Functions:

pipeline(data, module_list, feature_name=None, expected_types=None): Execute preprocessing pipeline
pipeline_connection_check(pipeline, input_type): Validate module compatibility
enforce_column_types(df, expected_types): Apply data type constraints

🧹 `preprocessing/dataPreprocessing.py`

Core preprocessing operations and interfaces.

Key Methods:

get_refinedData(data, refine_param): Data refinement and frequency standardization
get_errorToNaNData(data, outlier_param): Outlier detection and removal
get_imputedData(data, imputation_param): Missing value imputation
get_scaling_data(data, scaling_param): Data normalization and scaling
get_smoothed_data(data, smoothing_param): Time series smoothing

🔍 `preprocessing/errorDetection/`

Outlier and anomaly detection modules.

Components:

errorToNaN.py: Main outlier detection interface
certainError.py: Rule-based error removal
unCertainError.py: Statistical anomaly detection
dataRangeInfo_manager.py: Range validation utilities

🩹 `preprocessing/imputation/`

Missing data imputation strategies.

Components:

Imputation.py: Main imputation orchestrator
basicMethod.py: Simple imputation methods (mean, median, KNN)
DLMethod.py: Deep learning-based imputation
nanMasking.py: Missing data pattern analysis

📏 `preprocessing/refinement/`

Data cleaning and standardization utilities.

Components:

frequency.py: Frequency detection and resampling
redundancy.py: Duplicate removal and data deduplication

⚙️ Configuration Guide

Pipeline Module Configuration

Data Refinement

refine_config = {
    'remove_duplication': {'flag': True},
    'static_frequency': {
        'flag': True,
        'frequency': None  # Auto-detect | '10min' | pd.Timedelta('15min')
    }
}

Outlier Detection

outlier_config = {
    'certain_error_to_NaN': {
        'flag': True,
        'data_min_max_limit': {
            'max_num': {'temp': 50, 'humidity': 100},
            'min_num': {'temp': -20, 'humidity': 0}
        },
        'abnormal_value_list': {'all': [9999, -9999]}
    },
    'uncertain_error_to_NaN': {
        'flag': True,
        'param': {
            'outlierDetectorConfig': [{
                'algorithm': 'IQR',  # 'IQR' | 'zscore' | 'isolation'
                'percentile': 95,
                'alg_parameter': {'weight': 80}
            }]
        }
    }
}

Missing Data Imputation

imputation_config = {
    'flag': True,
    'imputation_method': [
        {'min': 0, 'max': 5, 'method': 'linear', 'parameter': {}},
        {'min': 6, 'max': 15, 'method': 'KNN', 'parameter': {'n_neighbors': 5}},
        {'min': 16, 'max': 50, 'method': 'mean', 'parameter': {}},
        {'min': 51, 'max': 100, 'method': 'ffill', 'parameter': {}}
    ],
    'total_non_NaN_ratio': 70  # Minimum percentage of non-NaN values required
}

Data Scaling

scaling_config = {
    'flag': True,
    'method': 'minmax'  # 'minmax' | 'standard' | 'robust' | 'maxabs'
}

Data Smoothing

smoothing_config = {
    'flag': True,
    'emw_param': 0.3  # Exponential smoothing parameter (0 < value < 1)
}

Pipeline Module Compatibility

Input Type	Module	Output Type
DF	data_refinement	DF
DFSet	data_refinement	DFSet
DF	data_outlier	DF
DFSet	data_outlier	DFSet
DF	data_imputation	DF
DFSet	data_imputation	DFSet
DF	data_scaling	DF
DFSet	data_scaling	DFSet
DF	data_smoothing	DF
DFSet	data_smoothing	DFSet

Legend:

DF: Single pandas DataFrame
DFSet: Dictionary of DataFrames

📁 Input Data Format

CSV Time Series Format

time,temperature,humidity,pressure,co2_level
2023-01-01 00:00:00,22.5,65.2,1013.2,400
2023-01-01 00:10:00,22.3,65.8,1013.1,405
2023-01-01 00:20:00,22.1,66.1,1012.9,410
2023-01-01 00:30:00,9999,9999,9999,9999  # Error values
2023-01-01 00:40:00,21.8,67.2,1012.5,415

Expected Data Types

expected_types = {
    'temperature': float,
    'humidity': float, 
    'pressure': float,
    'co2_level': int
}

Range Validation

range_limits = {
    'max_num': {
        'temperature': 40,
        'humidity': 100,
        'pressure': 1100,
        'co2_level': 5000
    },
    'min_num': {
        'temperature': -10,
        'humidity': 0,
        'pressure': 900,
        'co2_level': 300
    }
}

📊 Output & Monitoring

Pipeline Execution Monitoring

The pipeline provides real-time feedback during processing:

data_refinement
{'remove_duplication': {'flag': True}, 'static_frequency': {'flag': True, 'frequency': None}}
(1440, 4)

data_outlier  
{'certain_error_to_NaN': {'flag': True, 'data_min_max_limit': {...}}}
(1440, 4)

data_imputation
{'flag': True, 'imputation_method': [...]}
(1440, 4)

Quality Metrics

# After processing, check data quality
print(f"📊 Original shape: {original_data.shape}")
print(f"📊 Processed shape: {processed_data.shape}")
print(f"🧹 Missing values removed: {original_data.isnull().sum().sum() - processed_data.isnull().sum().sum()}")
print(f"📈 Data completeness: {(1 - processed_data.isnull().sum().sum() / processed_data.size) * 100:.1f}%")

🧪 Testing

Run the preprocessing test:

python preprocessing_test.py

This will:

✅ Load sample sensor data from raw.csv
✅ Apply comprehensive preprocessing pipeline
✅ Validate pipeline module compatibility
✅ Generate processed_data.csv with cleaned results
✅ Display processing statistics and data quality metrics

📦 Dependencies

Core Requirements:

pandas >= 1.3.0: Data manipulation and time series handling
numpy >= 1.21.0: Numerical computations and array operations
scikit-learn >= 1.0.0: Machine learning preprocessing utilities

Optional Extensions:

scipy >= 1.7.0: Advanced statistical computations
matplotlib >= 3.3.0: Visualization for EDA features
seaborn >= 0.11.0: Statistical plotting

🛣️ Roadmap

Advanced Imputation: LSTM, Transformer-based gap filling
Real-time Processing: Streaming data pipeline support
Quality Scoring: Automated data quality assessment
Visualization Dashboard: Interactive preprocessing monitoring
Configuration Templates: Pre-defined pipelines for common use cases
Performance Optimization: Parallel processing for large datasets
Export Formats: Support for Parquet, HDF5, and other formats

🎯 Use Cases

🏭 Industrial IoT Sensors

# Factory sensor data preprocessing
industrial_pipeline = [
    ['data_refinement', {'static_frequency': {'flag': True, 'frequency': '1min'}}],
    ['data_outlier', {'certain_error_to_NaN': {'flag': True, 'data_min_max_limit': sensor_ranges}}],
    ['data_imputation', {'flag': True, 'imputation_method': [{'min': 0, 'max': 5, 'method': 'linear'}]}]
]

🌡️ Environmental Monitoring

# Weather station data cleaning
environmental_pipeline = [
    ['data_refinement', {'remove_duplication': {'flag': True}}],
    ['data_outlier', {'uncertain_error_to_NaN': {'flag': True, 'param': {'outlierDetectorConfig': [{'algorithm': 'IQR'}]}}}],
    ['data_smoothing', {'flag': True, 'emw_param': 0.2}]
]

🏠 Smart Building Systems

# Building automation preprocessing
building_pipeline = [
    ['data_refinement', {'static_frequency': {'flag': True, 'frequency': '5min'}}],
    ['data_imputation', {'flag': True, 'total_non_NaN_ratio': 90}],
    ['data_scaling', {'flag': True, 'method': 'minmax'}]
]

🏢 Acknowledgments

KETI (Korea Electronics Technology Institute): Research foundation and domain expertise
Pandas Community: Excellent time series data structures and operations
Scikit-learn Contributors: Robust preprocessing and machine learning utilities
Open Source Ecosystem: Libraries and tools that enable this project

For questions, bug reports, or feature requests, please visit our GitHub Issues page.

📄 Contributor License Agreement (CLA)

본 프로젝트는 기여(Contribution)를 받기 위해 Contributor License Agreement(CLA) 서명을 요구합니다.

Pull Request를 제출하면 CLA Assistant가 자동으로 서명 여부를 확인하며,
최초 1회 온라인 동의를 완료해야 기여가 포함될 수 있습니다.

🔗 CLA 전문 보기: KETI Contributor License Agreement

📝 License

본 프로젝트는 Apache License 2.0을 따릅니다.
자세한 내용은 LICENSE 파일을 참고하세요.

Name		Name	Last commit message	Last commit date
.ipynb_checkpoints		.ipynb_checkpoints
__pycache__		__pycache__
preprocessing		preprocessing
LICENSE		LICENSE
README.md		README.md
data_pipeline.py		data_pipeline.py
preprocessing_test.py		preprocessing_test.py
raw.csv		raw.csv

Latest commit

History

Folders and files

Repository files navigation

KETI Time Series Preprocessing

🎯 Overview

✨ Key Features

🔧 Modular Pipeline Architecture

🧹 Data Refinement

🔍 Outlier & Error Detection

🩹 Missing Data Imputation

📊 Data Transformation

🚀 Quick Start

Installation

Basic Usage

1. Simple Pipeline Example

2. Advanced Configuration with Type Validation

3. Dataset Collection Processing

📚 Core Modules

🔧 data_pipeline.py

🧹 preprocessing/dataPreprocessing.py

🔍 preprocessing/errorDetection/

🩹 preprocessing/imputation/

📏 preprocessing/refinement/

⚙️ Configuration Guide

Pipeline Module Configuration

Data Refinement

Outlier Detection

Missing Data Imputation

Data Scaling

Data Smoothing

Pipeline Module Compatibility

📁 Input Data Format

CSV Time Series Format

Expected Data Types

Range Validation

📊 Output & Monitoring

Pipeline Execution Monitoring

Quality Metrics

🧪 Testing

📦 Dependencies

🛣️ Roadmap

🎯 Use Cases

🏭 Industrial IoT Sensors

🌡️ Environmental Monitoring

🏠 Smart Building Systems

🏢 Acknowledgments

📄 Contributor License Agreement (CLA)

📝 License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages

🔧 `data_pipeline.py`

🧹 `preprocessing/dataPreprocessing.py`

🔍 `preprocessing/errorDetection/`

🩹 `preprocessing/imputation/`

📏 `preprocessing/refinement/`