<jupyter_start><jupyter_text>Best Practices We'll put into practice what we've learned so far by working with some real-world data. Let's begin with loading a log of sports activities of a single person:<jupyter_code>import pandas as pd
activity_df = pd.read_json('data/activities.json', convert_dates=['start_date_local'])<jupyter_output><empty_output><jupyter_text>Some Simple and Useful DataFrame Methods<jupyter_code>activity_df.describe()<jupyter_output><empty_output><jupyter_text>__Question__: What does this overview tell us at a first glance?<jupyter_code>activity_df.head()
activity_df.tail(3)
activity_df.sample(10)
(
    activity_df
    .sort_values('start_date_local', ascending=False)
    .head(10)
)
activity_df['type']
activity_df['type'].unique()
activity_df['type'].value_counts()
len(activity_df)<jupyter_output><empty_output><jupyter_text>__Question__: in evaluation of the above expression, is any `DataFrame` method called?  Separation of Concerns Assume we want to create a process that reads such activities from file, selects all the running activities, sorts these by date, and saves the result in another file. A possible implementation may look like this:<jupyter_code>def process_activities():
    activity_df = pd.read_json('data/activities.json', convert_dates=['start_date_local'])
    runs_sorted_by_date_df = (
        activity_df
        .loc[lambda df: df['type'] == 'Run']
        .sort_values('start_date_local')
    )
    runs_sorted_by_date_df.to_csv('data/processed_activities.csv')
process_activities()<jupyter_output><empty_output><jupyter_text>**Discussion**:- What if there are many of these processes, and the data source changes from .json file to database?- How can we test the business logic without access to a file system or database?- How can this be improved?<jupyter_code># !rm data/processed_activities.csv<jupyter_output><empty_output><jupyter_text>As soon as our work shifts from pure exploratory analysis in a notebook to repeatable and robust production systems, separation of concerns becomes important. As discussed above, any code that combines business logic with side effects such as reading/writing data, handling user input from a GUI, etc, can easily break when there are any changes in infrastructure or user interaction.The example in the cells below shows how our naive process implementation is improved by abstracting the infrastructure and separating the business logic from the process.<jupyter_code>from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Callable
# Infrastructure Layer
class ActivitySource(ABC):
    @abstractmethod
    def load(self):
        pass

@dataclass
class JsonLocalActivitySource(ActivitySource):
    filename: str = 'data/activities.json'  # will usually come from a config file
    
    def load(self):
        return pd.read_json(self.filename, convert_dates=['start_date_local'])

@dataclass
class PostgressActivitySource(ActivitySource):
    db_connection: Any  # will usually come from a config file
    
    def load(self):
        # pseudo-code
        return db_connection.query('SELECT * FROM activities')

class ActivitySink(ABC):
    @abstractmethod
    def save(self, activity_df: pd.DataFrame):
        pass

@dataclass
class CsvLocalActivitySink(ActivitySink):
    filename: str = 'data/processed_activities.csv'  # will usually come from a config file
    
    def save(self, activity_df: pd.DataFrame):
        activity_df.to_csv(self.filename)
# Business Layer
def select_runs_and_sort_by_date(activity_df: pd.DataFrame) -> pd.DataFrame:
    return (
        activity_df
        .loc[lambda df: df['type'] == 'Run']
        .sort_values('start_date_local')
    )
# Process Layer
def process_activities(
    source: ActivitySource,
    processing_fn: Callable[[pd.DataFrame], pd.DataFrame],
    sink: ActivitySink
) -> None:
    sink.save(processing_fn(source.load()))
