from copy import deepcopy
from typing import Tuple, Union, List
import dill
import numpy as np
import pandas as pd
from .bivector import FEBivector, FEWedgeBivector
from .rotation import FEWhitening, FEShear
from .shape import FEGradeRatio, FEMultiGradeShape
from .row_geometry import FEDensity, FEBladeSpectrum
from .tree_features import FERFProximity, FESpectralProximity, FEDirectRS
from .phase import FEPhaseEncoder
from .kinematics import FEKinematics
from ..meta_info import DataMetaInfo, Feature
from ..utils.checks import check_type
[docs]
class FeatureEngineering:
def __init__(self, dataset):
self.__dataset = dataset
self.__raw_meta_info = None
self.__pending_steps = []
self.__executed_steps = []
[docs]
def reset(self):
self.__pending_steps = []
self.__executed_steps = []
if self.__raw_meta_info is not None:
self.__dataset._set_meta_info(meta_info=self.__raw_meta_info)
# ------------------
# Queue methods
# ------------------
[docs]
def fe_bivector(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
max_features: int = 300,
selection: str = "variance"):
kwargs = locals()
kwargs.pop('self')
test_obj = FEBivector(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_bivector",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
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')
test_obj = FEWedgeBivector(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_wedge_bivector",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
def fe_whitening(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
regularization: float = 1e-8):
kwargs = locals()
kwargs.pop('self')
test_obj = FEWhitening(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_whitening",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
def fe_shear(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
threshold: float = 0.0):
kwargs = locals()
kwargs.pop('self')
test_obj = FEShear(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_shear",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
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')
test_obj = FEGradeRatio(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_grade_ratio",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
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')
test_obj = FEMultiGradeShape(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_multi_grade_shape",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
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')
test_obj = FEDensity(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_density",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
def fe_blade_spectrum(self,
features: Union[str, Tuple] = None,
dataset: str = "main",
n_neighbors: int = 10):
kwargs = locals()
kwargs.pop('self')
test_obj = FEBladeSpectrum(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_blade_spectrum",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
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')
test_obj = FERFProximity(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_rf_proximity",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
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')
test_obj = FESpectralProximity(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_spectral_proximity",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
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')
test_obj = FEDirectRS(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_direct_rs",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
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')
test_obj = FEPhaseEncoder(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_phase_encoder",
"test_obj": test_obj,
"kwargs": kwargs})
[docs]
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')
test_obj = FEKinematics(dataset=self.__dataset)
self.__pending_steps.append({"name": "fe_kinematics",
"test_obj": test_obj,
"kwargs": kwargs})
# ------------------
# Execution
# ------------------
[docs]
@staticmethod
def engineer_one_step(data, step):
if not step["test_obj"].fitted_:
step["test_obj"].run(**step["kwargs"])
new_cols_df = step["test_obj"].transform(data)
# Append new columns (feature engineering adds, not replaces)
engineered_data = pd.concat([data.reset_index(drop=True),
new_cols_df.reset_index(drop=True)], axis=1)
return engineered_data
[docs]
def engineer(self, data):
new_data = data
self.__raw_meta_info = deepcopy(self.__dataset.meta_info)
while len(self.__pending_steps) > 0:
step = self.__pending_steps.pop(0)
new_data = self.engineer_one_step(data=new_data, step=step)
self.__executed_steps.append(step)
# Update meta_info with new columns
if list(new_data.columns) != list(self.__dataset.data.columns):
features = {}
features_dict = self.__dataset.meta_info.features_to_dict()
for fidx, fn in enumerate(new_data.columns):
type_value = check_type(new_data.iloc[:, [fidx]])[0][0]
if fn in features_dict:
features[fn] = features_dict[fn]
features[fn]["id"] = fidx
features[fn]["type"] = type_value
else:
features[fn] = Feature(id=fidx,
name=fn,
type=type_value).to_dict()
meta_info = DataMetaInfo(name=self.__dataset.meta_info.name,
description=self.__dataset.meta_info.description,
tags=self.__dataset.meta_info.tags,
features=features,
task_type=self.__dataset.task_type,
train_idx=self.__dataset.train_idx.tolist(),
test_idx=self.__dataset.test_idx.tolist(),
inactive_sample_idx=self.__dataset.inactive_sample_idx,
version=self.__dataset.version)
self.__dataset._set_meta_info(meta_info=meta_info)
self.__dataset._set_data(data=new_data)
[docs]
def save(self, path_or_buf):
status = {"raw_meta_info": self.__raw_meta_info,
"pending_steps": self.__pending_steps,
"executed_steps": self.__executed_steps
}
dill.dump(status, open(path_or_buf, "wb"))
[docs]
def load(self, path_or_buf):
status = dill.load(open(path_or_buf, "rb"))
self.__raw_meta_info = status["raw_meta_info"]
self.__pending_steps = status["pending_steps"]
self.__executed_steps = status["executed_steps"]
[docs]
def get_status(self):
return {"raw_meta_info": self.__raw_meta_info,
"pending_steps": self.__pending_steps,
"executed_steps": self.__executed_steps
}