Pythonデータ加工での自動テスト: pandas & pytest

# Pythonデータ加工での自動テスト: pandas & pytest

## やること

- pandasによるデータ加工をpytestで自動テストする

## 対象読者

- Python, pandasでのデータ加工を普段やってる人
- データ加工処理の品質を向上させたい人

## 背景

- データ分析の現場ではテストがほとんどなくて、処理がちゃんと正しく動いているのか自信を持てなかった。
- SIerの現場でテストをした際、「これ、人間がやる必要ある…？」と感じた。
- 本を読むと、世のプログラマーは「テストコード」なるものを書き、「自動テスト」をしているそうではないか。
- Pythonではどのように自動テストをするのか...?__pytestだ!!__

## お断り

- とりあえずpytestを導入するということをテーマとした記事です。あんまりスマートなコードではないです。
- そもそも筆者は探り探りでやっているので、ベストプラクティスはわかっていないです。お気づきの点がございましたら教えてください。


## トピック

- ファイル形式
- 関数として記載する
- pytestによる自動テスト

## 実行環境

- Ubuntu 20.04.5 LTS(WSL2)
- Python 3.9.5
- pytest 7.2.1
- pandas 1.5.3
- jupyter 1.0.0



## 処理概要

以下の架空のデータテーブルを加工する。

### 使用テーブル

トランザクション
|カラム名|説明|
|---|---|
|date|販売した日付|
|product_code|商品コード|
|quantity|販売個数|


商品マスタ
|カラム名|説明|
|---|---|
|product_code|商品コード|
|product_name|商品名|
|unit_price|商品単価|


### 処理内容
- データ読み込み(csv)
- トランザクションテーブルの期間を絞る今回は2023年2月1日~2023年2月28日のデータに絞る。
- 商品idごとで販売戸数合計を算出
- トランザクションテーブルへマスタテーブルを結合
- 単価 * 個数で売り上げ金額を産出
- データ出力(csv)

## コード

### Before

よくある感じの.ipynb形式

```python
import pandas as pd
```


```python
START_DATE ="2023-02-01"
END_DATE = "2023-02-28"
TRANSACTION_DATA_PATH = "./input_data_transaction.csv"
MASTER_DATA_PATH = "./input_data_master.csv"
OUTPUT_DATA_PATH = "./output_data.csv"
```


```python
df_tran = pd.read_csv(TRANSACTION_DATA_PATH)
df_tran
```




<div>
<style scoped>
    .dataframe tbody tr th:only-of-type {
        vertical-align: middle;
    }

    .dataframe tbody tr th {
        vertical-align: top;
    }

    .dataframe thead th {
        text-align: right;
    }
</style>
<table border="1" class="dataframe">
  <thead>
    <tr style="text-align: right;">
      <th></th>
      <th>date</th>
      <th>product_code</th>
      <th>quantity</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <th>0</th>
      <td>2022-12-21</td>
      <td>p_002</td>
      <td>3</td>
    </tr>
    <tr>
      <th>1</th>
      <td>2022-12-21</td>
      <td>p_003</td>
      <td>3</td>
    </tr>
    <tr>
      <th>2</th>
      <td>2022-12-22</td>
      <td>p_002</td>
      <td>10</td>
    </tr>
    <tr>
      <th>3</th>
      <td>2023-01-10</td>
      <td>p_001</td>
      <td>10</td>
    </tr>
    <tr>
      <th>4</th>
      <td>2023-01-20</td>
      <td>p_001</td>
      <td>10</td>
    </tr>
    <tr>
      <th>5</th>
      <td>2023-02-09</td>
      <td>p_002</td>
      <td>3</td>
    </tr>
    <tr>
      <th>6</th>
      <td>2023-02-08</td>
      <td>p_003</td>
      <td>5</td>
    </tr>
    <tr>
      <th>7</th>
      <td>2023-02-08</td>
      <td>p_001</td>
      <td>10</td>
    </tr>
    <tr>
      <th>8</th>
      <td>2023-02-09</td>
      <td>p_001</td>
      <td>10</td>
    </tr>
    <tr>
      <th>9</th>
      <td>2023-02-09</td>
      <td>p_002</td>
      <td>3</td>
    </tr>
    <tr>
      <th>10</th>
      <td>2023-03-01</td>
      <td>p_002</td>
      <td>3</td>
    </tr>
    <tr>
      <th>11</th>
      <td>2023-03-01</td>
      <td>p_002</td>
      <td>3</td>
    </tr>
    <tr>
      <th>12</th>
      <td>2023-03-02</td>
      <td>p_002</td>
      <td>10</td>
    </tr>
  </tbody>
