Summary  
This chapter demonstrates how to use pandas’ resample method to convert a datetime‐indexed series to a different frequency and apply aggregation functions like mean.  

General domain of usage  
Time series data analysis

import pandas as pd

# Create a simple daily time series
dates = pd.date_range("2024-01-01", periods=90, freq="D")
values = pd.Series(range(90), index=dates)

# Resample to monthly frequency, taking the average for each month
monthly_avg = values.resample("ME").mean()
print(monthly_avg)

When working with time series data in pandas, the `resample` method allows you to convert your data from one frequency to another, such as from daily to monthly or weekly. This is useful for summarizing, aggregating, or aligning your data to a different time scale.

The `resample` method works similarly to `groupby`, but it is specifically designed for time series data with a `DatetimeIndex`. You specify a new frequency using a frequency string, such as `"ME"` for month-end, `"W"` for week-end, or `"D"` for daily. After resampling, you apply an aggregation function, such as `.mean()`, `.sum()`, `.max()`, or `.min()`, to summarize the values within each new period.

In the code sample above, the daily series is resampled to monthly frequency using `"ME"`, and the `.mean()` function calculates the average value for each month. You can use other frequency strings to resample to different periods, and choose the aggregation function that best fits your analysis needs.


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import pandas as pd
import importlib
import sys
import types

class TestTask(unittest.TestCase):
    def setUp(self):
        # Ensure user_code is fresh for each test
        if "user_code" in sys.modules:
            del sys.modules["user_code"]

    def test_returns_series_with_weekly_frequency(self):
        import user_code
        importlib.reload(user_code)
        dates = pd.date_range("2024-01-01", periods=14, freq="D")
        values = pd.Series(range(1, 15), index=dates)
        result = user_code.resample_to_weekly_sum(values)
        _dynamic_test(
            self,
            isinstance(result, pd.Series) and hasattr(result.index, 'freqstr') or hasattr(result.index, 'freq'),
            "Returned object is a pandas Series with a DatetimeIndex",
            f"Expected a pandas Series with DatetimeIndex, got {type(result)}"
        )
        # Check frequency string
        freq_str = getattr(result.index, 'freqstr', None)
        _dynamic_test(
            self,
            freq_str == 'W' or result.index.freq is not None and result.index.freq.name == 'W-SUN',
            "Resulting index has weekly ('W') frequency",
            f"Expected weekly frequency, got {freq_str}"
        )

    def test_weekly_sum_values(self):
        import user_code
        importlib.reload(user_code)
        # 2024-01-01 (Mon) to 2024-01-14 (Sun)
        dates = pd.date_range("2024-01-01", periods=14, freq="D")
        values = pd.Series(range(1, 15), index=dates)
        result = user_code.resample_to_weekly_sum(values)
        expected = pd.Series([
            sum(range(1, 7)),   # Jan 1-6 (Mon-Sat)
            sum(range(7, 14)),  # Jan 7-13 (Mon-Sun)
            14                  # Jan 14 (Mon)
        ], index=[
            pd.Timestamp("2024-01-06"),
            pd.Timestamp("2024-01-13"),
            pd.Timestamp("2024-01-14")
        ])
        # Since resample('W') sums on Sunday, first week is Jan 1-6 (Mon-Sat, 6 days), second week Jan 7-13 (Mon-Sun, 7 days), third week Jan 14 (Mon)
        # But pandas' resample('W') by default sums up to Sunday, so Jan 1-7, Jan 8-14
        expected = pd.Series([
            sum(range(1,8)),   # Jan 1-7
            sum(range(8,15)),  # Jan 8-14
        ], index=[
            pd.Timestamp("2024-01-07"),
            pd.Timestamp("2024-01-14")
        ])
        _dynamic_test(
            self,
            result is not None and all(result.values == expected.values) and all(result.index == expected.index),
            "Weekly sums are correctly computed for each week",
            f"Expected:\\n{expected}\\nGot:\\n{result}"
        )

    def test_less_than_one_week(self):
        import user_code
        importlib.reload(user_code)
        dates = pd.date_range("2024-01-01", periods=3, freq="D")
        values = pd.Series([2, 4, 6], index=dates)
        result = user_code.resample_to_weekly_sum(values)
        expected = pd.Series([12], index=[pd.Timestamp("2024-01-07")])
        _dynamic_test(
            self,
            result is not None and all(result.values == expected.values) and all(result.index == expected.index),
            "Handles less than one week of data correctly",
            f"Expected: {expected}, Got: {result}"
        )

    def test_multiple_weeks(self):
        import user_code
        importlib.reload(user_code)
        dates = pd.date_range("2024-01-01", periods=21, freq="D")
        values = pd.Series(range(1, 22), index=dates)
        result = user_code.resample_to_weekly_sum(values)
        expected = pd.Series([
            sum(range(1,8)),   # Jan 1-7
            sum(range(8,15)),  # Jan 8-14
            sum(range(15,22)), # Jan 15-21
        ], index=[
            pd.Timestamp("2024-01-07"),
            pd.Timestamp("2024-01-14"),
            pd.Timestamp("2024-01-21"),
        ])
        _dynamic_test(
            self,
            result is not None and all(result.values == expected.values) and all(result.index == expected.index),
            "Correctly computes weekly sums for multiple weeks",
            f"Expected: {expected}, Got: {result}"
        )

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
    # Collect all assignment nodes to modify
    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 nothing to change, return unmodified code
    if not assign_nodes:
        return code

    # Perform replacements for all matching assignments (from last to first to not break line offsets)
    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

Establish foundational skills by manipulating datetime objects, indexing temporal data, utilizing interpolation techniques to impute missing anomalies, ultimately generating chronological visualizations.

Comprehensive coverage of time series fundamentals, from understanding data structure to visualization and handling missing values.

Resampling and Frequency Conversion

Рішення