PythonでCSVを自動集計する仕組み

売上CSV、アフィリエイト成果、広告費、ポイント履歴を毎朝開き、Excelへ転記していないでしょうか。

集計作業そのものは収益を生みません。それでも人間が毎回介在すると、データが増えるほど作業時間が膨らみ、転記ミスや確認漏れも起こります。自分が休んでいる間も収益状況を把握したいなら、PythonでCSVを自動集計し、異常が起きたときだけ通知する仕組みが役立ちます。

この記事では、Python初心者でも実行できるように、CSVの読み込み、入力検査、カテゴリ別集計、結果保存、実行ログ、定期実行までを順番に解説します。読了後には、次の状態を目指せます。

  • 指定フォルダへCSVを置くと自動で集計される
  • 売上、ポイント、件数を同じルールで計算できる
  • 不正な金額や重複取引を検知できる
  • 人間は全明細ではなく、異常と数字の変化だけを確認できる
  • 集計結果を商品改善や収益導線の判断材料に使える

CSV集計だけで収益が発生するわけではありません。集計は、すでにある事業や副業の数字を監視し、判断を速くするための基盤です。

本記事は一般的な技術情報であり、収益を保証するものでも、投資判断を勧めるものでもありません。

PythonによるCSV自動集計の全体像

CSVとは、表形式のデータをカンマなどで区切って保存するファイルです。たとえば、次のような取引データを想定します。

date,transaction_id,channel,amount,status
2026-07-01,A001,blog,1200,approved
2026-07-01,A002,mail,800,pending
2026-07-02,A003,blog,1500,approved

各用語を具体例に置き換えると、次のようになります。

  • 列：dateやamountなどのデータ項目
  • 行：A001の取引など、1件分の記録
  • 集計キー：channelなど、結果を分類する基準
  • ステータス：approvedなど、成果が確定したかを示す状態
  • 一意キー：transaction_idなど、同じ取引を識別する値

PythonでCSVを自動集計する流れは、以下のとおりです。

収益サービスからCSVを取得
        ↓
入力フォルダへ保存
        ↓
列名・日付・金額・取引IDを検査
        ↓
媒体別・日付別に集計
        ↓
集計CSVと実行ログを保存
        ↓
異常がある場合だけ通知

この構成なら、人間が毎回CSVを開く必要はありません。確認対象を「全明細」から「失敗した処理と重要な数字の変化」に絞れます。

さらに、集計結果を別の処理へ渡せば、売れ筋商品の抽出、成果が伸びた記事の発見、ポイント承認率の監視、改善すべきCTAの選定などへ発展させられます。

CSV自動集計の処理フロー

10万行の合成データで検証した結果

一般論だけで終わらせないため、2026年7月22日、このサイトの運用リポジトリ上で10万行の検証用CSVをメモリ内に生成し、Python標準ライブラリで集計しました。

検証条件

  • 実行日：2026年7月22日
  • Python：3.11.9
  • 入力：プログラムで生成した10万行の合成データ
  • 分類：blog、mail、sns、direct
  • 金額：1から500までを繰り返す検証値
  • 処理内容：CSV読み込み、取引ID重複検査、Decimalによる媒体別集計
  • 計測対象外：検証用データの生成
  • 計測回数：1回

今回の実行結果

{
  "rows": 100000,
  "unique_ids": 100000,
  "errors": 0,
  "grand_total": "25050000",
  "elapsed_seconds": 0.253672
}

媒体別の合計は、blogが6,225,000、mailが6,250,000、snsが6,275,000、directが6,300,000でした。4媒体の合計は25,050,000となり、検証用データから計算した期待値と一致しました。

この金額は売上ではなく、計算結果を検査するための合成値です。処理時間も、このPCで1回だけ測った参考値であり、一般的な性能を示すものではありません。保存先、列数、文字コード、ストレージ速度、セキュリティソフトなどによって変動します。

また、合成データによるテストだけでは、実サービス特有の文字コード、空欄、列名変更、取消取引などを再現できません。本番投入前には、個人情報を除去した実データの複製でも検証してください。