</table>
</div>




```python
df_mst = pd.read_csv(MASTER_DATA_PATH)
df_mst
```




<div>
<style scoped>
    .dataframe tbody tr th:only-of-type {
        vertical-align: middle;
    }

    .dataframe tbody tr th {
        vertical-align: top;
    }

    .dataframe thead th {
        text-align: right;
    }
</style>
<table border="1" class="dataframe">
  <thead>
    <tr style="text-align: right;">
      <th></th>
      <th>product_code</th>
      <th>product_name</th>
      <th>unit_price</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <th>0</th>
      <td>p_001</td>
      <td>pencil</td>
      <td>50</td>
    </tr>
    <tr>
      <th>1</th>
      <td>p_002</td>
      <td>notebook</td>
      <td>100</td>
    </tr>
    <tr>
      <th>2</th>
      <td>p_003</td>
      <td>eraser</td>
      <td>80</td>
    </tr>
  </tbody>
</table>
</div>




```python
df_tran = df_tran[df_tran["date"].between(START_DATE, END_DATE)]
df_tran
```




<div>
<style scoped>
    .dataframe tbody tr th:only-of-type {
        vertical-align: middle;
    }

    .dataframe tbody tr th {
        vertical-align: top;
    }

    .dataframe thead th {
        text-align: right;
    }
</style>
<table border="1" class="dataframe">
  <thead>
    <tr style="text-align: right;">
      <th></th>
      <th>date</th>
      <th>product_code</th>
      <th>quantity</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <th>5</th>
      <td>2023-02-09</td>
      <td>p_002</td>
      <td>3</td>
    </tr>
    <tr>
      <th>6</th>
      <td>2023-02-08</td>
      <td>p_003</td>
      <td>5</td>
    </tr>
    <tr>
      <th>7</th>
      <td>2023-02-08</td>
      <td>p_001</td>
      <td>10</td>
    </tr>
    <tr>
      <th>8</th>
      <td>2023-02-09</td>
      <td>p_001</td>
      <td>10</td>
    </tr>
    <tr>
      <th>9</th>
      <td>2023-02-09</td>
      <td>p_002</td>
      <td>3</td>
    </tr>
  </tbody>
</table>
</div>




```python
df_tran = df_tran.groupby("product_code").sum("quantity")
df_tran
```




<div>
<style scoped>
    .dataframe tbody tr th:only-of-type {
        vertical-align: middle;
    }

    .dataframe tbody tr th {
        vertical-align: top;
    }

    .dataframe thead th {
        text-align: right;
    }
</style>
<table border="1" class="dataframe">
  <thead>
    <tr style="text-align: right;">
      <th></th>
      <th>quantity</th>
    </tr>
    <tr>
      <th>product_code</th>
      <th></th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <th>p_001</th>
      <td>20</td>
    </tr>
    <tr>
      <th>p_002</th>
      <td>6</td>
    </tr>
    <tr>
      <th>p_003</th>
      <td>5</td>
    </tr>
  </tbody>
</table>
</div>




```python
df_join = pd.merge(left=df_tran, right=df_mst,how = "left", on = "product_code")
df_join
```




<div>
<style scoped>
    .dataframe tbody tr th:only-of-type {
        vertical-align: middle;
    }

    .dataframe tbody tr th {
        vertical-align: top;
    }

    .dataframe thead th {
        text-align: right;
    }
</style>
<table border="1" class="dataframe">
  <thead>
    <tr style="text-align: right;">
      <th></th>
      <th>product_code</th>
      <th>quantity</th>
      <th>product_name</th>
      <th>unit_price</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <th>0</th>
      <td>p_001</td>
      <td>20</td>
      <td>pencil</td>
      <td>50</td>
    </tr>
    <tr>
      <th>1</th>
      <td>p_002</td>
      <td>6</td>
      <td>notebook</td>
      <td>100</td>
    </tr>
    <tr>
      <th>2</th>
      <td>p_003</td>
      <td>5</td>
      <td>eraser</td>
      <td>80</td>
    </tr>
  </tbody>
