tsfresh Example
CONTENTS:
- tsfresh Example
- 0) Setup-setup)
- 1) Import Packages-import-packages)
- 3) Load HAR data (train/test splits)-load-har-data-(train/test-splits))
- 4) Convert to long format for tsfresh-convert-to-long-format-for-tsfresh)
- 5) Feature extraction with tsfresh-feature-extraction-with-tsfresh)
- 6) Feature Selection-feature-selection)
- 7) Train a classifier and evaluate-train-a-classifier-and-evaluate)
tsfresh Example¶
This notebook shows an end-to-end workflow using tsfresh on the UCI HAR dataset (smartphone accelerometer & gyroscope data).
The main objective of this notebook is to show tsfresh can be used to create multiple new features through feature engineering to help predict 6 different motions using the accelerometer and gyroscope time series data. The main function of tsfresh is to engineer features. However, it also does have functionalities for certain ML models built-in to perform predictions too.
We will:
- Download & load the dataset
- Reshape to long format suitable for tsfresh (columns:
id,time,kind,value) - Extract features with
tsfresh.extract_features - Select relevant features with
tsfresh.select_features - Train a RandomForest model and predict the 6 different motions and evaluate.
- Inspect top feature importances
# %pip install -q tsfresh scikit-learn pandas numpy matplotlib requests tqdm1) Import Packages¶
from pathlib import Path
import zipfile
import numpy as np
import requests
from tqdm import tqdm
np.random.seed(42)
DATA_URL = "https://archive.ics.uci.edu/ml/machine-learning-databases/00240/UCI%20HAR%20Dataset.zip"
DATA_DIR = Path("./uci_har_data") # ✅ corrected here
def fetch_data(url=DATA_URL, out_dir=DATA_DIR):
out_dir.mkdir(parents=True, exist_ok=True)
zip_path = out_dir / "UCI_HAR_Dataset.zip"
if not zip_path.exists():
resp = requests.get(url, stream=True, timeout=20)
resp.raise_for_status()
total = int(resp.headers.get("content-length", 0))
with open(zip_path, "wb") as f, tqdm(total=total, unit="B", unit_scale=True, desc="Downloading") as pbar:
for chunk in resp.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
pbar.update(len(chunk))
extract_path = out_dir / "UCI HAR Dataset"
if not extract_path.exists():
with zipfile.ZipFile(zip_path, "r") as zf:
zf.extractall(out_dir)
return extract_path
har_root = fetch_data()
print(f"Dataset ready at: {har_root}")Dataset ready at: uci_har_data/UCI HAR Dataset
3) Load HAR data (train/test splits)¶
Here we extract the raw signals captured by the sensors. Each example has 128 time steps for each sensor/channel.
def load_inertial_split(split=“train”, root=None): root = path(root) if root else har_root base = root / split / “Inertial Signals” channels = [ “body_acc_x”, “body_acc_y”, “body_acc_z”, “body_gyro_x”, “body_gyro_y”, “body_gyro_z”, “total_acc_x”, “total_acc_y”, “total_acc_z”, ] arrays = {} for ch in channels: fn = base / f"{ch}{split}.txt" arr = np.loadtxt(fn) arrays[ch] = arr y = np.loadtxt(root / split / f"y{split}.txt").astype(int) subjects = np.loadtxt(root / split / f"subject_{split}.txt").astype(int) return arrays, y, subjects
train_arrays, y_train_raw, subj_train = load_inertial_split(“train”, har_root) test_arrays, y_test_raw, subj_test = load_inertial_split(“test”, har_root)
channels = list(train_arrays.keys()) n_train = train_arrays[channels[0]].shape[0] n_test = test_arrays[channels[0]].shape[0] n_time = train_arrays[channels[0]].shape[1]
print( f"Train: {n_train}, Test: {n_test}, Timesteps: {n_time}, Channels: {len(channels)}" )
4) Convert to long format for tsfresh¶
We build a DataFrame with columns:
id: row id (unique per series example)time: timestep (0..127)kind: sensor name (e.g.,body_acc_x,body_gyro_z, ...)value: observed value
This lets tsfresh handle multivariate time series via the kind column.
def to_long(arrays, start_id=0): n_samples = arrays[next(iter(arrays))].shape[0] n_time = arrays[next(iter(arrays))].shape[1] frames = [] for i in range(n_samples): for ch, mat in arrays.items(): df = pd.DataFrame( { “id”: start_id + i, “time”: np.arange(n_time, dtype=int), “kind”: ch, “value”: mat[i, :].astype(float), } ) frames.append(df) return pd.concat(frames, ignore_index=True)
X_long_train = to_long(train_arrays, start_id=0) X_long_test = to_long(test_arrays, start_id=train_arrays[channels[0]].shape[0])
Build labels aligned with id space y_train_df = pd.DataFrame( {“id”: np.arange(0, train_arrays[channels[0]].shape[0]), “label”: y_train_raw} ) y_test_df = pd.DataFrame( { “id”: np.arange( train_arrays[channels[0]].shape[0], train_arrays[channels[0]].shape[0] + test_arrays[channels[0]].shape[0], ), “label”: y_test_raw, } )
X_long_train.head(), X_long_train.shape, y_train_df.head()
5) Feature extraction with tsfresh¶
There are many different parameter values
that decide the extent of feature engineering.
For our case, we’ll use MinimalFCParameters
that extracts features like mean, standard deviation,
length, max value, min value etc. This is the
best argument to be passed for fast and
lighweight use-cases. For more features,
use the ComprehensiveFCParameters() class
which extracts >800 features.
settings = fe.MinimalFCP()
X_feat_train = ext_feat( X_long_train, column_id=“id”, column_sort=“time”, column_kind=“kind”, column_value=“value”, default_fc_parameters=settings, n_jobs=0, disable_progressbar=False, ) X_feat_train = df_utils.imp(X_feat_train) X_feat_train.shape
Here we transform the test set using the same settings.
X_feat_test = ext_feat( X_long_test, column_id=“id”, column_sort=“time”, column_kind=“kind”, column_value=“value”, default_fc_parameters=settings, n_jobs=0, disable_progressbar=False, ) X_feat_test = df_utils.imp(X_feat_test) X_feat_test.shape
6) Feature Selection¶
Tsfresh has built-in functionality to select the best features out of all the ones artificially created. We select features most relevant to distinguishing the 6 classes/motions. le = label_e() y_train = pd.Series( le.fit_transform(y_train_df.set_index(“id”)[“label”]), index=y_train_df[“id”].values, ) y_test = pd.Series( le.transform(y_test_df.set_index(“id”)[“label”]), index=y_test_df[“id”].values )
X_sel_train = sel_feat(X_feat_train, y_train) X_sel_train.shape
Align the test feature matrix to the selected training columns. X_feat_test = X_feat_test.reindex(columns=X_sel_train.columns, fill_value=0.0) X_feat_test.shape
7) Train a classifier and evaluate¶
We use a RandomForestClassifier and report:
- Accuracy and Macro-F1
- Macro ROC-AUC (via one-vs-rest) clf = RFCLass(n_estimators=500, max_depth=None, n_jobs=-1, random_state=42) clf.fit(X_sel_train, y_train)
proba = clf.predict_proba(X_feat_test) # shape [n_samples, n_classes] pred = np.argmax(proba, axis=1)
acc = metrics.accuracy_score(y_test, pred) f1 = metrics.f1_score(y_test, pred, average=“macro”)
y_test_labels = label_b(y_test, classes=np.arange(len(le.classes_))) roc = metrics.roc_auc_score(y_test, proba, average=“macro”, multi_class=“ovr”)
print(f"Test Accuracy: {acc:.3f}“) print(f"Test Macro-F1: {f1:.3f}”) print(f"Test Macro ROC-AUC: {roc:.3f}“) print( “\nClassification report:\n”, metrics.classification_report( y_test, pred, target_names=[str(c) for c in le.classes_] ), ) One-vs-Rest ROC Curves (per class) plt.figure(figsize=(8, 6)) for k, cls in enumerate(le.classes_): fpr, tpr, _ = metrics.roc_curve(y_test_labels[:, k], proba[:, k]) roc_auc = metrics.auc(fpr, tpr) plt.plot(fpr, tpr, label=f"Class {cls} (AUC={roc_auc:.3f})”) plt.plot([0, 1], [0, 1], “--”) plt.xlabel(“False Positive Rate”) plt.ylabel(“True Positive Rate”) plt.title(“One-vs-Rest ROC on HAR (RandomForest + tsfresh)”) plt.legend(loc=“lower right”) plt.tight_layout() plt.show()
Top Feature Importances plt.figure(figsize=(8, 4)) importances = pd.Series( clf.feature_importances_, index=X_sel_train.columns ).sort_values(ascending=False) topk = 25 ax = importances.iloc[:topk].sort_values().plot(kind=“barh”) ax.set_title(f"Top {topk} Features") ax.set_xlabel(“Importance”) plt.tight_layout() plt.show()