類似記事との差は、groupbyの書き方だけを紹介するのではなく、重複防止、異常停止、ログ、定期実行、収益改善への接続までを一つの運用単位として扱う点です。

ステップ・バイ・ステップで作るCSV自動集計

1. Pythonの実行環境を確認する

WindowsではPowerShellを開き、次のコマンドを実行します。

python --version

今回の検証環境では、以下のバージョンが表示されました。

Python 3.11.9

今回は、Pythonに標準搭載されているcsvモジュールを使います。csvモジュールとは、CSVの各行を読み書きするための機能です。追加ライブラリを導入できないPCでも試せます。

pythonが見つからない場合は、Pythonをインストールしたうえで、PowerShellを開き直してください。

2. 入力CSVの仕様を固定する

自動集計を安定させるには、コードを書く前に入力ルールを決めます。今回は次の5列を必須とします。

列名意味例
date成果発生日2026-07-01
transaction_id取引を識別するIDA001
channel流入元・媒体blog
amount金額またはポイント1200
status承認状態approved

transaction_idがないと、同じCSVを再処理した際の二重計上を防ぎにくくなります。

提供元が取引IDを出力しない場合は、日付、商品、金額などから識別子を作る方法もあります。ただし、内容が同じ別取引を誤って重複扱いする可能性があるため、実データを使った事前検証が必要です。

3. フォルダを準備する

作業フォルダの中に、次の構成を作ります。

csv-automation/
├─ aggregate_csv.py
├─ input/
│  └─ sales.csv
├─ output/
└─ logs/
  • input：集計前のCSVを置く場所
  • output：集計結果の保存先
  • logs：成功・失敗を記録する場所

無人運転では、入力、出力、ログを分離しておくと障害調査が楽になります。

4. 入力検査付きのPythonコードを書く

次のコードをaggregate_csv.pyとして保存します。

import csv
import json
import sys
import time
from collections import defaultdict
from datetime import datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
INPUT_FILE = BASE_DIR / "input" / "sales.csv"
OUTPUT_FILE = BASE_DIR / "output" / "summary.csv"
TEMPORARY_FILE = OUTPUT_FILE.with_suffix(".tmp")
LOG_FILE = BASE_DIR / "logs" / "latest.json"

REQUIRED_COLUMNS = {
    "date",
    "transaction_id",
    "channel",
    "amount",
    "status",
}
ALLOWED_STATUS = {"approved", "pending", "rejected"}

started = time.perf_counter()

summary = defaultdict(
    lambda: {
        "rows": 0,
        "approved_amount": Decimal("0"),
        "pending_amount": Decimal("0"),
        "rejected_rows": 0,
    }
)

seen_ids = set()
error_messages = []

input_rows = 0
valid_rows = 0
duplicate_rows = 0
invalid_rows = 0