</table>
</div>




```python
df_join["total_price"] = df_join["quantity"] * df_join["unit_price"]
df_join
```




<div>
<style scoped>
    .dataframe tbody tr th:only-of-type {
        vertical-align: middle;
    }

    .dataframe tbody tr th {
        vertical-align: top;
    }

    .dataframe thead th {
        text-align: right;
    }
</style>
<table border="1" class="dataframe">
  <thead>
    <tr style="text-align: right;">
      <th></th>
      <th>product_code</th>
      <th>quantity</th>
      <th>product_name</th>
      <th>unit_price</th>
      <th>total_price</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <th>0</th>
      <td>p_001</td>
      <td>20</td>
      <td>pencil</td>
      <td>50</td>
      <td>1000</td>
    </tr>
    <tr>
      <th>1</th>
      <td>p_002</td>
      <td>6</td>
      <td>notebook</td>
      <td>100</td>
      <td>600</td>
    </tr>
    <tr>
      <th>2</th>
      <td>p_003</td>
      <td>5</td>
      <td>eraser</td>
      <td>80</td>
      <td>400</td>
    </tr>
  </tbody>
</table>
</div>




```python
df_join.to_csv(OUTPUT_DATA_PATH)
```




### After

#### 処理本体
```python
import pandas as pd

START_DATE = "2023-02-01"
END_DATE = "2023-02-28"
TRANSACTION_DATA_PATH = "./input_data_transaction.csv"
MASTER_DATA_PATH = "./input_data_master.csv"
OUTPUT_DATA_PATH = "./output_data.csv"


def filter_with_date(df, start_date, end_date):
    df = df[df["date"].between(start_date, end_date)]
    return df


def sum_groupby_product_code(df):
    df = df.groupby("product_code").sum("quantity")
    return df


def joinon_product_code(df_left, df_right):
    df = pd.merge(left=df_left, right=df_right, how="left", on="product_code")
    return df


def calculate_total_price(df):
    df["total_price"] = df["quantity"] * df["unit_price"]
    return df


if __name__ == "__main__":
    df_tran = pd.read_csv(TRANSACTION_DATA_PATH)
    df_mst = pd.read_csv(MASTER_DATA_PATH)
    df_tran = filter_with_date(df_tran, START_DATE, END_DATE)
    df_tran = sum_groupby_product_code(df_tran)
    df_join = joinon_product_code(df_tran, df_mst)
    df_join = calculate_total_price(df_join)
    df_join.to_csv(OUTPUT_DATA_PATH)
```



#### テストコード

