In this challenge, you will apply your knowledge of temporal feature engineering to a practical forecasting scenario. Imagine you are working with a daily sales dataset for a retail store, where each row represents the total sales for one day. Your goal is to prepare this data for a forecasting model by creating essential features that capture temporal patterns and seasonality, while carefully avoiding data leakage.

You need to engineer the following types of features:
- Lag features that reference previous sales values;
- Rolling window statistics that summarize recent sales trends;
- Calendar-based features, such as the day of the week, to capture recurring patterns.

It is crucial to ensure that for each date, only information available up to that point in time is used to construct features. Missing values that result from lagging or rolling operations should be handled appropriately so that your model receives clean, informative inputs. This exercise will help reinforce the best practices for building robust, leakage-free time series features.

import unittest
import ast
import re   
import importlib
import pandas as pd
import numpy as np

class TestTask(unittest.TestCase):
    def setUp(self):
        # Відключаємо print, щоб вивід студента не засмічував логи тестів
        def fake_print(*args, **kwargs):
            pass
        self.orig_print = __builtins__["print"]
        __builtins__["print"] = fake_print
        
        self.data = {
            "date": pd.date_range(start="2024-01-01", periods=10, freq="D"),
            "sales": [100, 120, 130, 150, 170, 160, 180, 200, 210, 220]
        }
        self.df = pd.DataFrame(self.data)
        
        self.module = importlib.import_module("user_code")
        importlib.reload(self.module)

    def tearDown(self):
        __builtins__["print"] = self.orig_print

    def test_1_lag_features(self):
        func = getattr(self.module, "engineer_features", None)
        if not func:
            _dynamic_test(self, False, "Lag features are computed correctly.", "Function engineer_features is missing.")
            return

        try:
            result = func(self.df.copy())
            expected_lag_1 = pd.Series([-1, 100, 120, 130, 150, 170, 160, 180, 200, 210], dtype=float)
            expected_lag_7 = pd.Series([-1, -1, -1, -1, -1, -1, -1, 100, 120, 130], dtype=float)

            is_lag1_correct = np.allclose(result['sales_lag_1'], expected_lag_1)
            is_lag7_correct = np.allclose(result['sales_lag_7'], expected_lag_7)

            _dynamic_test(
                self,
                is_lag1_correct and is_lag7_correct,
                "Lag features are computed correctly and missing values are filled with -1.",
                "Lag features do not match expected values. Did you use .shift() and .fillna(-1)?"
            )
        except Exception as e:
            _dynamic_test(self, False, "Lag features are computed correctly.", f"Function raised an error: {e}")

    def test_2_rolling_features(self):
        func = getattr(self.module, "engineer_features", None)
        try:
            result = func(self.df.copy())
            
            # Динамічно розраховуємо очікувані значення за правильним алгоритмом
            sales_shifted = self.df['sales'].shift(1)
            expected_mean = sales_shifted.rolling(window=7, min_periods=1).mean().fillna(-1)
            expected_std = sales_shifted.rolling(window=7, min_periods=1).std().fillna(-1)

            is_mean_correct = np.allclose(result['sales_rolling_mean_7'], expected_mean)
            is_std_correct = np.allclose(result['sales_rolling_std_7'], expected_std)

            _dynamic_test(
                self,
                is_mean_correct and is_std_correct,
                "Rolling mean and std features are computed correctly using only past values.",
                "Rolling features do not match expected values. Did you calculate them on the shifted sales column?"
            )
        except Exception as e:
            _dynamic_test(self, False, "Rolling mean and std features are computed correctly.", f"Function raised an error: {e}")

    def test_3_day_of_week(self):
        func = getattr(self.module, "engineer_features", None)
        try:
            result = func(self.df.copy())
            expected_dow = self.df['date'].dt.dayofweek
            
            is_dow_correct = (result['day_of_week'] == expected_dow).all()
            
            _dynamic_test(
                self,
                is_dow_correct,
                "Day of week feature is extracted correctly.",
                "Day of week feature is incorrect. Did you use dt.dayofweek?"
            )
        except Exception as e:
            _dynamic_test(self, False, "Day of week feature is extracted correctly.", f"Function raised an error: {e}")


def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\s{2,}", " ", text)
    text = re.sub(r"\s*([,:?])\s*", r"\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    
    assign_nodes = [
        (i, node)
        for i, node in enumerate(tree.body)
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets)
    ]

    if not assign_nodes:
        return code

    for i, node in reversed(assign_nodes):
        start_line = node.lineno - 1
        line = lines[start_line]
        indent = ' ' * (len(line) - len(line.lstrip()))
        lines[start_line] = f"{indent}{var_name} = {value}"
        next_line = len(lines)
        for next_node in tree.body[i+1:]:
            if hasattr(next_node, 'lineno'):
                next_line = next_node.lineno - 1
                break
        if next_line > start_line + 1:
            lines[start_line+1:next_line] = []
        changed = True

    return '\\n'.join(lines) if changed else code

if __name__ == "__main__":
    unittest.main()

test_main.py

Learn to transform raw time series data into powerful machine learning features. This course covers the creation of lag variables, rolling window statistics, and the design of validation splits that prevent data leakage, ensuring robust model evaluation.

Explore advanced techniques for engineering features from time series data, focusing on lag variables, rolling statistics, and robust validation strategies.

Challenge: Engineer Features for a Forecasting Task

Lösung