try:
    with INPUT_FILE.open(
        "r", encoding="utf-8-sig", newline=""
    ) as file:
        reader = csv.DictReader(file)

        missing = REQUIRED_COLUMNS - set(reader.fieldnames or [])
        if missing:
            raise ValueError(
                f"必須列がありません: {sorted(missing)}"
            )

        for line_number, row in enumerate(reader, start=2):
            input_rows += 1

            transaction_id = row["transaction_id"].strip()
            channel = row["channel"].strip()
            status = row["status"].strip()
            date_text = row["date"].strip()
            amount_text = row["amount"].strip()

            if not transaction_id:
                invalid_rows += 1
                error_messages.append(
                    f"{line_number}行目: IDが空です"
                )
                continue

            if transaction_id in seen_ids:
                duplicate_rows += 1
                error_messages.append(
                    f"{line_number}行目: IDが重複しています"
                )
                continue

            if not channel:
                invalid_rows += 1
                error_messages.append(
                    f"{line_number}行目: channelが空です"
                )
                continue

            if status not in ALLOWED_STATUS:
                invalid_rows += 1
                error_messages.append(
                    f"{line_number}行目: 不明なstatusです"
                )
                continue

            try:
                datetime.strptime(date_text, "%Y-%m-%d")
            except ValueError:
                invalid_rows += 1
                error_messages.append(
                    f"{line_number}行目: 日付が不正です"
                )
                continue

            try:
                amount = Decimal(amount_text)
                if not amount.is_finite():
                    raise InvalidOperation
            except InvalidOperation:
                invalid_rows += 1
                error_messages.append(
                    f"{line_number}行目: 金額が不正です"
                )
                continue

            seen_ids.add(transaction_id)
            valid_rows += 1

            data = summary[channel]
            data["rows"] += 1

            if status == "approved":
                data["approved_amount"] += amount
            elif status == "pending":
                data["pending_amount"] += amount
            else:
                data["rejected_rows"] += 1

    if input_rows == 0:
        raise ValueError("データ行がありません")

    if error_messages:
        raise ValueError("; ".join(error_messages[:10]))

    OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True)

    with TEMPORARY_FILE.open(
        "w", encoding="utf-8-sig", newline=""
    ) as file:
        writer = csv.writer(file)
        writer.writerow([
            "channel",
            "rows",
            "approved_amount",
            "pending_amount",
            "rejected_rows",
        ])

        for channel, data in sorted(summary.items()):
            writer.writerow([
                channel,
                data["rows"],
                data["approved_amount"],
                data["pending_amount"],
                data["rejected_rows"],
            ])

    TEMPORARY_FILE.replace(OUTPUT_FILE)

    result = {
        "ok": True,
        "finished_at": datetime.now().isoformat(),
        "input_rows": input_rows,
        "valid_rows": valid_rows,
        "duplicate_rows": duplicate_rows,
        "invalid_rows": invalid_rows,
        "unique_ids": len(seen_ids),
        "output_groups": len(summary),
        "error_count": 0,
        "elapsed_seconds": round(
            time.perf_counter() - started, 6
        ),
    }

except Exception as exc:
    if TEMPORARY_FILE.exists():
        TEMPORARY_FILE.unlink()

    result = {
        "ok": False,
        "finished_at": datetime.now().isoformat(),
        "input_rows": input_rows,
        "valid_rows": valid_rows,
        "duplicate_rows": duplicate_rows,
        "invalid_rows": invalid_rows,
        "unique_ids": len(seen_ids),
        "error_count": len(error_messages),
        "error": str(exc),
        "elapsed_seconds": round(
            time.perf_counter() - started, 6
        ),
    }

LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
LOG_FILE.write_text(
    json.dumps(result, ensure_ascii=False, indent=2),
    encoding="utf-8",
)

print(json.dumps(result, ensure_ascii=False))
sys.exit(0 if result["ok"] else 1)

金額にはfloatではなくDecimalを使用しています。Decimalとは、10進数を意図した桁で扱う型です。金額計算で発生し得る小さな丸め誤差を避けやすくなります。

このサンプルでは負の金額も受け付けます。返金を負数で表現するサービスがあるためです。負数を認めない運用なら、amount < 0をエラーにする検査を追加してください。

出力は一度.tmpへ保存し、完成後にsummary.csvへ置き換えています。これはアトミック更新と呼ばれる考え方で、処理途中の不完全なCSVが正式な結果として残る危険を減らします。

ただし、同じスクリプトを同時に複数起動すると、一時ファイルが競合する可能性があります。タスクスケジューラでは「既に実行中の場合は新しいインスタンスを開始しない」設定にしてください。

5. サンプルCSVで正常系を確認する

input/sales.csvへ次の内容を保存します。

date,transaction_id,channel,amount,status
2026-07-01,A001,blog,1200,approved
2026-07-01,A002,mail,800,pending
2026-07-02,A003,blog,1500,approved
2026-07-02,A004,sns,300,rejected

PowerShellから実行します。

python aggregate_csv.py

成功すると、output/summary.csvへ次の結果が出ます。