```python
import pandas as pd
import pytest
from samplecode_02 import (
    filter_with_date,
    sum_groupby_product_code,
    joinon_product_code,
    calculate_total_price,
)


def test_filter_with_date():
    input_column_list = ["date", "strings"]
    input_data_list = [
        ["2023-01-15", "good morning"],
        ["2023-02-08", "hello"],
        ["2023-03-8", "good night"],
    ]
    START_DATE = "2023-02-01"
    END_DATE = "2023-02-28"
    actual_df = pd.DataFrame(data=input_data_list, columns=input_column_list)
    actual_df = filter_with_date(actual_df, START_DATE, END_DATE)
    actual_df = actual_df.reset_index(drop=True)
    expected_data_list = [["2023-02-08", "hello"]]
    expected_df = pd.DataFrame(data=expected_data_list, columns=input_column_list)

    pd.testing.assert_frame_equal(left=expected_df, right=actual_df)


def test_sum_groupby_product_code():
    input_column_list = ["product_code", "quantity"]
    input_data_list = [
        ["AAAAA", 1],
        ["AAAAA", 1],
        ["AAAAA", 1],
        ["BBBBB", 1],
        ["BBBBB", 1],
        ["CCCCC", 1],
    ]
    actual_df = pd.DataFrame(data=input_data_list, columns=input_column_list)
    actual_df = sum_groupby_product_code(actual_df)
    actual_df = actual_df.reset_index(drop=False)
    expected_data_list = [
        ["AAAAA", 3],
        ["BBBBB", 2],
        ["CCCCC", 1],
    ]
    expected_df = pd.DataFrame(data=expected_data_list, columns=input_column_list)
    pd.testing.assert_frame_equal(left=expected_df, right=actual_df)


def test_joinon_product_code():
    input_left_column_list = ["product_code", "quantity"]
    input_left_data_list = [
        ["AAAAA", 3],
        ["BBBBB", 2],
        ["CCCCC", 1],
    ]
    df_left = pd.DataFrame(data=input_left_data_list, columns=input_left_column_list)

    input_right_column_list = ["product_code", "product_name"]
    input_right_data_list = [
        ["AAAAA", "name_of_AAAAA"],
        ["BBBBB", "name_of_BBBB"],
        ["CCCCC", "name_of_CCCCC"],
    ]
    df_right = pd.DataFrame(data=input_right_data_list, columns=input_right_column_list)
    actual_df = joinon_product_code(df_left, df_right)
    actual_df = actual_df.reset_index(drop=True)
    expected_column_list = ["product_code", "quantity", "product_name"]
    expected_data_list = [
        ["AAAAA", 3, "name_of_AAAAA"],
        ["BBBBB", 2, "name_of_BBBB"],
        ["CCCCC", 1, "name_of_CCCCC"],
    ]
    expected_df = pd.DataFrame(data=expected_data_list, columns=expected_column_list)
    pd.testing.assert_frame_equal(left=expected_df, right=actual_df)


def test_calculate_total_price():
    input_column_list = ["quantity", "unit_price"]
    input_data_list = [
        [3, 100],
        [2, 5],
        [1, 1000],
    ]
    actual_df = pd.DataFrame(data=input_data_list, columns=input_column_list)
    actual_df = calculate_total_price(actual_df)
    expected_column_list = ["quantity", "unit_price", "total_price"]
    expected_data_list = [
        [3, 100, 300],
        [2, 5, 10],
        [1, 1000, 1000],
    ]
    expected_df = pd.DataFrame(data=expected_data_list, columns=expected_column_list)
    print(expected_df.head())
    print(actual_df.head())
```

## ファイル形式.ipynbではなく.pyを使いましょう

.ipynb形式は分析者界隈で普及しているが、できることに制約が多い。  
Pythonの基本的なファイルフォーマットは.py。  
Pythonのいろいろな機能を使うのに都合がいいのは.py。


### .pyを使う理由

- .ipynbファイルにはアウトプットやメタデータといったコード以外の情報が含まれる。Gitで管理する場合に邪魔。<br>Windows環境の人は、メモ帳やWinmergeで.ipynbを開いてゲッソリしたことがあるのでは...
- .ipynbファイルは自動テストしづらい。できなくはないが、いろいろ煩雑になる

### .ipynbのいいところ

- Pythonを使う分析者はみんな使える
- 処理のコードとその実行結果を単一のファイル内に保持できる。
- つまり...単発の調査やレポーティングに向いている。

### ちなみに

- 「Jupyterだとセル毎に実行できる。けど.pyだとそれができない!」という人
  - IDEのデバッグ機能を使いましょう。VScodeなどのエディタにPython用拡張機能を導入すると使える!

- 「既存notebookファイルをどうにか使いたいけどどうすれば..」という人
  - 既存notebookを.pyファイルに変換する際はnbconvertを使いましょう。
  - マジックコマンドは使えなくなるので注意!
  - VScodeで編集した.ipynbは`to script`オプションが使えない`to python`で変換する。

## 関数として記載する

処理は関数にしましょう。
自動テストをするために必要なので。
また、コードを再利用するという観点でも関数にしておくことはおススメ。

例えば、beforeのコードの日付範囲で抽出する部分を関数にしてみると...

```python
# 元のコード
# In[5]:

df_tran = df_tran[df_tran["date"].between(START_DATE, END_DATE)]
df_tran
```
```python
# 関数に書き換えると...
def filter_with_date(df, start_date, end_date):
    df = df[df["date"].between(start_date, end_date)]
    return df
df_tran = filter_with_date(df_tran, START_DATE, END_DATE)
```

こんな感じで各処理を書き換えて、さらに処理実行部分をスクリプトの最後に集めて読みやすくする。

## pytest

pytestはPythonの自動テスト用ライブラリ。

### pytestを選ぶ理由

- 標準ライブラリunittestはJavaっぽい。unittestはPythonっぽくない。(らしい)
- 情報が豊富
- 便利なプラグインが多い

### pytestを使ったテストの大まかな流れ

