In this chapter, you will apply your understanding of **rolling statistics** and **time series decomposition** to a new dataset. You will use Python's `pandas` and `matplotlib` libraries to calculate a **rolling mean**, visualize seasonality using a **heatmap**, and decompose the dataset into its **trend**, **seasonal**, and **residual** components. By analyzing the resulting plots and components, you will deepen your ability to interpret the structure of time series data and extract meaningful insights.

import unittest
import ast
import re   
import importlib
import pandas as pd
import numpy as np
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
from statsmodels.tsa.seasonal import seasonal_decompose

class TestTask(unittest.TestCase):
    def setUp(self):
        # Mute plt.show to prevent blocking
        def fake_show(*args, **kwargs):
            pass
        self.orig_show = plt.show
        plt.show = fake_show
        
        self.module = importlib.import_module("user_code")
        importlib.reload(self.module)
        
        # Generate a sample DataFrame
        dates = pd.date_range(start='2021-01-01', periods=24, freq='M')
        np.random.seed(42)
        values = 20 + np.sin(2 * np.pi * dates.month / 12) * 3 + np.random.normal(0, 0.5, size=len(dates))
        self.df = pd.DataFrame({'date': dates, 'value': values})
        self.date_col = 'date'
        self.value_col = 'value'
        self.window_size = 5
        self.freq = 12

    def tearDown(self):
        plt.show = self.orig_show
        plt.close('all')

    def test_1_returns_correct_types(self):
        try:
            result = self.module.analyze_time_series(self.df, self.date_col, self.value_col, self.window_size, self.freq)
            
            is_valid_tuple = isinstance(result, tuple) and len(result) == 4
            all_series = is_valid_tuple and all(isinstance(x, pd.Series) for x in result)
            
            _dynamic_test(
                self,
                all_series,
                "Function returns a tuple of 4 pandas Series.",
                "Expected function to return 4 pandas Series (rolling_mean, trend, seasonal, resid)."
            )
        except Exception as e:
            _dynamic_test(self, False, "Function returns a tuple of 4 pandas Series.", f"Function raised an error: {e}")

    def test_2_rolling_mean_values(self):
        try:
            result = self.module.analyze_time_series(self.df, self.date_col, self.value_col, self.window_size, self.freq)
            rolling_mean = result[0]
            
            expected = self.df.set_index(self.date_col)[self.value_col].rolling(window=self.window_size, center=True).mean()
            
            # Use np.allclose for safe float comparison, filling NaNs with 0 temporarily for the check
            is_correct = np.allclose(rolling_mean.fillna(0), expected.fillna(0), atol=1e-5)
            
            _dynamic_test(
                self,
                is_correct,
                "Rolling mean is calculated correctly with the specified window and center.",
                "Rolling mean values do not match the expected output. Did you use center=True?"
            )
        except Exception as e:
            _dynamic_test(self, False, "Rolling mean is calculated correctly.", f"Function raised an error: {e}")

    def test_3_decomposition_values(self):
        try:
            result = self.module.analyze_time_series(self.df, self.date_col, self.value_col, self.window_size, self.freq)
            trend, seasonal, resid = result[1], result[2], result[3]
            
            # Create reference decomposition
            series = self.df.set_index(self.date_col)[self.value_col]
            reference = seasonal_decompose(series, model='additive', period=self.freq)
            
            # Check seasonal component as a proxy for correct decomposition
            is_correct = np.allclose(seasonal.fillna(0), reference.seasonal.fillna(0), atol=1e-5)
            
            _dynamic_test(
                self,
                is_correct,
                "Time series is correctly decomposed using an additive model.",
                "Decomposition components do not match the expected output. Check the model and period parameters."
            )
        except Exception as e:
             _dynamic_test(self, False, "Time series is correctly decomposed.", f"Function raised an error: {e}")

    def test_4_heatmap_is_plotted(self):
        try:
            plt.clf()
            self.module.analyze_time_series(self.df, self.date_col, self.value_col, self.window_size, self.freq)
            
            # Ensure at least one heatmap (AxesImage) was generated across all figures
            has_image = any(len(ax.images) > 0 for fig in plt.get_fignums() for ax in plt.figure(fig).axes)
            
            _dynamic_test(
                self,
                has_image,
                "Heatmap plot (imshow) is generated successfully.",
                "No heatmap image found. Did you correctly use pivot_table and plt.imshow?"
            )
        except Exception as e:
            _dynamic_test(self, False, "Heatmap plot is generated successfully.", 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

Isolate underlying signals within sequential data by analyzing rolling statistics, generating seasonal heatmaps, extracting distinct trend components to understand complex cyclical behaviors.

Explore techniques to break down and visualize temporal patterns in time series data, including rolling statistics, seasonal heatmaps, decomposition, and advanced analysis.

Challenge: Apply Decomposition

解答