channel,rows,approved_amount,pending_amount,rejected_rows
blog,2,2700,0,0
mail,1,0,800,0
sns,1,0,0,1

blogの確定額2,700は、サンプル内の1,200と1,500を足した値です。説明用のデータであり、実際の売上ではありません。

同時にlogs/latest.jsonが作られます。時刻と処理時間を省略すると、主要部分は次のようになります。

{
  "ok": true,
  "input_rows": 4,
  "valid_rows": 4,
  "duplicate_rows": 0,
  "invalid_rows": 0,
  "unique_ids": 4,
  "output_groups": 3,
  "error_count": 0
}

無人運用では、出力ファイルの存在だけでなく、今回のログでokがtrueになっているかを監視します。前回の出力が残っていると、今回も成功したように見えるためです。

6. 壊れたデータで異常系を試す

次のように、金額を文字列へ変更します。

2026-07-01,A002,mail,未確定,pending

この状態で実行すると、プログラムは終了コード1を返し、ログのokはfalseになります。正式なsummary.csvは更新されません。

PowerShellで終了コードを確認するには、実行直後に次のコマンドを入力します。

$LASTEXITCODE

不正な金額をゼロとして処理すると、実際には入力ミスなのに「売上が減った」という誤ったレポートが生成されます。変換できない値は隔離し、担当者へ通知する方が安全です。

最低限、次の異常系も試してください。

  • 取引IDを重複させる
  • 必須列を削る
  • 日付を2026/07/01など別形式にする
  • statusへ未定義の値を入れる
  • 見出しだけでデータ行がないCSVを渡す
  • 入力ファイル自体を削除する

7. Windowsで定期実行する

手動実行が安定したら、Windowsのタスクスケジューラへ登録します。

設定例は次のとおりです。

  • トリガー：CSVの保存完了後、または毎日決まった時刻
  • プログラム：Pythonの実行ファイル
  • 引数：aggregate_csv.py
  • 開始場所：スクリプトを保存したフォルダ
  • 多重起動：既に実行中なら新しいインスタンスを開始しない
  • 成功条件：終了コードが0かつログのokがtrue
  • 失敗時：再試行後にメールやチャットへ通知

Pythonの実行ファイルの場所は、PowerShellで次のように確認できます。

Get-Command python

CSVの取得時刻より前に集計すると、前日のファイルを再処理する危険があります。提供元の更新時刻に余裕を持たせてください。

さらに安定させるなら、処理済みファイルを別フォルダへ移し、ファイル名とハッシュ値を保存します。ハッシュ値とは、ファイル内容から作る識別文字列です。同じCSVの再投入を検知できます。

8. 集計結果を収益改善へ接続する

CSVを保存して終わると、便利な集計ツールの範囲を出ません。次の判断や配信まで接続すると、自動化資産へ育てられます。

  • 確定額が前回より減ったチャネルだけ通知する
  • 否認率が基準を超えた案件を確認リストへ送る
  • 成果が伸びた記事テーマを次回の企画候補へ登録する
  • 商品別のクリック数と購入数から導線を比較する
  • ポイントの承認待ち期間が長い案件を一覧化する
  • 集計済みレポートを会員向けに自動配信する

ただし、前回比だけでは曜日や月末の影響を受けます。実務では「前日比」だけでなく、「直近4週間の同じ曜日との比較」や「7日移動平均」も併用すると、不要な通知を減らせます。

価格変更、広告停止、金融取引など、誤動作時の影響が大きい操作は、集計結果から直結させず、承認段階を残すのが安全です。

専門家目線のチェックポイント

入力件数と処理結果を照合する

次の関係が成立するか確認します。

入力件数
＝ 正常処理件数
＋ 重複件数
＋ 不正データ件数

今回のログ項目では、次のように照合できます。

input_rows
＝ valid_rows
＋ duplicate_rows
＋ invalid_rows

数字が一致しなければ、分類されないまま消えた行があるかもしれません。無人処理では、合計金額だけでなく件数も検算します。

確定・承認待ち・否認を分ける

