Day 39：CSVデータ変換ツールで「生データをレポート用データに変える」
Day 39では、pandasを使って CSVデータ変換ツール を作っていきます。 現場から渡される「生のCSV」を、業務で使いやすい形に 読み込み → 加工 → 条件抽出 → 集計 → 新CSV出力 まで一気に流すのがゴールです。
機能の一覧は次の通りです。
- CSV読み込み
- データ加工
- 条件抽出
- 集計
- 新CSV出力
これを1つのスクリプトにまとめて、「毎回同じ処理を自動でやってくれるツール」にしていきます。
全体像：CSVデータ変換ツールの流れを言葉で整理する
まず、やりたいことをステップに分解します。
- 元のCSVファイルを読み込む
- 型変換・欠損値処理などの「データ加工」を行う
- 条件抽出で「欲しいデータだけ」に絞り込む
- 集計で「店舗ごとの売上合計」などを計算する
- 加工後データ・集計結果をそれぞれ新しいCSVとして出力する
この流れを、1つのPythonスクリプトにまとめていきます。
サンプルCSVの準備：少し“現場っぽい”データを作る
まずは、Day 39用のサンプルCSVを作ります。
# day39_make_csv.py
def create_sample_csv(filename="sales_raw.csv"):
"""Day 39用のサンプルCSVファイルを作成する関数です。"""
content = """日付,店舗,商品名,数量,単価,売上
2024-01-01,A,りんご,10,120,
2024-01-01,B,みかん,8,100,
2024-01-01,C,ぶどう,,300,
2024-01-02,A,りんご,5,120,
2024-01-02,B,みかん,abc,100,
2024-01-02,C,ぶどう,3,300,
2024-01-03,A,りんご,4,120,
2024-01-03,B,みかん,9,100,
2024-01-03,C,ぶどう,2,300,
"""
with open(filename, "w", encoding="utf-8") as f:
f.write(content)
print(f"{filename} を作成しました。")
if __name__ == "__main__":
create_sample_csv()
ポイント：
- 「売上」列は空欄（後で計算する前提）。
- 「数量」に
abcが混ざっていて、数値として扱えない行がある。 - 欠損値（空欄）と型の不整合がある、典型的な「生データ」です。
ステップ1：CSV読み込み
import pandas as pd
def load_csv(filename="sales_raw.csv"):
"""CSVを読み込んで基本情報を確認する関数です。"""
df = pd.read_csv(filename)
print("=== 読み込んだ生データ ===")
print(df)
print("\n=== 列のデータ型 ===")
print(df.dtypes)
return df
if __name__ == "__main__":
load_csv()
ここでのポイント：
pd.read_csv()でCSVをDataFrameとして読み込む。dtypesを見て、「数量」「単価」「売上」がobject（文字列扱い）になっていることを確認する。
ステップ2：データ加工（型変換＋欠損値処理＋売上再計算）
型変換：文字列の数値を「本物の数値」にする
import pandas as pd
def clean_data(df):
"""型変換・欠損値処理・売上再計算を行う関数です。"""
# 数量・単価・売上を数値に変換します。
df["数量"] = pd.to_numeric(df["数量"], errors="coerce")
df["単価"] = pd.to_numeric(df["単価"], errors="coerce")
df["売上"] = pd.to_numeric(df["売上"], errors="coerce")
print("=== 型変換後のDataFrame ===")
print(df)
# 欠損値を0で埋めます（数量・単価）。
df["数量"] = df["数量"].fillna(0)
df["単価"] = df["単価"].fillna(0)
# 売上を再計算します（数量 × 単価）。
df["売上"] = df["数量"] * df["単価"]
print("\n=== 欠損値処理＋売上再計算後のDataFrame ===")
print(df)
return df
重要ポイント：
pd.to_numeric(..., errors="coerce")で、変換できない値をNaNにする。- 欠損値（NaN）を
fillna(0)で0に置き換える。 - 「売上」は元のCSVでは空欄なので、
数量 × 単価で再計算する。
ステップ3：条件抽出（欲しいデータだけに絞る）
例：売上が1000以上の行だけを抽出する
def filter_data(df):
"""条件抽出で欲しいデータだけに絞り込む関数です。"""
# 売上が1000以上の行だけを抽出します。
df_filtered = df[df["売上"] >= 1000]
print("\n=== 売上1000以上のデータ ===")
print(df_filtered)
return df_filtered
ポイント：
df[条件式]で条件に合う行だけを抽出できる。- 条件式は「列に対する比較」（例：
df["売上"] >= 1000）。
ステップ4：集計（店舗ごとの売上合計）
groupby() + sum() で店舗別集計
def aggregate_data(df):
"""店舗ごとの売上合計を集計する関数です。"""
grouped = df.groupby("店舗")["売上"].sum()
print("\n=== 店舗ごとの売上合計（Series） ===")
print(grouped)
result_df = grouped.reset_index()
result_df.rename(columns={"売上": "売上合計"}, inplace=True)
print("\n=== 店舗ごとの売上合計（DataFrame） ===")
print(result_df)
return result_df
ポイント：
groupby("店舗")["売上"].sum()で「店舗ごとに売上合計」を計算。reset_index()で「店舗」「売上合計」の2列からなる表に整形。
ステップ5：新CSV出力（加工後データ＋集計結果）
加工後データを出力
def export_clean_data(df, filename="sales_clean.csv"):
"""加工後のデータを新しいCSVとして出力する関数です。"""
df.to_csv(filename, index=False, encoding="utf-8")
print(f"\n加工後データを {filename} に出力しました。")
集計結果を出力
def export_aggregated_data(df_agg, filename="sales_aggregated.csv"):
"""集計結果を新しいCSVとして出力する関数です。"""
df_agg.to_csv(filename, index=False, encoding="utf-8")
print(f"集計結果を {filename} に出力しました。")
ポイント：
to_csv()でDataFrameをCSVとして保存できる。index=Falseで行番号をCSVに含めないようにしている。
完成版：CSVデータ変換ツールテンプレート
ここまでの流れを1つのスクリプトにまとめた「CSVデータ変換ツール」のテンプレートです。
# day39_csv_transform_tool.py
import pandas as pd
def load_csv(filename="sales_raw.csv"):
df = pd.read_csv(filename)
print("=== 読み込んだ生データ ===")
print(df)
return df
def clean_data(df):
# 型変換
df["数量"] = pd.to_numeric(df["数量"], errors="coerce")
df["単価"] = pd.to_numeric(df["単価"], errors="coerce")
df["売上"] = pd.to_numeric(df["売上"], errors="coerce")
# 欠損値処理
df["数量"] = df["数量"].fillna(0)
df["単価"] = df["単価"].fillna(0)
# 売上再計算
df["売上"] = df["数量"] * df["単価"]
print("\n=== 整形後のDataFrame ===")
print(df)
return df
def filter_data(df):
# 売上1000以上の行だけに絞り込みます。
df_filtered = df[df["売上"] >= 1000]
print("\n=== 売上1000以上のデータ ===")
print(df_filtered)
return df_filtered
def aggregate_data(df):
# 店舗ごとの売上合計を計算します。
grouped = df.groupby("店舗")["売上"].sum()
result_df = grouped.reset_index()
result_df.rename(columns={"売上": "売上合計"}, inplace=True)
print("\n=== 店舗ごとの売上合計 ===")
print(result_df)
return result_df
def export_clean_data(df, filename="sales_clean.csv"):
df.to_csv(filename, index=False, encoding="utf-8")
print(f"\n加工後データを {filename} に出力しました。")
def export_aggregated_data(df_agg, filename="sales_aggregated.csv"):
df_agg.to_csv(filename, index=False, encoding="utf-8")
print(f"集計結果を {filename} に出力しました。")
def main():
# 1. CSV読み込み
df_raw = load_csv()
# 2. データ加工（型変換＋欠損値処理＋売上再計算）
df_clean = clean_data(df_raw)
# 3. 条件抽出（売上1000以上）
df_filtered = filter_data(df_clean)
# 4. 集計（店舗ごとの売上合計）
df_agg = aggregate_data(df_filtered)
# 5. 新CSV出力
export_clean_data(df_clean, "sales_clean.csv")
export_aggregated_data(df_agg, "sales_aggregated.csv")
if __name__ == "__main__":
main()
このスクリプトを実行すると、
sales_raw.csvを読み込み- 型変換・欠損値処理・売上再計算を行った整形済みデータを
sales_clean.csvに出力し - 売上1000以上のデータを対象に店舗ごとの売上合計を計算し
- 集計結果を
sales_aggregated.csvに出力します。
Day 39のまとめ
Day 39では、CSV処理として、
- 「生のCSV」をpandasで読み込むところから始めて
- 型変換・欠損値処理・売上再計算というデータ加工の基本フローを作り
- 条件抽出で「売上1000以上」など、欲しいデータだけに絞り込み
groupby()＋sum()で店舗ごとの売上合計を集計し- 加工後データと集計結果をそれぞれ新しいCSVとして出力するツールを完成させました。
ここまで来ると、 「毎回ExcelやCSVを開いて手作業で整形・集計している仕事」を、1本のPythonスクリプトに置き換える というイメージがかなり具体的になっているはずです。
このCSVデータ変換ツールの型をベースに、 列名や条件、集計内容を変えることで、さまざまな業務データに応用していくことができます。