Go to Qiita Advent Calendar 2024 Top
LoginSignup
2
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 1 year has passed since last update.

@nobhri

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

Last updated at Posted at 2023-02-08

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形式

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"

df_tran = pd.read_csv(TRANSACTION_DATA_PATH)
df_tran
date product_code quantity
0 2022-12-21 p_002 3
1 2022-12-21 p_003 3
2 2022-12-22 p_002 10
3 2023-01-10 p_001 10
4 2023-01-20 p_001 10
5 2023-02-09 p_002 3
6 2023-02-08 p_003 5
7 2023-02-08 p_001 10
8 2023-02-09 p_001 10
9 2023-02-09 p_002 3
10 2023-03-01 p_002 3
11 2023-03-01 p_002 3
12 2023-03-02 p_002 10 
df_mst = pd.read_csv(MASTER_DATA_PATH)
df_mst
product_code product_name unit_price
0 p_001 pencil 50
1 p_002 notebook 100
2 p_003 eraser 80 
df_tran = df_tran[df_tran["date"].between(START_DATE, END_DATE)]
df_tran
date product_code quantity
5 2023-02-09 p_002 3
6 2023-02-08 p_003 5
7 2023-02-08 p_001 10
8 2023-02-09 p_001 10
9 2023-02-09 p_002 3 
df_tran = df_tran.groupby("product_code").sum("quantity")
df_tran
quantity
product_code
p_001 20
p_002 6
p_003 5 
df_join = pd.merge(left=df_tran, right=df_mst,how = "left", on = "product_code")
df_join
product_code quantity product_name unit_price
0 p_001 20 pencil 50
1 p_002 6 notebook 100
2 p_003 5 eraser 80 
df_join["total_price"] = df_join["quantity"] * df_join["unit_price"]
df_join
product_code quantity product_name unit_price total_price
0 p_001 20 pencil 50 1000
1 p_002 6 notebook 100 600
2 p_003 5 eraser 80 400 
df_join.to_csv(OUTPUT_DATA_PATH)

After

処理本体

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)

テストコード

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で管理する場合に邪魔。
    Windows環境の人は、メモ帳やWinmergeで.ipynbを開いてゲッソリしたことがあるのでは...
  • .ipynbファイルは自動テストしづらい。できなくはないが、いろいろ煩雑になる

.ipynbのいいところ

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

ちなみに

  • 「Jupyterだとセル毎に実行できる。けど.pyだとそれができない!」という人

    • IDEのデバッグ機能を使いましょう。VScodeなどのエディタにPython用拡張機能を導入すると使える!

  • 「既存notebookファイルをどうにか使いたいけどどうすれば..」という人

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

関数として記載する

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

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

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

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

# 関数に書き換えると...
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関数のテストを書いてみる

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コマンド実行結果のイメージはこんな感じ。

===================================== 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は自動テスト導入しづらい。なぜならウォーターフォールを基本とした考え方なので、製造とテストを完全に別物として捉えるから」という意見を見かけたが、
    デバッグツールとして自動テストを導入するのはできるし、アリなのではないだろうか。あくまで製造とテストは別と捉えて進めて、テスト工程ではあわよくばテスト工程でテストコードを流用する。部分的な自動化でもよい。
  • 課題でも挙げたが、いかにして普及させていくかが大変だ…

まとめ

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

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?