pendingを確定収益へ混ぜると、利用可能な金額を過大評価します。ポイント案件やアフィリエイトでは、発生額、確定額、否認額を別々に保存してください。

また、件数ベースの承認率と金額ベースの承認率は分けてください。高額案件だけが否認されている場合、件数ベースでは問題が小さく見えることがあります。

同じ入力で結果が変わらないか確認する

同じCSVを複数回処理しても、二重加算されない性質を冪等性と呼びます。

今回のサンプルは、1ファイルを読み直して毎回summary.csvを作り直すため、同じ入力なら同じ結果になります。ただし、複数日のCSVをデータベースへ追記する設計へ発展させる場合は、取引ID、ファイルハッシュ、処理済み記録を永続的に保存する必要があります。

エラー時に古い結果を新しい結果として扱わない

アトミック更新によって壊れたCSVの上書きは防げますが、前回成功時のsummary.csvは残ります。

そのため、後続処理はsummary.csvの存在だけで判断せず、必ず次の条件を確認してください。

  • ログのokがtrue
  • finished_atが今回の実行時刻と一致している
  • input_rowsが想定範囲内
  • 入力ファイル名またはハッシュが今回の対象と一致している

ログへ個人情報を残しすぎない

氏名、メールアドレス、注文番号をそのままログへ出すと、漏えい時の影響が広がります。行番号、エラー分類、マスクしたIDなど、復旧に必要な情報へ絞ります。

完全自動化と無監視を混同しない

通常処理を無人化しても、CSV仕様の変更、認証切れ、規約改定、ストレージ不足は発生します。

目標は、正常時には人が触らず、例外時だけ通知を受けて短時間で復旧できる状態です。

視覚的証拠として残すべき画面

CSV入力と実行ログを照合するダッシュボード

上の画像はダッシュボード構成のイメージであり、今回の実行結果を撮影したスクリーンショットではありません。

実際の運用では、次の画面やファイルを証拠として保存すると、集計結果を検証しやすくなります。

  • 入力CSVの列名と先頭数行
  • 入力ファイルの更新日時とハッシュ
  • input_rows、valid_rows、error_count、okを表示した実行ログ
  • チャネル別の確定額と承認待ち額
  • 正常系・異常系テストの実行結果
  • 失敗時にsummary.csvが更新されなかったことを示す更新日時
  • 実行したPythonのバージョン
  • タスクスケジューラの最終実行結果と終了コード

抽象的なロボット画像より、入力件数と処理件数を照合できる実画面の方が、集計結果の信頼性を示せます。

スクリーンショットを公開する際は、顧客情報、APIキー、ローカルのユーザー名、取引IDを隠してください。

よくある失敗と対策

UnicodeDecodeErrorが出る

原因： CSVがUTF-8ではなく、CP932などで保存されている可能性があります。

対策： 提供元の仕様を確認し、必要に応じてencoding="cp932"へ変更します。文字コードを推測だけで固定すると、別サービスのCSVで文字化けすることがあります。

金額の合計が合わない

原因： 1,200円、空欄、全角数字、税込・税抜などが混在しています。

対策： 通貨記号や桁区切りを正規化し、変換できない行はエラーへ分離します。税込・税抜、円・ポイントも別の列で管理します。

同じ成果が二重計上される

原因： 同じCSVを再度読み込んでいるか、提供元が過去分を含む累積CSVを出力しています。

対策： 取引ID、処理済みファイル名、ファイルハッシュを保存します。CSVが日次差分か累積データかも確認してください。

タスクスケジューラでは動かない

原因： タスク実行時の作業フォルダやPython環境が、手動実行時と異なっている可能性があります。

対策： 今回のコードのように、Path(__file__).resolve().parentを基準にパスを作ります。タスクの「開始場所」とPythonの絶対パスも設定します。

出力CSVはあるのに失敗している

原因： 前回成功時のファイルが残っています。

対策： 出力の更新日時に加え、今回のログにあるok、finished_at、input_rowsを確認します。

エラー件数が急に増えた

