-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathpreprocessing_utils.py
More file actions
104 lines (87 loc) · 3.98 KB
/
Copy pathpreprocessing_utils.py
File metadata and controls
104 lines (87 loc) · 3.98 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
def create_features(self, df, feature_definitions=None):
"""
Create engineered features.
Args:
df (pd.DataFrame): Input DataFrame
feature_definitions (list): List of feature definitions
Returns:
pd.DataFrame: DataFrame with engineered features
"""
df_features = df.copy()
# Get numeric columns for dynamic feature creation
numeric_cols = df_features.select_dtypes(include=[np.number]).columns.tolist()
# Remove target column from feature engineering if it exists
if 'target' in self.config and self.config['target'] in numeric_cols:
numeric_cols.remove(self.config['target'])
# Default feature engineering if none provided
if not feature_definitions:
feature_definitions = []
# Create features based on available columns
if len(numeric_cols) >= 2:
# Create ratio features if we have at least 2 numeric columns
feature_definitions.append({
'name': f'{numeric_cols[0]}_to_{numeric_cols[1]}_ratio',
'formula': f'{numeric_cols[0]} / ({numeric_cols[1]} + 1)', # +1 to avoid division by zero
'description': f'Ratio of {numeric_cols[0]} to {numeric_cols[1]}'
})
# Create log transformation for each numeric column
for col in numeric_cols:
# Check if column has non-negative values for log transform
if (df_features[col].dropna() >= 0).all():
feature_definitions.append({
'name': f'log_{col}',
'formula': f'np.log1p({col})',
'description': f'Log-transformed {col}'
})
else:
# For columns with negative values, use scaled version
feature_definitions.append({
'name': f'scaled_{col}',
'formula': f'({col} - {col}.mean()) / ({col}.std() + 1e-8)',
'description': f'Scaled version of {col}'
})
for feat in feature_definitions:
try:
# Create namespace with numpy, pandas, and dataframe columns
namespace = {'np': np, 'pd': pd}
# Add dataframe columns to namespace for eval
for col in df_features.columns:
namespace[col] = df_features[col]
# Also add Series methods
namespace.update({
'mean': df_features.mean,
'std': df_features.std,
'min': df_features.min,
'max': df_features.max
})
df_features[feat['name']] = eval(feat['formula'], namespace)
print(f"Created feature: {feat['name']} - {feat['description']}")
except Exception as e:
print(f"Failed to create feature {feat['name']}: {e}")
# Optionally create a NaN column if feature creation fails
df_features[feat['name']] = np.nan
return df_features
def scale_features(self, df, scaling_strategy='standard'):
"""
Scale numerical features.
Args:
df (pd.DataFrame): Input DataFrame
scaling_strategy (str): Scaling strategy
Returns:
pd.DataFrame: Scaled DataFrame
"""
df_scaled = df.copy()
numeric_cols = df_scaled.select_dtypes(include=[np.number]).columns
# Remove target column from scaling if it exists
if 'target' in self.config and self.config['target'] in numeric_cols:
numeric_cols = numeric_cols.drop(self.config['target'])
for col in numeric_cols:
if scaling_strategy == 'standard':
scaler = StandardScaler()
df_scaled[col] = scaler.fit_transform(df_scaled[[col]]).ravel()
self.scalers[col] = scaler
elif scaling_strategy == 'minmax':
scaler = MinMaxScaler()
df_scaled[col] = scaler.fit_transform(df_scaled[[col]]).ravel()
self.scalers[col] = scaler
return df_scaled