Automated Feature Engineering

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

import featuretools as ft
import pandas as pd
import numpy as np
import time
from collections import defaultdict
 
def measure_dfs_complexity(es, target, depths=[1, 2, 3], 
                           n_primitives=[2, 5, 10]):
    """
    Empirically measure DFS complexity across parameters.
    """
    results = []
    
    all_agg_prims = ["sum", "mean", "std", "max", "min", 
                     "count", "num_unique", "mode", "median", "trend"]
    
    for depth in depths:
        for n_prim in n_primitives:
            if n_prim > len(all_agg_prims):
                continue
                
            agg_prims = all_agg_prims[:n_prim]
            
            start_time = time.time()
            start_mem = get_memory_usage()
            
            feature_matrix, features = ft.dfs(
                entityset=es,
                target_dataframe_name=target,
                agg_primitives=agg_prims,
                trans_primitives=[],
                max_depth=depth,
                verbose=False
            )
            
            elapsed = time.time() - start_time
            mem_used = get_memory_usage() - start_mem
            
            results.append({
                'depth': depth,
                'n_primitives': n_prim,
                'n_features': len(features),
                'time_seconds': elapsed,
                'memory_mb': mem_used / (1024 * 1024),
                'features_per_second': len(features) / elapsed
            })
            
            print(f"Depth {depth}, Prims {n_prim}: "
                  f"{len(features)} features in {elapsed:.2f}s")
    
    return pd.DataFrame(results)
 
def get_memory_usage():
    """Get current memory usage in bytes."""
    import psutil
    process = psutil.Process()
    return process.memory_info().rss
 
