from pathlib import Path
from typing import Union, List, Tuple, Dict, Optional
import numpy as np
from .base import DataSetBase
from .data_tests.data_drift import DataDriftTest
from .exploration.base import DataExplorer
from .exploration.eda1d import EDA1D
from .exploration.eda2d import EDA2D
from .exploration.eda3d import EDA3D
from .exploration.eda_correlation import EDACorrelation
from .exploration.eda_pca import EDAPCA
from .exploration.eda_umap import EDAUMAP
from .exploration.summary import DataSummary
from .feature_selection.base import FeatureSelection
from .feature_selection.correlation import FeatureSelectCORR
from .feature_selection.rcit import FeatureSelectRCIT
from .feature_selection.xgbpfi import FeatureSelectXGBPFI
from .outlier_detection.base import OutlierDetection
from .outlier_detection.cblof import ODCBLOF
from .outlier_detection.isolation_forest import ODIsolationForest
from .outlier_detection.pca import ODPCA
from .feature_engineering.base import FeatureEngineering
from .feature_engineering.bivector import FEBivector, FEWedgeBivector
from .feature_engineering.rotation import FEWhitening, FEShear
from .feature_engineering.shape import FEGradeRatio, FEMultiGradeShape
from .feature_engineering.row_geometry import FEDensity, FEBladeSpectrum
from .feature_engineering.tree_features import FERFProximity, FESpectralProximity, FEDirectRS
from .feature_engineering.phase import FEPhaseEncoder
from .feature_engineering.kinematics import FEKinematics
from .preprocessing.base import Preprocessing
from .preprocessing.binning import NumericalBinning
from .preprocessing.encoding import CategoricalEncoding
from .preprocessing.imputataion import DataImputation
from .preprocessing.scaling import NumericalScaling
from .subsampling.base import SubSampling
from .subsampling.random import SubsampleRandom
from ..auth import auth
from ..utils.helper import add_docs
[docs]
class LocalDataSet(DataSetBase):
def __init__(self, name: str = "dataset", experiment_name: str = "default"):
super().__init__(name=name, experiment_name=experiment_name)
self.__explorer = DataExplorer(dataset=self)
self.__preprocessing = Preprocessing(dataset=self)
self.__feature_engineering = FeatureEngineering(dataset=self)
self.__feature_selection = FeatureSelection(dataset=self)
self.__outlier_detection = OutlierDetection(dataset=self)
self.__subsampling = SubSampling(dataset=self)
self.__data_testing = DataDriftTest(dataset=self)
self._init_mlflow_settings()
auth.run()
## alias functions
self.plot = self.eda_1d
# ------------------
# Data Exploration
# ------------------
[docs]
@add_docs(DataSummary.run.__doc__)
def summary(self,
dataset: str = "main"):
kwargs = locals()
kwargs.pop('self')
return self.__explorer.summary(**kwargs)
[docs]
@add_docs(EDA1D.run.__doc__)
def eda_1d(self,
feature: str,
dataset: str = "main",
plot_type: str = "density",
bins: int = 10,
sample_size: int = None,
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__explorer.eda_1d(**kwargs)
[docs]
@add_docs(EDA2D.run.__doc__)
def eda_2d(self,
feature_x: str,
feature_y: str,
feature_color: str = None,
dataset: str = "main",
sample_size: int = None,
smoother_order: int = None,
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__explorer.eda_2d(**kwargs)
[docs]
@add_docs(EDA3D.run.__doc__)
def eda_3d(self,
feature_x: str,
feature_y: str,
feature_z: str,
feature_color: str = None,
dataset: str = "main",
sample_size: int = 1000,
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__explorer.eda_3d(**kwargs)
[docs]
@add_docs(EDACorrelation.run.__doc__)
def eda_correlation(self,
features: Union[Tuple, List] = None,
dataset: str = "main",
method: str = "pearson",
sample_size: int = None,
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__explorer.eda_correlation(**kwargs)
[docs]
@add_docs(EDAPCA.run.__doc__)
def eda_pca(self,
features: Union[Tuple, List] = None,
n_components: int = None,
dataset: str = "main",
sample_size: int = None,
categorical_encoding: str = "ordinal",
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__explorer.eda_pca(**kwargs)
[docs]
@add_docs(EDAUMAP.run.__doc__)
def eda_umap(self,
features: Union[Tuple, List] = None,
n_neighbors: int = 15,
n_components: int = 2,
metric: str = "euclidean",
dataset: str = "main",
sample_size: int = None,
categorical_encoding: str = "ordinal",
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__explorer.eda_umap(**kwargs)
# ------------------
# Preprocessing
# ------------------
[docs]
@add_docs(DataImputation.run.__doc__)
def impute_missing(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
method: str = "mean",
fill_value: Union[int, float, str] = None,
missing_values: Union[int, float, str] = np.nan,
special_values: Union[int, float, str, list, tuple] = None,
add_indicators: bool = True):
kwargs = locals()
kwargs.pop('self')
self.__preprocessing.impute_missing(**kwargs)
[docs]
@add_docs(NumericalScaling.run.__doc__)
def scale_numerical(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
method: str = "minmax",
minmax_range: Optional[tuple] = (0, 1),
n_quantiles: int = 1000):
kwargs = locals()
kwargs.pop('self')
self.__preprocessing.scale_numerical(**kwargs)
[docs]
@add_docs(NumericalBinning.run.__doc__)
def bin_numerical(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
method: str = "uniform",
bins: Union[int, Dict] = 10):
kwargs = locals()
kwargs.pop('self')
self.__preprocessing.bin_numerical(**kwargs)
[docs]
@add_docs(CategoricalEncoding.run.__doc__)
def encode_categorical(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
method: str = "ordinal",
target: str = None):
kwargs = locals()
kwargs.pop('self')
self.__preprocessing.encode_categorical(**kwargs)
def _preprocess_data(self):
"""Preprocess raw data and extra raw data (if exist).
"""
for name in self.get_extra_data_list():
data = self.raw_extra_data[name]
new_data = self.__preprocessing.transform(data)
self._set_extra_data(name=name, data=new_data)
[docs]
def get_preprocessor(self):
"""Get preprocessor.
"""
return self.__preprocessing
[docs]
def reset_preprocess(self):
"""Remove all previous preprocess steps.
"""
self.__preprocessing.reset()
self._set_data()
self._set_extra_data()
[docs]
def preprocess(self):
"""Preprocess the internal raw data and extra data based on existing settings.
"""
self.__preprocessing.preprocess(self.to_df(raw_data=True))
self._preprocess_data()
# ------------------
# Feature Engineering
# ------------------
[docs]
@add_docs(FEBivector.run.__doc__)
def fe_bivector(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
max_features: int = 300,
selection: str = "variance"):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_bivector(**kwargs)
[docs]
@add_docs(FEWedgeBivector.run.__doc__)
def fe_wedge_bivector(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
max_features: int = 300,
selection: str = "variance"):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_wedge_bivector(**kwargs)
[docs]
@add_docs(FEWhitening.run.__doc__)
def fe_whitening(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
regularization: float = 1e-8):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_whitening(**kwargs)
[docs]
@add_docs(FEShear.run.__doc__)
def fe_shear(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
threshold: float = 0.0):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_shear(**kwargs)
[docs]
@add_docs(FEGradeRatio.run.__doc__)
def fe_grade_ratio(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
max_triplets: int = 50,
regularization: float = 1e-8):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_grade_ratio(**kwargs)
[docs]
@add_docs(FEMultiGradeShape.run.__doc__)
def fe_multi_grade_shape(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
grades: List[int] = None,
n_bins: int = 5,
regularization: float = 1e-8):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_multi_grade_shape(**kwargs)
[docs]
@add_docs(FEDensity.run.__doc__)
def fe_density(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
bandwidth: Union[str, float] = "median",
n_reference: int = 500,
log_density: bool = True):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_density(**kwargs)
[docs]
@add_docs(FEBladeSpectrum.run.__doc__)
def fe_blade_spectrum(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
n_neighbors: int = 10):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_blade_spectrum(**kwargs)
[docs]
@add_docs(FERFProximity.run.__doc__)
def fe_rf_proximity(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
n_estimators: int = 100,
max_depth: int = 6,
n_landmarks: int = 50):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_rf_proximity(**kwargs)
[docs]
@add_docs(FESpectralProximity.run.__doc__)
def fe_spectral_proximity(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
n_estimators: int = 100,
max_depth: int = 6,
n_components: int = 10):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_spectral_proximity(**kwargs)
[docs]
@add_docs(FEDirectRS.run.__doc__)
def fe_direct_rs(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
n_estimators: int = 50,
max_depth: int = 4,
regularization: float = 1e-8):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_direct_rs(**kwargs)
[docs]
@add_docs(FEPhaseEncoder.run.__doc__)
def fe_phase_encoder(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
n_thresholds: int = 5,
include_original: bool = False):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_phase_encoder(**kwargs)
[docs]
@add_docs(FEKinematics.run.__doc__)
def fe_kinematics(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
n_timesteps: int = None,
n_features_per_step: int = None):
kwargs = locals()
kwargs.pop('self')
self.__feature_engineering.fe_kinematics(**kwargs)
def _engineer_extra_data(self):
"""Apply fitted feature engineering to extra data (if exist).
"""
for name in self.get_extra_data_list():
data = self.extra_data[name]
new_data = self.__feature_engineering.transform(data)
self._set_extra_data(name=name, data=new_data)
[docs]
def get_feature_engineer(self):
"""Get the feature engineering aggregator.
"""
return self.__feature_engineering
[docs]
def reset_feature_engineering(self):
"""Remove all previous feature engineering steps.
"""
self.__feature_engineering.reset()
self._set_data()
self._set_extra_data()
[docs]
def engineer_features(self):
"""Execute all queued feature engineering steps on the current data.
"""
self.__feature_engineering.engineer(self.to_df())
self._engineer_extra_data()
# ------------------
# Feature Selection
# ------------------
[docs]
@add_docs(FeatureSelectCORR.run.__doc__)
def feature_select_corr(self,
dataset: str = "train",
method: str = "pearson",
threshold: float = 0.2,
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__feature_selection.feature_select_corr(**kwargs)
[docs]
@add_docs(FeatureSelectXGBPFI.run.__doc__)
def feature_select_xgbpfi(self,
dataset: str = "train",
n_repeats: int = 10,
threshold: float = 0.1,
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__feature_selection.feature_select_xgbpfi(**kwargs)
[docs]
@add_docs(FeatureSelectRCIT.run.__doc__)
def feature_select_rcit(self,
dataset: str = "train",
threshold: float = 1e-6,
n_fourier: int = 25,
n_fourier2: int = 5,
n_forwards: int = 2,
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__feature_selection.feature_select_rcit(**kwargs)
# ------------------
# Outlier Detection
# ------------------
[docs]
@add_docs(ODPCA.run.__doc__)
def detect_outlier_pca(self,
dataset: str = "main",
threshold: float = 0.99,
cumulative_variance_threshold: float = 0.9,
method: str = "mahalanobis",
sparse_pca: bool = False,
alpha: float = 1.0,
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__outlier_detection.detect_outlier_pca(**kwargs)
[docs]
@add_docs(ODIsolationForest.run.__doc__)
def detect_outlier_isolation_forest(self,
dataset: str = "main",
threshold: float = 0.99,
n_estimators: int = 100):
kwargs = locals()
kwargs.pop('self')
return self.__outlier_detection.detect_outlier_isolation_forest(**kwargs)
[docs]
@add_docs(ODCBLOF.run.__doc__)
def detect_outlier_cblof(self,
dataset: str = "main",
threshold: float = 0.99,
method: str = 'kmeans',
n_clusters: int = 10,
cluster_threshold: float = 0.1,
use_weights: bool = False,
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__outlier_detection.detect_outlier_cblof(**kwargs)
# ------------------
# Subsampling
# ------------------
[docs]
@add_docs(SubsampleRandom.run.__doc__)
def subsample_random(self,
dataset: str = "main",
sample_size: [int, float] = 0.2,
shuffle: bool = True,
stratify: str = None,
random_state: int = 0):
kwargs = locals()
kwargs.pop('self')
return self.__subsampling.subsample_random(**kwargs)
# ------------------
# Data Testing
# ------------------
[docs]
@add_docs(DataDriftTest.run.__doc__)
def data_drift_test(self,
dataset1: str = "train",
dataset2: str = "test",
sample_idx1: Union[Tuple[int], np.ndarray] = None,
sample_idx2: Union[Tuple[int], np.ndarray] = None,
name1: str = None,
name2: str = None,
distance_metric: str = "PSI",
psi_method: str = "uniform",
psi_bins: int = 10):
kwargs = locals()
kwargs.pop('self')
return self.__data_testing.run(**kwargs)
[docs]
def save_preprocessing(self, path_or_buf: Union[str, Path], run_id: str):
"""Save the preprocessing steps to file.
Parameters
----------
path_or_buf : str or Path
The path of the file.
run_id : str
The run id of mlflow.
"""
full_path = f"./{self.name}.pkl" # default file name
if path_or_buf is not None:
full_path = path_or_buf + ".pkl"
self.__preprocessing.save(full_path)
self.mlflow_client.log_artifact(run_id, full_path)
[docs]
def load_preprocessing(self, path_or_buf: Union[str, Path]):
"""Load preprocessing steps from a saved file.
Parameters
----------
path_or_buf : str or Path
The path to the file containing saved preprocessing steps. The file should have been created
using the save_preprocessing method.
Examples
--------
>>> ds.load_preprocessing("path/to/preprocessing.pkl")
"""
self.__preprocessing.load(path_or_buf)