- テスト対象の関数をテストコードにimportする
- テストコードの中にはテスト関数を記載する
- テスト関数の中で
    - テスト対象コードを実行する
    - 期待する結果を記載する
    - 期待する結果とテスト対象コードの実行結果を比較する
    - 比較結果が一致すればテストに合格、一致しなければテスト不合格
- pytest コマンドでテストを実行する。

### やってみる

上で作ったfilter_with_date関数のテストを書いてみる

```python
import pandas as pd
import pytest
from samplecode_02 import filter_with_date

def test_filter_with_date():
    input_column_list = ["date", "strings"]
    input_data_list = [
        ["2023-01-15", "good morning"],
        ["2023-02-08", "hello"],
        ["2023-03-8", "good night"],
    ]
    START_DATE = "2023-02-01"
    END_DATE = "2023-02-28"
    actual_df = pd.DataFrame(data=input_data_list, columns=input_column_list)
    actual_df = filter_with_date(actual_df, START_DATE, END_DATE)
    actual_df = actual_df.reset_index(drop=True)
    expected_data_list = [["2023-02-08", "hello"]]
    expected_df = pd.DataFrame(data=expected_data_list, columns=input_column_list)

    pd.testing.assert_frame_equal(left=expected_df, right=actual_df)
```

### コード説明

- `actual_df = actual_df.reset_index(drop = True) `としているのは、フィルターによってインデックスが期待するデータとズレてしまうから。
- 逆に`test_sum_groupby_product_code`関数では`actual_df = actual_df.reset_index(drop=False)`としている。その理由はgroupby()で選択されたカラムがindexに設定しているため、dropせずにカラムとして保持したいから。
- `pd.testing.assert_frame_equal(left=expected_df, right=actual_df)`で期待する加工結果のdataframeと実際に関数で加工した結果のdataframeを比較する。
- 同じ要領で各処理のテストコードを書くと最終的なテストコードになる。
- テスト用データ作成はpytestのfixture機能を使うともっとスマートに書けるが、今回は割愛。


### 実行

- `pytest`コマンドで実行する。
- `pytest -v`にすると結果の詳細を教えてくれる

`pytest -v`コマンド実行結果のイメージはこんな感じ。
```bash
===================================== test session starts ======================================
platform linux -- Python 3.9.5, pytest-7.2.1, pluggy-1.0.0 -- /home/username/venv/bin/python
cachedir: .pytest_cache
rootdir: /home/username/sample-pandas-pytest
plugins: anyio-3.6.2
collected 4 items                                                                              

test_samplecode.py::test_filter_with_date PASSED                                         [ 25%]
test_samplecode.py::test_sum_groupby_product_code PASSED                                 [ 50%]
test_samplecode.py::test_joinon_product_code PASSED                                      [ 75%]
test_samplecode.py::test_calculate_total_price PASSED                                    [100%]

====================================== 4 passed in 0.20s =======================================
```

## 残課題

### pytest関連

- fixtureを使ってテストデータを作成し、コードをスッキリさせる
- Database連携してSQLクエリを実行する場合のテストをする。mockとか使う?
- ディレクトリ構造/ファイル構成の最適化

### テスト全般

- テストケースの検討(異常な入力とか境界値とか...)
- データ加工部分の品質と、データそのものの品質の切り分け
- 負荷テスト用のサイズが大きいダミーデータ作成

### 普及活動

- 自動テストという仕組みや概念、有用性、そして意外と簡単だということを理解してもらう。
- Jupyter notebook中心に使っている人に、.py形式に慣れてもらう。

## 感想

- テストを書くと、コードがスッキリする。テストしやすさを基準にコードを書くことが大事
- 数ヶ月後に自分のコードを読むと絶望するのが常だが、テストコードがあるとだいぶマシ。なぜならテストコードはそれ自体が動作を説明するものだから
-  「SIerは自動テスト導入しづらい。なぜならウォーターフォールを基本とした考え方なので、製造とテストを完全に別物として捉えるから」という意見を見かけたが、<br>デバッグツールとして自動テストを導入するのはできるし、アリなのではないだろうか。あくまで製造とテストは別と捉えて進めて、テスト工程ではあわよくばテスト工程でテストコードを流用する。部分的な自動化でもよい。
- 課題でも挙げたが、いかにして普及させていくかが大変だ…

## まとめ

- 自動テストを使って効率と品質を向上させよう！
- Pythonの場合、pytestを使うと簡単！