# Run complexity analysis
complexity_df = measure_dfs_complexity(es, "customers")
print("
Complexity Analysis Results:")
print(complexity_df)
 
# Fit complexity model
from sklearn.linear_model import LinearRegression
 
# Features: depth, n_primitives, depth^2, depth*n_prims
X_complexity = complexity_df[['depth', 'n_primitives']].copy()
X_complexity['depth_sq'] = X_complexity['depth'] ** 2
X_complexity['depth_prims'] = X_complexity['depth'] * X_complexity['n_primitives']
 
model = LinearRegression()
model.fit(X_complexity, complexity_df['n_features'])
 
print("
Complexity Model Coefficients:")
print(f"  Intercept: {model.intercept_:.2f}")
for name, coef in zip(X_complexity.columns, model.coef_):
    print(f"  {name}: {coef:.2f}")

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

import featuretools as ft
import pandas as pd
import numpy as np
from itertools import islice
 
def chunked_dfs(es, target, cutoff_times, chunk_size=10000, **dfs_kwargs):
    """
    Process DFS in chunks to manage memory.
    
    Args:
        es: EntitySet
        target: Target dataframe name
        cutoff_times: DataFrame with entity_id and time columns
        chunk_size: Number of instances per chunk
        **dfs_kwargs: Additional arguments for dfs()
    
    Yields:
        Feature matrix chunks
    """
    # Get entity ID column
    target_df = es[target]
    id_col = target_df.ww.index
    
    # Sort cutoff times for consistent ordering
    cutoff_times = cutoff_times.sort_values('time').reset_index(drop=True)
    
    # Process in chunks
    n_chunks = (len(cutoff_times) + chunk_size - 1) // chunk_size
    
    for i in range(n_chunks):
        start_idx = i * chunk_size
        end_idx = min((i + 1) * chunk_size, len(cutoff_times))
        chunk_cutoffs = cutoff_times.iloc[start_idx:end_idx]
        
        print(f"Processing chunk {i+1}/{n_chunks} "
              f"({len(chunk_cutoffs)} instances)...")
        
        # Run DFS on chunk
        chunk_features, feature_defs = ft.dfs(
            entityset=es,
            target_dataframe_name=target,
            cutoff_time=chunk_cutoffs,
            **dfs_kwargs
        )
        
        yield chunk_features, feature_defs
        
        # Explicitly free memory
        del chunk_features
        import gc
        gc.collect()
 
# Usage
all_chunks = []
for chunk_fm, features in chunked_dfs(
    es, "customers", cutoff_df,
    chunk_size=5000,
    agg_primitives=["mean", "sum", "count"],
    max_depth=2
):
    # Process or save each chunk
    chunk_fm.to_parquet(f"features_chunk_{len(all_chunks)}.parquet")
    all_chunks.append(f"features_chunk_{len(all_chunks)}.parquet")
 
# Combine chunks if needed
feature_matrix = pd.concat([
    pd.read_parquet(f) for f in all_chunks
])

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

import pandas as pd
import numpy as np
 
def optimize_dtypes(df, verbose=True):
    """
    Reduce DataFrame memory by optimizing data types.
    """
    start_mem = df.memory_usage(deep=True).sum() / (1024**2)
    
    for col in df.columns:
        col_type = df[col].dtype
        
        if col_type == 'float64':
            # Check if can downcast to float32
            if df[col].min() > np.finfo(np.float32).min and \
               df[col].max() < np.finfo(np.float32).max:
                df[col] = df[col].astype(np.float32)
        
        elif col_type == 'int64':
            # Downcast integers
            col_min, col_max = df[col].min(), df[col].max()
            
            if col_min >= 0:
                if col_max < 255:
                    df[col] = df[col].astype(np.uint8)
                elif col_max < 65535:
                    df[col] = df[col].astype(np.uint16)
                elif col_max < 4294967295:
                    df[col] = df[col].astype(np.uint32)
            else:
                if col_min > -128 and col_max < 127:
                    df[col] = df[col].astype(np.int8)
                elif col_min > -32768 and col_max < 32767:
                    df[col] = df[col].astype(np.int16)
                elif col_min > -2147483648 and col_max < 2147483647:
                    df[col] = df[col].astype(np.int32)
        
        elif col_type == 'object':
            # Convert low-cardinality objects to category
            if df[col].nunique() / len(df) < 0.5:
                df[col] = df[col].astype('category')
    
    end_mem = df.memory_usage(deep=True).sum() / (1024**2)
    
    if verbose:
        print(f"Memory: {start_mem:.1f} MB → {end_mem:.1f} MB "
              f"({100 * (1 - end_mem/start_mem):.1f}% reduction)")
    
    return df
 
# Apply to feature matrix
feature_matrix = optimize_dtypes(feature_matrix)

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

import featuretools as ft
import time
 
# Sequential processing
start = time.time()
fm_seq, features = ft.dfs(
    entityset=es,
    target_dataframe_name="customers",
    cutoff_time=cutoff_times,
    agg_primitives=["mean", "sum", "count", "std"],
    max_depth=2,
    n_jobs=1,  # Single core
    verbose=True
)
time_sequential = time.time() - start
 
# Parallel processing
start = time.time()
fm_par, features = ft.dfs(
    entityset=es,
    target_dataframe_name="customers",
    cutoff_time=cutoff_times,
    agg_primitives=["mean", "sum", "count", "std"],
    max_depth=2,
    n_jobs=-1,  # All available cores
    verbose=True
)
time_parallel = time.time() - start
 
print(f"
Sequential: {time_sequential:.2f}s")
print(f"Parallel: {time_parallel:.2f}s")
print(f"Speedup: {time_sequential/time_parallel:.2f}x")
 
# Optimal n_jobs depends on:
# - Number of CPU cores
# - Memory available per process
# - Size of data chunks
 
import os
n_cores = os.cpu_count()
print(f"
Available cores: {n_cores}")
print(f"Recommended n_jobs: {min(n_cores, 8)}")  # Memory can limit scaling

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

import dask.dataframe as dd
import featuretools as ft
from dask.distributed import Client, LocalCluster
 
# Start Dask cluster
cluster = LocalCluster(n_workers=4, threads_per_worker=2)
client = Client(cluster)
print(f"Dashboard: {client.dashboard_link}")
 
# Load data as Dask DataFrames
customers_dask = dd.read_parquet("s3://bucket/customers/*.parquet")
orders_dask = dd.read_parquet("s3://bucket/orders/*.parquet")
items_dask = dd.read_parquet("s3://bucket/order_items/*.parquet")
 
# Create EntitySet with Dask DataFrames
es = ft.EntitySet(id="distributed")
 
# Add Dask dataframes (Featuretools handles distribution)
es = es.add_dataframe(
    dataframe_name="customers",
    dataframe=customers_dask,
    index="customer_id",
    time_index="signup_date"
)
 
es = es.add_dataframe(
    dataframe_name="orders",
    dataframe=orders_dask,
    index="order_id",
    time_index="order_date"
)
 
es = es.add_dataframe(
    dataframe_name="order_items",
    dataframe=items_dask,
    index="item_id"
)
 
# Add relationships
es = es.add_relationship("customers", "customer_id", 
                          "orders", "customer_id")
es = es.add_relationship("orders", "order_id", 
                          "order_items", "order_id")
 
# DFS works with Dask backend
feature_matrix, features = ft.dfs(
    entityset=es,
    target_dataframe_name="customers",
    agg_primitives=["mean", "sum", "count"],
    max_depth=2
)
 
# Result is a Dask DataFrame - compute when needed
print(f"Feature matrix partitions: {feature_matrix.npartitions}")
 
# Trigger computation and save
feature_matrix.to_parquet("s3://bucket/features/", compute=True)

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

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
import featuretools as ft
 
# Initialize Spark
spark = SparkSession.builder \
    .appName("FeatureEngineering") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.shuffle.partitions", "200") \
    .getOrCreate()
 
# Load data
customers_spark = spark.read.parquet("s3://bucket/customers/")
orders_spark = spark.read.parquet("s3://bucket/orders/")
items_spark = spark.read.parquet("s3://bucket/order_items/")
 
# Strategy 1: Use Spark for pre-aggregation, Featuretools for final features
# This reduces data volume before DFS
 
# Pre-aggregate order_items to order level in Spark
order_summaries = items_spark.groupBy("order_id").agg(
    F.count("*").alias("item_count"),
    F.sum("quantity").alias("total_quantity"),
    F.sum(F.col("quantity") * F.col("unit_price")).alias("order_value"),
    F.avg("unit_price").alias("avg_item_price"),
    F.max("unit_price").alias("max_item_price")
)
 
# Join to orders
orders_enriched = orders_spark.join(order_summaries, "order_id", "left")
 
# Convert to pandas for Featuretools (sample if needed)
orders_pd = orders_enriched.toPandas()
customers_pd = customers_spark.toPandas()
 
# Now use standard Featuretools on reduced data
es = ft.EntitySet(id="spark_prepped")
es = es.add_dataframe(dataframe_name="customers", dataframe=customers_pd,
                       index="customer_id")
es = es.add_dataframe(dataframe_name="orders", dataframe=orders_pd,
                       index="order_id")
es = es.add_relationship("customers", "customer_id", "orders", "customer_id")
 
# DFS on pre-aggregated data
feature_matrix, features = ft.dfs(
    entityset=es,
    target_dataframe_name="customers",
    agg_primitives=["mean", "sum", "std", "max", "min", "count"],
    max_depth=1  # Already aggregated one level
)
 
# Strategy 2: Sample for feature discovery, Spark for production
# Use small sample with Featuretools to identify valuable features
# Then implement those features in Spark SQL for production
 
spark_features = orders_enriched.groupBy("customer_id").agg(
    F.mean("order_value").alias("MEAN_orders_order_value"),
    F.sum("order_value").alias("SUM_orders_order_value"),
    F.count("*").alias("COUNT_orders"),
    F.stddev("order_value").alias("STD_orders_order_value")
)

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
105

import numpy as np
import pandas as pd
 
class IncrementalAggregator:
    """
    Maintains incremental aggregation state for features.
    """
    
    def __init__(self):
        self.stats = {}  # entity_id -> {feature -> state}
    
    def _get_state(self, entity_id, feature):
        if entity_id not in self.stats:
            self.stats[entity_id] = {}
        if feature not in self.stats[entity_id]:
            self.stats[entity_id][feature] = {
                'count': 0,
                'sum': 0.0,
                'sum_sq': 0.0,  # For variance
                'min': float('inf'),
                'max': float('-inf')
            }
        return self.stats[entity_id][feature]
    
    def update(self, entity_id, feature, value):
        """
        Incrementally update aggregation with new value.
        """
        if pd.isna(value):
            return
            
        state = self._get_state(entity_id, feature)
        
        state['count'] += 1
        state['sum'] += value
        state['sum_sq'] += value ** 2
        state['min'] = min(state['min'], value)
        state['max'] = max(state['max'], value)
    
    def get_aggregations(self, entity_id, feature):
        """
        Compute all aggregations from current state.
        """
        state = self._get_state(entity_id, feature)
        n = state['count']
        
        if n == 0:
            return {
                'count': 0,
                'sum': np.nan,
                'mean': np.nan,
                'std': np.nan,
                'min': np.nan,
                'max': np.nan
            }
        
        mean = state['sum'] / n
        
        # Welford's algorithm for stable variance
        if n > 1:
            variance = (state['sum_sq'] - state['sum']**2/n) / (n-1)
            std = np.sqrt(max(0, variance))
        else:
            std = 0.0
        
        return {
            'count': n,
            'sum': state['sum'],
            'mean': mean,
            'std': std,
            'min': state['min'] if state['min'] != float('inf') else np.nan,
            'max': state['max'] if state['max'] != float('-inf') else np.nan
        }
 
# Usage example
aggregator = IncrementalAggregator()
 
# Process historical orders
historical_orders = [
    {'customer_id': 1, 'total_amount': 150.0},
    {'customer_id': 1, 'total_amount': 200.0},
    {'customer_id': 1, 'total_amount': 175.0},
]
 
for order in historical_orders:
    aggregator.update(
        order['customer_id'], 
        'orders.total_amount', 
        order['total_amount']
    )
 
print("After 3 orders:")
print(aggregator.get_aggregations(1, 'orders.total_amount'))
 
# New order arrives
new_order = {'customer_id': 1, 'total_amount': 225.0}
aggregator.update(
    new_order['customer_id'],
    'orders.total_amount',
    new_order['total_amount']
)
 
print("
After 4th order:")
print(aggregator.get_aggregations(1, 'orders.total_amount'))

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

import featuretools as ft
import hashlib
import json
import os
 
class FeatureCache:
    """
    Cache computed features and feature definitions.
    """
    
    def __init__(self, cache_dir="./feature_cache"):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)
    
    def _get_cache_key(self, es_id, target, agg_prims, trans_prims, 
                       max_depth, cutoff_hash=None):
        """Generate cache key from DFS parameters."""
        key_dict = {
            'es_id': es_id,
            'target': target,
            'agg_prims': sorted(agg_prims),
            'trans_prims': sorted(trans_prims),
            'max_depth': max_depth,
            'cutoff_hash': cutoff_hash
        }
        key_str = json.dumps(key_dict, sort_keys=True)
        return hashlib.md5(key_str.encode()).hexdigest()[:16]
    
    def get_feature_defs(self, cache_key):
        """Load cached feature definitions."""
        path = os.path.join(self.cache_dir, f"{cache_key}_features.json")
        if os.path.exists(path):
            return ft.load_features(path)
        return None
    
    def save_feature_defs(self, features, cache_key):
        """Save feature definitions to cache."""
        path = os.path.join(self.cache_dir, f"{cache_key}_features.json")
        ft.save_features(features, path)
    
    def get_feature_matrix(self, cache_key, cutoff_times):
        """Load cached feature matrix if valid."""
        path = os.path.join(self.cache_dir, f"{cache_key}_matrix.parquet")
        if os.path.exists(path):
            cached = pd.read_parquet(path)
            # Validate cutoff times match
            if len(cached) == len(cutoff_times):
                return cached
        return None
    
    def save_feature_matrix(self, matrix, cache_key):
        """Save feature matrix to cache."""
        path = os.path.join(self.cache_dir, f"{cache_key}_matrix.parquet")
        matrix.to_parquet(path)
 
# Usage
cache = FeatureCache()
 
cache_key = cache._get_cache_key(
    es.id, "customers", 
    ["mean", "sum", "count"],
    ["month", "year"],
    max_depth=2
)
 
# Check cache first
features = cache.get_feature_defs(cache_key)
 
if features is None:
    # Compute and cache
    feature_matrix, features = ft.dfs(
        entityset=es,
        target_dataframe_name="customers",
        agg_primitives=["mean", "sum", "count"],
        trans_primitives=["month", "year"],
        max_depth=2
    )
    cache.save_feature_defs(features, cache_key)
else:
    # Use cached definitions, compute matrix
    print("Using cached feature definitions")
    feature_matrix = ft.calculate_feature_matrix(
        features=features,
        entityset=es
    )

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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

import featuretools as ft
from datetime import datetime, timedelta
import logging
import time
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
class ProductionFeaturePipeline:
    """
    Production-ready feature engineering pipeline.
    """
    
    def __init__(self, feature_definitions, entityset_builder,
                 max_latency_seconds=30, alert_threshold_ms=5000):
        self.feature_defs = feature_definitions
        self.es_builder = entityset_builder
        self.max_latency = max_latency_seconds
        self.alert_threshold = alert_threshold_ms / 1000
        self.metrics = {
            'total_computations': 0,
            'total_time_seconds': 0,
            'errors': 0,
            'cache_hits': 0
        }
    
    def compute_features(self, entity_ids, cutoff_time=None):
        """
        Compute features for given entities with production safeguards.
        """
        start = time.time()
        
        try:
            # Build EntitySet with only relevant data
            es = self.es_builder(entity_ids)
            
            # Set cutoff time
            if cutoff_time is None:
                cutoff_time = datetime.now()
            
            cutoff_df = pd.DataFrame({
                'entity_id': entity_ids,
                'time': [cutoff_time] * len(entity_ids)
            })
            
            # Compute features
            feature_matrix = ft.calculate_feature_matrix(
                features=self.feature_defs,
                entityset=es,
                cutoff_time=cutoff_df
            )
            
            # Record metrics
            elapsed = time.time() - start
            self.metrics['total_computations'] += 1
            self.metrics['total_time_seconds'] += elapsed
            
            # Alert if slow
            if elapsed > self.alert_threshold:
                logger.warning(
                    f"Slow feature computation: {elapsed:.2f}s "
                    f"for {len(entity_ids)} entities"
                )
            
            # Timeout check
            if elapsed > self.max_latency:
                logger.error(
                    f"Feature computation exceeded max latency: {elapsed:.2f}s"
                )
            
            return feature_matrix
            
        except Exception as e:
            self.metrics['errors'] += 1
            logger.error(f"Feature computation failed: {e}")
            raise
    
    def get_metrics(self):
        """Return pipeline metrics."""
        metrics = self.metrics.copy()
        if metrics['total_computations'] > 0:
            metrics['avg_latency_seconds'] = (
                metrics['total_time_seconds'] / metrics['total_computations']
            )
        return metrics
    
    def health_check(self):
        """Check pipeline health."""
        try:
            # Compute features for a single test entity
            test_features = self.compute_features([1])
            return {
                'status': 'healthy',
                'feature_count': len(test_features.columns),
                'latency_ms': (time.time() - time.time()) * 1000
            }
        except Exception as e:
            return {
                'status': 'unhealthy',
                'error': str(e)
            }
 
# Deploy pipeline
def build_entityset(entity_ids):
    # Load only relevant data for efficiency
    customers = load_customers(entity_ids)
    orders = load_orders_for_customers(entity_ids)
    items = load_items_for_orders(orders['order_id'])
    
    es = ft.EntitySet(id="production")
    es.add_dataframe(dataframe_name="customers", dataframe=customers,
                     index="customer_id")
    es.add_dataframe(dataframe_name="orders", dataframe=orders,
                     index="order_id")
    es.add_dataframe(dataframe_name="items", dataframe=items,
                     index="item_id")
    es.add_relationship("customers", "customer_id", "orders", "customer_id")
    es.add_relationship("orders", "order_id", "items", "order_id")
    return es
 
# Load saved feature definitions
features = ft.load_features("production_features.json")
 
pipeline = ProductionFeaturePipeline(
    feature_definitions=features,
    entityset_builder=build_entityset
)
 
# Compute features for batch scoring
customer_ids = [1001, 1002, 1003, 1004]
feature_matrix = pipeline.compute_features(customer_ids)