100本ノック / システム開発 / システム開発100本ノック
製造業のCSV取込からダッシュボードまで｜業務システム実践10本ノック
CSVからダッシュボードまでつなぐ製造実績管理 — 業務システム機能10本ノック
製造現場の日次生産実績管理を題材に、CSVアップロード、DB登録、ページネーション、複数条件・日付範囲検索、集計API、グラフ用API、ダッシュボード、帳票、バッチ処理を一続きで設計します。個々の機能を作るだけでなく、入力データが経営・現場の判断材料になるまでの流れを、架空データとPythonで確認します。
対象はシステム開発100本ノックの No.081〜No.090 です。
[!NOTE] 本資料は、数理工房 (もしくは代表である和山個人) が過去に企業研修において使用した notebook を企業様の許可を得て再構成・編集のうえ公開しています。
掲載データはすべて架空のものであり、実在する企業・工場・数値とは一切関係ありません。
はじめに：この記事で扱う製造業の実務課題
工場では、設備や現場端末から得た生産実績をCSVで集め、業務システムへ登録し、朝会の一覧、管理者のダッシュボード、月次帳票へ展開することがあります。この流れの途中で列名の揺れ、重複登録、検索の遅延、集計定義の不一致が生じると、同じ工場の数字なのに画面と帳票が一致しません。
本Notebookでは「取り込めたか」だけでなく、完全性・一意性・適時性・追跡可能性を管理します。最終目的は画面数を増やすことではなく、品質・納期・稼働に関する意思決定を、再現可能なデータ処理で支えることです。
現場でよくある状況
- 拠点ごとにCSVの列名や文字コードが異なり、担当者が手修正している
- 再アップロードで同じ実績が二重登録される
- 一覧が数万件になり、画面表示と検索に時間がかかる
- 画面・集計・帳票が別々の条件で計算され、値が一致しない
- 日付の境界、夜勤、タイムゾーンの扱いが曖昧で日次実績がずれる
- バッチ失敗に翌朝まで気づかず、古いKPIで会議を始める
なぜこの問題は判断が難しいのか
業務システムの品質は、正常系の画面だけでは評価できません。大容量、欠損、重複、遅延、部分失敗、再実行を含めて、業務上の締切までに正しい数字を提供できるかが重要です。また「不良率」一つでも、分母を投入数・完成数・検査数のどれにするかで値が変わります。
そこで、機能要件を次のKPIへ置き換えます。
機能、データ定義、性能、運用を同じ設計対象として扱うことが判断のぶれを抑えます。
今回扱うノックの全体像
| No. | テーマ | 製造実績管理での確認点 |
|---|---|---|
| 081 | CSVアップロード | 形式・サイズ・必須列の入口検査 |
| 082 | CSVデータをDB登録 | 検証、重複排除、トランザクション |
| 083 | ページネーション | 大量明細を安定して閲覧 |
| 084 | 複数条件検索 | 現場の調査手順に沿った絞り込み |
| 085 | 日付範囲検索 | 境界と夜勤を含む期間定義 |
| 086 | 集計API | KPI定義を一元化 |
| 087 | グラフ表示用API | 描画しやすい時系列データ |
| 088 | ダッシュボード | 異常発見から明細確認への導線 |
| 089 | 帳票・レポート | 確定値、版、出力条件の保存 |
| 090 | バッチ処理 | 定刻実行、再実行、監視 |
データは「入口 → 永続化 → 検索 → 集計・可視化 → 配布・定期実行」の順に流れます。
Python環境の準備
pandasとnumpyでデータを扱い、matplotlibで可視化します。日本語表示にはjapanize_matplotlibを利用します。外部サービスや外部データへは接続せず、乱数シードを固定します。ここでのPython処理はWeb実装そのものではなく、APIやDBへ実装すべき業務ルールを小さく検証するものです。
%matplotlib inline
%config InlineBackend.figure_format = 'svg'
import sys
import json
import hashlib
from io import StringIO
import numpy as np
import pandas as pd
import matplotlib
import matplotlib.pyplot as plt
import japanize_matplotlib
from IPython.display import display
SEED = 20260712
rng = np.random.default_rng(SEED)
pd.set_option("display.max_columns", 20)
pd.set_option("display.width", 140)
print("Python:", sys.version.split()[0])
print("pandas:", pd.__version__, "numpy:", np.__version__, "matplotlib:", matplotlib.__version__)
print("seed:", SEED)Python: 3.13.1
pandas: 3.0.3 numpy: 2.5.1 matplotlib: 3.11.0
seed: 20260712架空データの作成
2026年4〜6月の3工場・6ラインについて、1,800件の生産実績を生成します。各行は実績ID、完了日時、工場、ライン、品番、良品数、不良数、停止時間、登録元を持ちます。CSV取込試験用には、正常行に加えて必須値欠損、数量不整合、重複を意図的に含めます。
本番では生産実績IDの採番主体、設備時刻の同期、訂正履歴、シフト境界を業務部門と合意する必要があります。
n = 1800
start = pd.Timestamp("2026-04-01 00:00")
completed_at = start + pd.to_timedelta(rng.integers(0, 91 * 24 * 60, n), unit="m")
factory = rng.choice(["東工場", "西工場", "中央工場"], n, p=[0.40, 0.35, 0.25])
line_map = {"東工場": ["E-1", "E-2"], "西工場": ["W-1", "W-2"], "中央工場": ["C-1", "C-2"]}
line = [rng.choice(line_map[f]) for f in factory]
good = rng.integers(70, 241, n)
base_defect = np.where(np.array(line) == "W-2", 0.030, 0.015)
defect = rng.binomial(np.maximum(good, 1), base_defect)
production = pd.DataFrame({
"result_id": [f"PR-{i:06d}" for i in range(1, n + 1)],
"completed_at": completed_at,
"factory": factory,
"line": line,
"product": rng.choice(["AX-100", "AX-200", "BZ-310", "CZ-500"], n),
"good_qty": good,
"defect_qty": defect,
"downtime_min": np.round(rng.gamma(1.4, 8.0, n), 1),
"source": rng.choice(["設備連携", "現場端末", "CSV"], n, p=[0.55, 0.30, 0.15]),
}).sort_values(["completed_at", "result_id"]).reset_index(drop=True)
production["total_qty"] = production["good_qty"] + production["defect_qty"]
production["defect_rate_pct"] = production["defect_qty"] / production["total_qty"] * 100
upload_sample = production.sample(115, random_state=SEED).copy()
upload_sample = pd.concat([upload_sample, upload_sample.iloc[[0, 1]]], ignore_index=True)
upload_sample.loc[3, "factory"] = None
upload_sample.loc[7, "good_qty"] = -5
display(production.head())
print("生産実績:", production.shape, "期間:", production.completed_at.min(), "〜", production.completed_at.max())
print("アップロード試験行数:", len(upload_sample))| result_id | completed_at | factory | line | product | good_qty | defect_qty | downtime_min | source | total_qty | defect_rate_pct | |
|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | PR-001150 | 2026-04-01 00:40:00 | 東工場 | E-1 | BZ-310 | 211 | 1 | 4.6 | CSV | 212 | 0.471698 |
| 1 | PR-000624 | 2026-04-01 00:58:00 | 東工場 | E-1 | CZ-500 | 117 | 4 | 14.8 | CSV | 121 | 3.305785 |
| 2 | PR-001316 | 2026-04-01 01:13:00 | 西工場 | W-1 | BZ-310 | 218 | 3 | 21.6 | CSV | 221 | 1.357466 |
| 3 | PR-000365 | 2026-04-01 01:38:00 | 東工場 | E-1 | AX-200 | 173 | 1 | 4.8 | 設備連携 | 174 | 0.574713 |
| 4 | PR-001490 | 2026-04-01 02:27:00 | 中央工場 | C-2 | AX-200 | 106 | 1 | 17.3 | CSV | 107 | 0.934579 |
生産実績: (1800, 11) 期間: 2026-04-01 00:40:00 〜 2026-06-30 22:36:00
アップロード試験行数: 117No.081：CSVアップロード機能を作る
実務での意味
アップロードは外部データがシステムへ入る境界です。ファイル名だけを信用せず、拡張子、サイズ、文字コード、ヘッダー、行数を確認し、処理受付IDを返します。大容量ファイルは同期処理で待たせず、受付後に非同期処理し、利用者が進捗とエラーを確認できる設計が適します。
分析・モデル化の考え方
入口検査と行単位の業務検査を分けます。ここでは必須列集合 と受信列集合 の差 を求め、欠けていれば処理を開始しません。受付時にはファイル内容のハッシュを保存すると、同一ファイルの再送検知に使えます。
Pythonで確認する
required = {"result_id", "completed_at", "factory", "line", "product", "good_qty", "defect_qty"}
csv_text = upload_sample.to_csv(index=False)
received = pd.read_csv(StringIO(csv_text))
missing_columns = sorted(required - set(received.columns))
file_hash = hashlib.sha256(csv_text.encode("utf-8")).hexdigest()[:16]
upload_check = pd.DataFrame({
"検査": ["ファイルサイズ", "行数", "必須列", "内容ハッシュ"],
"結果": [f"{len(csv_text.encode('utf-8')):,} bytes", f"{len(received):,} 行", "OK" if not missing_columns else str(missing_columns), file_hash],
"判定": ["OK", "OK", "OK" if not missing_columns else "NG", "記録"],
})
display(upload_check)| 検査 | 結果 | 判定 | |
|---|---|---|---|
| 0 | ファイルサイズ | 11,174 bytes | OK |
| 1 | 行数 | 117 行 | OK |
| 2 | 必須列 | OK | OK |
| 3 | 内容ハッシュ | 60b1f4d6aa6d7e9d | 記録 |
結果の読み取り
必須列が揃っているため、行単位検証へ進めます。一方、列が存在することは値が正しいことを保証しません。受付ID、ハッシュ、受信者、受信時刻を保存し、同じファイルが再送された場合に「拒否」「前回結果を表示」「明示的に再処理」のどれにするかを決めます。CSV式インジェクションや過大ファイルも入口で対策します。
No.082：CSVデータをDBに登録する
実務での意味
DB登録では、一部だけ登録されて数字が不整合になることを防ぎます。不正行を除外して正常行だけ登録する方式と、1行でも不正なら全件を戻す方式は、業務の締切と訂正手順に応じて選びます。
分析・モデル化の考え方
実績IDを一意キーとし、必須値、非負数量、既存IDとの重複を検証します。再実行しても結果が増殖しない冪等性が重要です。DBでは一意制約とトランザクションを最後の防波堤にし、アプリ側検査だけに依存しません。
Pythonで確認する
staging = upload_sample.copy()
staging["必須値欠損"] = staging[["result_id", "completed_at", "factory", "line", "product"]].isna().any(axis=1)
staging["数量不整合"] = staging["good_qty"].lt(0) | staging["defect_qty"].lt(0)
staging["ファイル内重複"] = staging.duplicated("result_id", keep=False)
staging["登録可否"] = ~(staging[["必須値欠損", "数量不整合", "ファイル内重複"]].any(axis=1))
validation = staging[["必須値欠損", "数量不整合", "ファイル内重複"]].sum().rename("該当行数").to_frame()
validation.loc["登録可能", "該当行数"] = staging["登録可否"].sum()
display(validation.astype(int))
display(staging.loc[~staging["登録可否"], ["result_id", "factory", "good_qty", "必須値欠損", "数量不整合", "ファイル内重複"]])| 該当行数 | |
|---|---|
| 必須値欠損 | 1 |
| 数量不整合 | 1 |
| ファイル内重複 | 4 |
| 登録可能 | 111 |
| result_id | factory | good_qty | 必須値欠損 | 数量不整合 | ファイル内重複 | |
|---|---|---|---|---|---|---|
| 0 | PR-001092 | 西工場 | 121 | False | False | True |
| 1 | PR-001300 | 西工場 | 86 | False | False | True |
| 3 | PR-001520 | NaN | 192 | True | False | False |
| 7 | PR-000098 | 東工場 | -5 | False | True | False |
| 115 | PR-001092 | 西工場 | 121 | False | False | True |
| 116 | PR-001300 | 西工場 | 86 | False | False | True |
結果の読み取り
エラー理由を行番号・項目名・修正例とともに返せば、担当者は元ファイルを直せます。重複行を単に削除すると、内容の異なる訂正データを見逃すため、同一ID・同一内容と同一ID・異なる内容を分けます。本番登録はステージング表で検証し、監査ログと件数照合を残してから確定します。
No.083：一覧画面にページネーションを実装する
実務での意味
大量明細を一度に送ると、DB、API、ブラウザのすべてへ負荷がかかります。ページネーションは表示を速くするだけでなく、利用者が調査位置を保ち、同じ並び順で再確認できるようにする機能です。
分析・モデル化の考え方
LIMIT/OFFSETは実装しやすい一方、深いページほど読み飛ばしが増え、閲覧中の追加登録で行がずれることがあります。安定した複合ソートキーを使うカーソル方式を検討します。ここでは1ページ50件として、転送行数と概算応答時間を比較します。
Pythonで確認する
page_sizes = np.array([25, 50, 100, 250, 500, 1800])
pagination = pd.DataFrame({"1ページ件数": page_sizes})
pagination["概算応答時間_ms"] = np.round(35 + page_sizes * 0.55 + (page_sizes / 100) ** 1.5 * 5).astype(int)
pagination["概算転送量_KB"] = np.round(page_sizes * 0.72, 1)
display(pagination)
fig, ax = plt.subplots(figsize=(8, 4))
ax.plot(pagination["1ページ件数"], pagination["概算応答時間_ms"], marker="o")
ax.set_title("1ページ件数と概算応答時間")
ax.set_xlabel("1ページ件数")
ax.set_ylabel("概算応答時間（ms）")
ax.grid(True, alpha=0.3)
plt.tight_layout(); plt.show()| 1ページ件数 | 概算応答時間_ms | 概算転送量_KB | |
|---|---|---|---|
| 0 | 25 | 49 | 18.0 |
| 1 | 50 | 64 | 36.0 |
| 2 | 100 | 95 | 72.0 |
| 3 | 250 | 192 | 180.0 |
| 4 | 500 | 366 | 360.0 |
| 5 | 1800 | 1407 | 1296.0 |
結果の読み取り
ページ件数を増やすほど往復回数は減りますが、初期表示と転送量は増えます。50〜100件を初期候補に、実測で決めます。APIはcompleted_at DESC, result_id DESCのように一意になる順序を定義し、総件数の計算が高コストなら概数表示や「次ページあり」のみ返す設計も選択肢です。
No.084：複数条件検索を実装する
実務での意味
品質担当者は「西工場・W-2・AX-200・不良率2%以上」のように条件を組み合わせ、異常の範囲を特定します。検索条件はDB列を並べるのではなく、現場の調査手順と用語に合わせます。
分析・モデル化の考え方
条件間は通常AND、同一項目の複数値はORとして定義します。自由文字列をSQLへ連結せず、パラメータ化します。検索ログから利用頻度と選択性を測り、複合インデックスは頻用条件と並び順に基づいて設計します。
Pythonで確認する
steps = []
searched = production.copy(); steps.append(("全件", len(searched)))
searched = searched[searched.factory.eq("西工場")]; steps.append(("西工場", len(searched)))
searched = searched[searched.line.eq("W-2")]; steps.append(("W-2", len(searched)))
searched = searched[searched["product"].isin(["AX-100", "AX-200"])]; steps.append(("対象2品番", len(searched)))
searched = searched[searched.defect_rate_pct.ge(2.0)]; steps.append(("不良率2%以上", len(searched)))
funnel = pd.DataFrame(steps, columns=["条件適用後", "件数"])
funnel["全件比_pct"] = (funnel["件数"] / len(production) * 100).round(1)
display(funnel)
display(searched.nlargest(5, "defect_rate_pct")[["result_id", "completed_at", "product", "good_qty", "defect_qty", "defect_rate_pct"]].round(2))| 条件適用後 | 件数 | 全件比_pct | |
|---|---|---|---|
| 0 | 全件 | 1800 | 100.0 |
| 1 | 西工場 | 612 | 34.0 |
| 2 | W-2 | 291 | 16.2 |
| 3 | 対象2品番 | 142 | 7.9 |
| 4 | 不良率2%以上 | 104 | 5.8 |
/var/folders/3y/fmw40k0x78xblvb3gkcyvy1h0000gn/T/ipykernel_45011/1531514336.py:10: UserWarning: obj.round has no effect with datetime, timedelta, or period dtypes. Use obj.dt.round(...) instead.
display(searched.nlargest(5, "defect_rate_pct")[["result_id", "completed_at", "product", "good_qty", "defect_qty", "defect_rate_pct"]].round(2))| result_id | completed_at | product | good_qty | defect_qty | defect_rate_pct | |
|---|---|---|---|---|---|---|
| 358 | PR-000029 | 2026-04-19 17:01:00 | AX-100 | 105 | 9 | 7.89 |
| 761 | PR-000391 | 2026-05-09 13:45:00 | AX-200 | 196 | 13 | 6.22 |
| 65 | PR-001092 | 2026-04-04 04:18:00 | AX-200 | 121 | 8 | 6.20 |
| 187 | PR-000723 | 2026-04-10 06:34:00 | AX-200 | 108 | 7 | 6.09 |
| 436 | PR-000704 | 2026-04-22 19:37:00 | AX-100 | 92 | 5 | 5.15 |
結果の読み取り
段階別件数を示すと、どの条件が対象を絞っているか説明できます。0件のときは条件を保持したまま緩められるUIが有効です。検索APIには許可する項目・演算子・最大期間を定義し、無制限な曖昧検索による負荷を防ぎます。頻用条件は保存検索として共有すると、朝会の再現性も上がります。
No.085：日付範囲検索を実装する
実務での意味
夜勤が0時をまたぐ工場では、暦日と生産日が一致しません。「6月1日分」を0:00〜24:00とすると、夜勤実績が別日に分かれます。日時の意味を画面、API、DBで統一する必要があります。
分析・モデル化の考え方
期間は開始を含み終了を含まない半開区間 とすると、隣接期間の重複を避けられます。ここでは生産日の境界を朝8時とし、completed_at - 8時間の日付を生産日と定義します。DBではUTC保存、表示時に工場タイムゾーンへ変換する設計が基本です。
Pythonで確認する
boundary_sample = pd.DataFrame({
"completed_at": pd.to_datetime(["2026-06-01 00:30", "2026-06-01 07:59", "2026-06-01 08:00", "2026-06-01 23:30", "2026-06-02 07:59"])
})
boundary_sample["暦日"] = boundary_sample.completed_at.dt.date
boundary_sample["生産日_8時境界"] = (boundary_sample.completed_at - pd.Timedelta(hours=8)).dt.date
display(boundary_sample)
start_at, end_at = pd.Timestamp("2026-06-01 08:00"), pd.Timestamp("2026-06-02 08:00")
one_day = production[(production.completed_at >= start_at) & (production.completed_at < end_at)]
print("2026-06-01生産日（8時境界）の実績:", len(one_day), "件、良品:", f"{one_day.good_qty.sum():,}")| completed_at | 暦日 | 生産日_8時境界 | |
|---|---|---|---|
| 0 | 2026-06-01 00:30:00 | 2026-06-01 | 2026-05-31 |
| 1 | 2026-06-01 07:59:00 | 2026-06-01 | 2026-05-31 |
| 2 | 2026-06-01 08:00:00 | 2026-06-01 | 2026-06-01 |
| 3 | 2026-06-01 23:30:00 | 2026-06-01 | 2026-06-01 |
| 4 | 2026-06-02 07:59:00 | 2026-06-02 | 2026-06-01 |
2026-06-01生産日（8時境界）の実績: 18 件、良品: 2,415結果の読み取り
8時より前の実績は前の生産日に属します。APIには日付文字列だけでなく、解釈した開始・終了日時とタイムゾーンをログへ残します。締め後訂正を許す場合、現在値だけでなく「いつの時点で確定した帳票か」を再現できるよう、訂正版と確定日時を管理します。
No.086：集計APIを作成する
実務での意味
集計APIは、朝会カード、管理画面、帳票が同じKPI定義を使うための土台です。合計してから率を計算するのか、行ごとの率を平均するのかで不良率が変わるため、式と粒度を明示します。
分析・モデル化の考え方
全体不良率は加重集計として次式で求めます。
分子・分母・対象期間・除外条件もレスポンスへ含めると検算できます。キャッシュする場合はデータ更新後の無効化と鮮度時刻が必要です。
Pythonで確認する
agg = production.groupby("factory", as_index=False).agg(
実績件数=("result_id", "size"), 良品数=("good_qty", "sum"), 不良数=("defect_qty", "sum"), 停止時間分=("downtime_min", "sum")
)
agg["不良率_pct"] = agg["不良数"] / (agg["良品数"] + agg["不良数"]) * 100
agg["1件当たり停止分"] = agg["停止時間分"] / agg["実績件数"]
display(agg.round(2))
simple_mean = production.groupby("factory")["defect_rate_pct"].mean()
weighted = agg.set_index("factory")["不良率_pct"]
comparison = pd.concat([simple_mean.rename("行別率の単純平均"), weighted.rename("数量加重率")], axis=1)
display(comparison.round(3))| factory | 実績件数 | 良品数 | 不良数 | 停止時間分 | 不良率_pct | 1件当たり停止分 | |
|---|---|---|---|---|---|---|---|
| 0 | 中央工場 | 481 | 74153 | 1070 | 5321.5 | 1.42 | 11.06 |
| 1 | 東工場 | 707 | 110702 | 1649 | 8082.6 | 1.47 | 11.43 |
| 2 | 西工場 | 612 | 93811 | 2038 | 6882.8 | 2.13 | 11.25 |
| 行別率の単純平均 | 数量加重率 | |
|---|---|---|
| factory | ||
| 中央工場 | 1.416 | 1.422 |
| 東工場 | 1.468 | 1.468 |
| 西工場 | 2.113 | 2.126 |
結果の読み取り
率の単純平均と数量加重率には差が出ます。意思決定に使う集計APIでは、KPI名だけでなく計算式、単位、粒度、最終更新時刻、対象件数を契約として固定します。0除算、欠損、取消実績の扱いもテストし、画面側で再計算しない設計にします。
No.087：グラフ表示用のAPIを作成する
実務での意味
グラフ用APIは明細をすべて返すのではなく、表示粒度へ集約した点列を返します。通信量を抑え、同じ時系列を複数画面で再利用できます。欠けた日を0として扱うか欠測として扱うかも業務上の判断です。
分析・モデル化の考え方
日次の良品数、不良数、不良率を計算し、日付順のseriesとして返します。値がない日を補完する場合、「生産なし」と「データ未到着」を別状態にしなければ、停止と連携障害を誤認します。
Pythonで確認する
daily = (production.assign(date=production.completed_at.dt.floor("D"))
.groupby("date", as_index=False)
.agg(good_qty=("good_qty", "sum"), defect_qty=("defect_qty", "sum")))
daily["defect_rate_pct"] = daily.defect_qty / (daily.good_qty + daily.defect_qty) * 100
api_payload = {
"metric": "defect_rate_pct", "unit": "%", "granularity": "day",
"series": [{"x": d.strftime("%Y-%m-%d"), "y": round(v, 3)} for d, v in zip(daily.date.tail(5), daily.defect_rate_pct.tail(5))]
}
print(json.dumps(api_payload, ensure_ascii=False, indent=2))
fig, ax = plt.subplots(figsize=(10, 4))
ax.plot(daily.date, daily.defect_rate_pct, linewidth=1.5)
ax.axhline(2.0, color="crimson", linestyle="--", label="注意基準 2.0%")
ax.set_title("全工場の日次不良率")
ax.set_xlabel("日付"); ax.set_ylabel("不良率（%）")
ax.grid(True, alpha=0.3); ax.legend()
plt.tight_layout(); plt.show(){
"metric": "defect_rate_pct",
"unit": "%",
"granularity": "day",
"series": [
{
"x": "2026-06-26",
"y": 1.618
},
{
"x": "2026-06-27",
"y": 1.528
},
{
"x": "2026-06-28",
"y": 2.218
},
{
"x": "2026-06-29",
"y": 1.964
},
{
"x": "2026-06-30",
"y": 1.761
}
]
}
結果の読み取り
時系列に基準線を重ねると、単日の悪化と継続的な悪化を区別しやすくなります。APIは表示色やピクセル座標ではなく、日付、値、単位、欠測状態を返し、描画責務を画面へ残します。点数上限と許可粒度を定め、長期間では週次・月次へ自動集約する設計も有効です。
No.088：ダッシュボード画面を作成する
実務での意味
ダッシュボードの役割は、すべての数字を並べることではなく、「どこに異常があり、次に何を確認するか」を短時間で示すことです。全社KPI、工場比較、推移、要注意明細の順に掘り下げられる構成にします。
分析・モデル化の考え方
ライン別に不良率と停止時間を標準化し、注意スコアを作ります。これは正式な品質判定ではなく、確認順を決める例です。実務では閾値、更新頻度、責任者、クリック後の明細条件をKPIごとに定義します。
Pythonで確認する
line_kpi = production.groupby(["factory", "line"], as_index=False).agg(
total_qty=("total_qty", "sum"), defect_qty=("defect_qty", "sum"), downtime_min=("downtime_min", "sum"), records=("result_id", "size")
)
line_kpi["defect_rate_pct"] = line_kpi.defect_qty / line_kpi.total_qty * 100
line_kpi["downtime_per_record"] = line_kpi.downtime_min / line_kpi.records
for col in ["defect_rate_pct", "downtime_per_record"]:
line_kpi[col + "_z"] = (line_kpi[col] - line_kpi[col].mean()) / line_kpi[col].std(ddof=0)
line_kpi["attention_score"] = 0.65 * line_kpi.defect_rate_pct_z + 0.35 * line_kpi.downtime_per_record_z
display(line_kpi.sort_values("attention_score", ascending=False).round(2))
fig, ax = plt.subplots(figsize=(8, 4))
ordered = line_kpi.sort_values("attention_score")
ax.barh(ordered.line, ordered.attention_score, color=np.where(ordered.attention_score > 0.5, "tomato", "steelblue"))
ax.set_title("ライン別の要注意スコア（確認順の例）")
ax.set_xlabel("要注意スコア"); ax.set_ylabel("ライン")
ax.grid(True, axis="x", alpha=0.3)
plt.tight_layout(); plt.show()| factory | line | total_qty | defect_qty | downtime_min | records | defect_rate_pct | downtime_per_record | defect_rate_pct_z | downtime_per_record_z | attention_score | |
|---|---|---|---|---|---|---|---|---|---|---|---|
| 5 | 西工場 | W-2 | 46065 | 1305 | 3122.5 | 291 | 2.83 | 10.73 | 2.23 | -0.94 | 1.12 |
| 2 | 東工場 | E-1 | 55605 | 811 | 4168.7 | 350 | 1.46 | 11.91 | -0.43 | 1.20 | 0.14 |
| 4 | 西工場 | W-1 | 49784 | 733 | 3760.3 | 321 | 1.47 | 11.71 | -0.40 | 0.84 | 0.03 |
| 1 | 中央工場 | C-2 | 35041 | 479 | 2694.0 | 230 | 1.37 | 11.71 | -0.60 | 0.84 | -0.10 |
| 3 | 東工場 | E-2 | 56746 | 838 | 3913.9 | 357 | 1.48 | 10.96 | -0.39 | -0.52 | -0.44 |
| 0 | 中央工場 | C-1 | 40182 | 591 | 2627.5 | 251 | 1.47 | 10.47 | -0.40 | -1.42 | -0.76 |
結果の読み取り
W-2は生成時に不良率を高く設定したため、上位に現れます。ダッシュボードではスコアだけでなく、不良率と停止時間という根拠を表示し、該当期間・ラインで絞り込んだ明細へ遷移させます。赤色だけに依存せず、ラベルと数値を併記し、最終更新時刻とデータ鮮度も常に見せます。
No.089：帳票・レポート出力を実装する
実務での意味
帳票は会議、顧客報告、監査で「その時点の確定値」を共有する成果物です。画面の印刷だけで済ませず、出力条件、集計定義、版、作成者、作成時刻を記録します。Excel、CSV、PDFは利用目的に応じて選びます。
分析・モデル化の考え方
帳票の合計とDB集計を照合し、件数・良品数・不良数のチェックサムを持たせます。大量出力は非同期ジョブにし、権限、保存期間、個人情報、数式として解釈される文字列への対策を設計します。
Pythonで確認する
june = production[(production.completed_at >= "2026-06-01") & (production.completed_at < "2026-07-01")]
report = june.groupby(["factory", "line"], as_index=False).agg(
実績件数=("result_id", "size"), 良品数=("good_qty", "sum"), 不良数=("defect_qty", "sum"), 停止時間分=("downtime_min", "sum")
)
report["不良率_pct"] = report.不良数 / (report.良品数 + report.不良数) * 100
total = pd.DataFrame({
"factory": ["全工場"], "line": ["合計"], "実績件数": [report.実績件数.sum()],
"良品数": [report.良品数.sum()], "不良数": [report.不良数.sum()], "停止時間分": [report.停止時間分.sum()],
})
total["不良率_pct"] = total.不良数 / (total.良品数 + total.不良数) * 100
final_report = pd.concat([report, total], ignore_index=True)
display(final_report.round(2))
print("照合:", "OK" if total.良品数.iloc[0] == june.good_qty.sum() and total.不良数.iloc[0] == june.defect_qty.sum() else "NG")
print("帳票条件: 2026-06-01 00:00 <= completed_at < 2026-07-01 00:00 / version 1")| factory | line | 実績件数 | 良品数 | 不良数 | 停止時間分 | 不良率_pct | |
|---|---|---|---|---|---|---|---|
| 0 | 中央工場 | C-1 | 84 | 13353 | 212 | 876.9 | 1.56 |
| 1 | 中央工場 | C-2 | 73 | 11097 | 161 | 937.2 | 1.43 |
| 2 | 東工場 | E-1 | 110 | 17017 | 245 | 1460.8 | 1.42 |
| 3 | 東工場 | E-2 | 110 | 17571 | 253 | 1123.2 | 1.42 |
| 4 | 西工場 | W-1 | 108 | 16584 | 248 | 1274.5 | 1.47 |
| 5 | 西工場 | W-2 | 104 | 15804 | 447 | 1181.0 | 2.75 |
| 6 | 全工場 | 合計 | 589 | 91426 | 1566 | 6853.6 | 1.68 |
照合: OK
帳票条件: 2026-06-01 00:00 <= completed_at < 2026-07-01 00:00 / version 1結果の読み取り
ライン小計と全工場合計が元データに一致しています。実際の帳票では単体テストに加え、出力後ファイルを再読込して列、型、式、改ページ、文字切れを確認します。締め後に値が変わる業務では上書きせず、版番号と訂正理由を付け、過去版を再取得できるようにします。
No.090：バッチ処理を実装する
実務での意味
日次集計、外部連携、帳票生成は定刻バッチで自動化されます。重要なのはスケジュール登録ではなく、失敗を検知し、安全に再実行し、業務締切までに復旧できることです。
分析・モデル化の考え方
処理日を一意キーにして同じ日を再実行しても二重計上しないようにします。実行履歴には開始・終了、対象日、入力件数、出力件数、状態、エラー概要を残します。成功率だけでなく、締切前完了率、処理時間のp95、再実行回数を運用KPIにします。
Pythonで確認する
run_days = pd.date_range("2026-06-01", periods=30, freq="D")
duration = np.round(rng.lognormal(mean=np.log(7), sigma=0.35, size=len(run_days)), 1)
failed = rng.random(len(run_days)) < 0.10
recovered = failed & (rng.random(len(run_days)) < 0.75)
batch = pd.DataFrame({"対象日": run_days, "処理時間_分": duration, "初回失敗": failed, "再実行で復旧": recovered})
batch["最終成功"] = ~batch.初回失敗 | batch.再実行で復旧
batch["締切前完了"] = batch.最終成功 & (batch.処理時間_分 + np.where(batch.初回失敗, 12, 0) <= 30)
summary = pd.Series({
"予定実行回数": len(batch), "初回成功率_pct": (~batch.初回失敗).mean() * 100,
"最終成功率_pct": batch.最終成功.mean() * 100, "締切前完了率_pct": batch.締切前完了.mean() * 100,
"処理時間p95_分": batch.処理時間_分.quantile(0.95),
})
display(summary.round(1).to_frame("値"))
display(batch[batch.初回失敗])| 値 | |
|---|---|
| 予定実行回数 | 30.0 |
| 初回成功率_pct | 96.7 |
| 最終成功率_pct | 100.0 |
| 締切前完了率_pct | 100.0 |
| 処理時間p95_分 | 10.4 |
| 対象日 | 処理時間_分 | 初回失敗 | 再実行で復旧 | 最終成功 | 締切前完了 | |
|---|---|---|---|---|---|---|
| 20 | 2026-06-21 | 8.4 | True | True | True | True |
結果の読み取り
初回成功率だけでなく、再実行後の最終成功率と締切前完了率を見ると業務影響を評価できます。再実行は失敗地点から継続するのか全体をやり直すのかを決め、二重計上をテストします。アラートは「失敗した」だけでなく、対象日、処理名、影響、再実行可否、ログへの導線を通知します。
対象ノックを通して見える実務上の示唆
-
入口で止めるルールと、DBで守る制約を重ねる
CSV検査、ステージング、一意制約、トランザクションによって、誤登録と二重登録を段階的に防ぎます。 -
検索・集計・帳票を同じデータ定義へ接続する
日付境界、不良率の分母、取消データの扱いを共通化すると、画面と帳票の数字を説明できます。 -
ダッシュボードは異常発見から明細確認まで設計する
KPIの表示だけで終わらず、根拠、更新時刻、絞り込み済み明細、担当アクションへつなぎます。 -
非同期処理は運用までが機能である
受付ID、進捗、監視、アラート、冪等な再実行があって初めて、CSV取込やバッチを安心して自動化できます。
実務導入する場合に必要なこと
1. 業務定義とデータ契約
実績IDの採番、必須項目、単位、シフト境界、締め、訂正、取消、KPI計算式を、生産・品質・情報システム部門で合意します。CSV仕様とAPI仕様は版管理し、変更時の移行期間を設けます。
2. 性能・容量設計
平常時と締め時間帯の件数、ファイルサイズ、保持年数、同時利用者、許容応答時間を測り、インデックス、ページサイズ、キャッシュ、非同期化を設計します。架空データの結果をそのまま性能保証には使えません。
3. セキュリティと監査
工場・職務別の権限、アップロードファイルの検査、SQLインジェクション対策、帳票の保存期限、操作・訂正・出力履歴を整備します。機密情報をログやエラーメッセージへ出さないことも重要です。
4. テストと運用KPI
境界日時、0件、大容量、重複、部分失敗、再実行、締め後訂正をテストします。本番では取込成功率、検索p95、データ鮮度、締切前完了率、照合差異を監視し、担当者と復旧手順を定めます。
まとめ
No.081〜No.090では、CSVで届いた製造実績を安全にDBへ登録し、一覧・検索・集計・グラフ・ダッシュボード・帳票へ展開し、バッチで継続運用する流れを確認しました。
重要なのは、各機能を独立した画面やAPIとして考えず、同じデータ定義、時点、監査証跡でつなぐことです。正常系だけでなく重複、境界、遅延、失敗、再実行を設計すると、業務システムは「データを保管する箱」から、現場が根拠を持って判断する基盤になります。
法人向けのご相談
数理工房では、製造業のCSV・Excel業務の整理、データ基盤・業務システムの設計開発、KPI定義、ダッシュボード、帳票・バッチ運用まで一貫してご支援しています。
- 工場ごとに異なる実績ファイルを統合したい
- 画面・会議資料・帳票の数字を一致させたい
- 大量データの検索や集計を高速化したい
- 手作業の集計・帳票作成を安全に自動化したい
- AI・最適化モデルの結果を既存業務へ組み込みたい
要件が固まっていない段階でも、現行業務とデータの棚卸しからご相談いただけます。
📩 お問い合わせ: surikobo.co.jp/contact
まずはお気軽にご相談ください。