process_activities(JsonLocalActivitySource(), select_runs_and_sort_by_date, CsvLocalActivitySink())<jupyter_output><empty_output><jupyter_text>Hint: for large/complex workflows, consider tools such as [Apache Airflow](https://airflow.apache.org/)  Decluttering We could see before (using the `unique()` and `value_counts()` methods) that there were several activity types in our data. If we want to know the average speed per activity type in the year 2018, we could proceed as follows:<jupyter_code>activity_df['year'] = activity_df['start_date_local'].dt.year
activity_df.head()<jupyter_output><empty_output><jupyter_text>Please note:1. the usage of the [`.dt`](https://pandas.pydata.org/pandas-docs/stable/user_guide/basics.htmldt-accessor) _accessor_ to retrieve the year from our `start_date_local` column ([here](https://pandas.pydata.org/pandas-docs/stable/user_guide/timeseries.htmltime-date-components) is a complete list of properties that can be retrieved from timestamps)2. the usage of `activity_df[COLUMN_NAME] = ...` to assign a new column to the DataFrame instead of calling the `assign()` method. What is the main difference between the two forms of assignment? Which is preferable? Having retrieved the year, we can now select all activities in 2018, and group these by activity type:<jupyter_code>activities_2018_df = activity_df.loc[activity_df['year'] == 2018]
activities_2018_by_type = activities_2018_df.groupby('type')<jupyter_output><empty_output><jupyter_text>Finally, we compute the mean of the `velocity_mean` column per group and show the results:<jupyter_code>mean_velocity_by_type = activities_2018_by_type['velocity_mean'].mean()
mean_velocity_by_type<jupyter_output><empty_output><jupyter_text>In the above approach, we assigned all intermediate results to variables, while also mutating our source data by adding the `year` column. While storing these provisional results may assist with some rare cases of debugging, we usually don't need them. If they keep hanging around in our notebook we may incidentally refer to them later, causing bugs and confusion. Also, by mutating the source data, any additional analysis that we make may not work anymore (in isolation) when re-loading the data from disk.To start with a clean slate, let's reload our activity data:<jupyter_code>activity_df = pd.read_json('data/activities.json', convert_dates=['start_date_local'])<jupyter_output><empty_output><jupyter_text>__Exercise__: Rewrite the same analysis as above without using any temporary variables. The solution should be a single expression that evaluates to the exact same value as the `mean_velocity_by_type` variable above._Bonus_: Try to write the analysis in such a way that it doesn't need any assignment of a `year` column._Hint_: What is being returned by `loc[]` or `assign()`? You may want to refer back to the last example of method chaining (in the notebook about Object-Oriented Programming)<jupyter_code># Your solution:
# %load solutions/decluttering_by_method_chaining.py<jupyter_output><empty_output><jupyter_text>Readable, Testable Pipelines As analyses get larger, one large chain of methods may become hard to write (and even harder to read!). The solution is to write functions that perform understandable single-purpose analysis steps. This makes it easy to test each step of an analysis, and to reuse steps in other analyses.<jupyter_code>def is_non_warmup(activity_df):
    return (
        (activity_df['elapsed_time'] > 600) |
        (activity_df['heartrate_mean'] > 150)
    )

def select_non_warmup_runs(activity_df):
    return (
        activity_df
        .loc[is_non_warmup]
        .loc[lambda df: df['type'] == 'Run']
    )

def to_z_score(series):
    # What if series has a lenght of 1?
    return (series - series.mean()) / series.std()

def add_z_score(activity_df, column):
    return activity_df.assign(**{f'{column}_z': lambda df: to_z_score(df[column])})

def show_best_n_years(activity_df, metrics=['velocity_mean'], n_years=1):
    return (
        activity_df
        .assign(year=lambda df: df['start_date_local'].dt.year)
        .groupby('year')[metrics].mean()
        .sort_values(metrics, ascending=False)
        .head(n_years)
    )<jupyter_output><empty_output><jupyter_text>Without even looking at the exact implementation of the functions in the cell above, using `pipe()` we can clearly express what is happening at a high level in a fairly complicated analysis:<jupyter_code>(
    activity_df
    .pipe(select_non_warmup_runs)
    .pipe(add_z_score, column='velocity_mean')
    .pipe(show_best_n_years, metrics=['velocity_mean_z', 'velocity_mean'], n_years=5)
)<jupyter_output><empty_output><jupyter_text>__Discussion__: Is this a reliable analysis? Which question does it answer? What can be improved?  [Don't Repeat Yourself](https://en.wikipedia.org/wiki/Don%27t_repeat_yourself)<jupyter_code>import logging
import time
from functools import wraps
logging.basicConfig(level=logging.DEBUG)  # You may need to restart the notebook to make this work
logger = logging.getLogger(__name__)
logger.warning(f'There are {len(activity_df)} activities in the DataFrame')
logger.debug(f'If this line is not visible, restart the notebook kernel')<jupyter_output><empty_output><jupyter_text>__Exercise__: Create a decorator that logs the number of rows and columns of the DataFrame that is passed to the pipeline functions above. You can use the `@skip_None` and `@check_value` decorators of the previous module as examples.__Bonus Exercise__: Extend this decorator such that it also logs the same information about the DataFrame that is returned by the pipeline function.__Bonus Exercise__: Extend this decorator such that it also logs how many rows are missing for each column. The [`isnull()`](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.isnull.html) method and `sum()` built-in function may be helpful for this purpose.__Bonus Exercise__: Extend this decorator such that it also logs how much time it took to call the decorated pipeline function.<jupyter_code># Your solution:
# %load solutions/pipe_logging.py<jupyter_output><empty_output><jupyter_text>Preventing Joins Without going into details of joining data in Pandas (see the documentation for [`merge()`](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.merge.html) and [`join()`](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.join.html)), one often-used (anti-) pattern is to use `groupby()` for computing group-wise aggregates such as the mean, and then using a join to compare individual observations with the group aggregate. For example:<jupyter_code>(
    activity_df
    .groupby('type')[['velocity_mean']].mean()
    .reset_index()
    .merge(activity_df, on='type', suffixes=('_of_type',None))
    .assign(velocity_mean_diff=lambda df: df['velocity_mean'] - df['velocity_mean_of_type'])
    .loc[:, ['type', 'name', 'start_date_local', 'velocity_mean_diff', 'velocity_mean']]
)<jupyter_output><empty_output><jupyter_text>We can get the same result (although in a different order because of the lack of a join) by using the groupby's [`transform()`](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.core.groupby.DataFrameGroupBy.transform.html) method. Instead of returning the aggregate result per group, `transform()` returns that result for _every observation_ in the group.<jupyter_code>(
    activity_df
    .groupby('type')[['velocity_mean']].transform('mean')
)<jupyter_output><empty_output><jupyter_text>Because the index of the result of `transform()` still corresponds to our source DataFrame, we can simply assign it, thus preventing any (potentially expensive) join/merge and making the analysis more concise and readable:<jupyter_code>(
    activity_df
    .assign(
        velocity_mean_of_type=lambda df: df.groupby('type')[['velocity_mean']].transform('mean'),
        velocity_mean_diff=lambda df: df['velocity_mean'] - df['velocity_mean_of_type']
    )
    .loc[:, ['type', 'name', 'start_date_local', 'velocity_mean_diff', 'velocity_mean']]
)<jupyter_output><empty_output><jupyter_text>This more elaborate example shows how we can also pass custom transformation functions to `transform()`:<jupyter_code>(
    activity_df
    .loc[lambda df: df['start_date_local'].dt.year == 2018]
    .assign(v_mean_diff=lambda df: (
        df.groupby('type')['velocity_mean']
        .transform(lambda type_v: type_v - type_v.mean())))
    .sample(10)
)<jupyter_output><empty_output><jupyter_text>Bonus Exercises What is the longest run per year?<jupyter_code># Your solution:
# %load solutions/longest_run_per_year.py<jupyter_output><empty_output><jupyter_text>What is the average (mean) number of hours spent cycling by weekday? It is ok to ignore the days without activity (i.e. these don't need to be counted as 0 for the mean).<jupyter_code># Your solution:
# %load solutions/hours_cycling_by_weekday.py<jupyter_output><empty_output><jupyter_text>What are the 10 longest time gaps between activities? The method [`diff()`](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.Series.diff.html) may be useful for this purpose.<jupyter_code># Your solution:
# %load solutions/longest_time_gaps.py<jupyter_output><empty_output><jupyter_text>What are the 7-day periods with the highest total amount of activity (in hours)? Similar to `groupby()`, Pandas has a method [`rolling()`](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.rolling.html) for using an (aggregating) sliding window. It can even use a flexible time-based window size, given that the DataFrame index is datetime-like (and sorted).<jupyter_code>(
    pd.DataFrame({'a': range(5), 'b': range(5, 10)})
    .rolling(3).mean()
)
(
    pd.DataFrame({'a': range(5), 'b': range(5, 10)})
    .rolling(3, min_periods=1).mean()
)
# Your solution:
# %load solutions/total_time_7days.py<jupyter_output><empty_output><jupyter_text>Observe the 20 or 30 runs with highest mean heartrate. Looking at the name and distance, do you see some common pattern? Looking at other columns such as velocity_mean, do you see any activities that are out of the ordinary? Can it be that there are outliers based on measurement errors?Try to create an analysis that detects runs with an uncommon combination of mean heartrate and mean velocity, for example, by comparing z values. If you're familiar with Scikit-Learn, consider using one of its outlier detection methods or fitting a (linear) regression and looking for large residuals.<jupyter_code>activity_df.sort_values('heartrate_mean', ascending=False).head(20)
# Your solution:<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09541384731050509, 'fraction_numerical': 0.005057754083794682, 'mean_word_length': 5.579136690647482, 'pattern_counts': {'":': 0, '<': 97, '<?xml version=': 0, '>': 101, 'https://': 10, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '24', 'dir': 'jupyter-structured-clean-dedup', 'id': '111959', 'license': 'MIT', 'n_tokens_mistral': 4470, 'n_tokens_neox': 4280, 'n_words': 1408, 'path': '04_best_practices.ipynb', 'repo_name': 'jsamoocha/python-for-datascience', 'size': '14631'}