原因： 提供元が列名、日付形式、ステータス名などを変更した可能性があります。

対策： エラーをゼロへ置き換えて処理を続けず、入力ファイルを隔離します。前回成功したCSVと列名・値の種類を比較し、仕様変更か一時的なデータ不良かを切り分けます。

PythonによるCSV自動集計が向かないケース

以下の業務では、別の方法が適する場合があります。

  • 元データの形式が毎回大幅に変わる
  • 紙や画像を目視しなければ判断できない
  • 月に数件しかなく、手作業時間も短い
  • 集計ルールが担当者の経験や文脈に依存する
  • サービス規約が自動取得や自動操作を禁止している
  • 誤集計による会計・法務上の影響が大きい
  • リアルタイム性が必要で、CSV出力では遅すぎる
  • 複数人が同時更新し、監査履歴や権限管理が必要

データ量、同時実行、履歴管理の要件が大きくなったら、CSVではなくデータベースへの移行も検討してください。

税務や会計の確定処理では、自動集計結果をそのまま申告値に使わず、正式な帳簿や専門家による確認と照合してください。

成果を測るKPI

KPI計算・確認方法改善の方向
自動処理成功率成功回数 ÷ 全実行回数失敗原因を分類する
データ完全率正常・重複・不正件数の合計 ÷ 入力件数行の消失を防ぐ
重複防止件数再処理を止めた取引数識別方法を改善する
人間介在時間確認・修正・復旧に使った時間通知内容を具体化する
平均復旧時間障害発生から再開までの時間ログと再実行手順を整える
件数ベース承認率確定件数 ÷ 発生件数案件や流入元を比較する
金額ベース承認率確定金額 ÷ 発生金額高額案件の否認を検知する
収益／介在時間粗利益 ÷ 人間の作業時間高価値な処理へ集中する
集計コストサーバー・API費 ÷ 実行回数不要な実行を減らす
CVR購入数 ÷ 商品ページ訪問数CTAや商品設計を検証する

売上だけを見ていると、問い合わせ対応や障害復旧で自由時間が減っていても気づけません。収益と人間介在時間を並べて測ることで、労働時間への依存度を下げられているか判断できます。

まとめ｜今日から取るべきアクション

PythonでCSVを自動集計する基本パターンは、次の流れです。

  1. 入力CSVの列と意味を固定する
  2. 必須列、日付、金額、取引IDを検査する
  3. 確定・承認待ち・否認を分けて集計する
  4. 一時ファイルを使って安全に保存する
  5. 成功・失敗・処理件数をログへ残す
  6. 壊れたCSVでもテストする
  7. 定期実行と異常通知を設定する
  8. 商品改善や収益計測へ接続する

読了後の最初のアクションとして、普段手作業で集計しているCSVを一つ選び、次の4項目を書き出してください。

  • 必須列
  • 重複判定に使うID
  • 成功条件
  • 失敗時の動作

その後、この記事のサンプル4行を使い、正常系と異常系を試します。最初から実データ全体を投入する必要はありません。

収益を約束する自動化は作れません。しかし、入力、検査、集計、通知、改善判断を接続すれば、「自分が毎回作業した分だけ結果が出る状態」から、通常処理が無人で進む自動化資産へ近づけます。

本気で自動化・不労所得を構築したい方へ

CSV集計を自動化しても、データ取得、商品提供、集客、販売、障害対応が手作業のままでは、自由な時間は増えません。

目指したいのは、便利なスクリプトを一つ作ることではなく、情報収集から品質検査、販売導線、収益計測、異常通知までが連動する仕組みです。

通常処理は無人で進み、人間は例外対応と次の改善へ集中する。その設計が、時間の切り売りから離れる足場になります。

Hiro運営サイトでは、アイデアを眺めて終わらず、実際に動く自動化資産へ組み上げたい方向けに、実践マニュアルを用意しています。遠回りを減らし、今日から構築を始めたい方は、目的に合う設計図を確認してください。

本気で自動化・不労所得を構築する実践マニュアルを見る →