関数・モジュール・例外処理で製造 KPI を再利用可能にする

100本ノック 第5章（No.041〜No.050）：関数・モジュール・例外処理

本記事は「データ分析のための Python 入門 100本ノック」シリーズの 第5章 です。
第4章（No.031〜040）では条件分岐と繰り返し処理を学びました。
本章では 関数によるコードの再利用・標準ライブラリ（math・datetime）の活用・
try-except による堅牢なデータパイプライン を、
ハーネス製造ラインの KPI 計算を題材に習得します。

[!NOTE] 本資料は、数理工房 (もしくは代表である和山個人) が過去に企業研修において使用した notebook を企業様の許可を得て再構成・編集のうえ公開しています。
掲載データはすべて架空のものであり、実在する企業・工場・数値とは一切関係ありません。

はじめに：この記事で扱う製造業の実務課題

ハーネス製造工場の品質管理チーム T さんの週末業務です。

毎週末の繰り返し作業
  1. 5本の製品ラインそれぞれに対し、日次生産数・不良率・稼働率を手作業で計算
  2. 不良率の計算式を5ライン × 5日分、計25回コピー&修正している
  3. 部品の発注日にリードタイム日数を加算して納入予定日をメモ帳で計算
  4. CSV 読み込み時に「文字列を数値に変換できません」エラーで処理が止まる

Python の 関数・標準ライブラリ・例外処理 を使いこなせば、
25回のコピー作業を1つの関数呼び出しに、カレンダー計算を1行のコードに、
エラーで止まるパイプラインを堅牢な処理フローに変えられます。

現場でよくある状況

場面	課題
KPI 計算の繰り返し	不良率・稼働率を5ライン分コピー＆貼り付けで手計算
発注量の算出	最低在庫・リードタイム・日次消費量から必要な整数発注数を電卓で計算
納入日の確認	発注日＋リードタイム日数のカレンダー計算をメモ帳で実施
CSV エラー対応	数値フィールドに文字列が混入すると処理全体が停止する
エラー原因特定	どの行・どの列で何が起きたか分からず対応に時間がかかる

「5ラインぶん同じコードを書く」という状況は、
関数化のサイン です。本章でその解決策を身につけます。

なぜこの問題は判断が難しいのか

関数・モジュール・例外処理には以下の落とし穴があります。

1. return の書き忘れ
   return がないと関数の戻り値は None になります。
   result = calc_revenue(...) としても result は None のまま集計できません

2. 関数スコープの誤解
   関数内で定義した変数は関数の外から見えません。
   global 変数の乱用は避け、必要な値は return で返すのが原則です

3. datetime モジュールの名前衝突
   import datetime の後は datetime.date(...) と書く必要があります。
   from datetime import datetime と混在すると混乱しやすいです

4. except Exception の乱用
   すべてのエラーを1つで捕捉すると、予期しないバグが隠れます。
   ZeroDivisionError, ValueError, KeyError など具体的な型を指定しましょう

今回扱うノックの全体像

No.	タイトル	製造業での活用場面
041	関数を定義する	不良率計算の共通関数化・5ライン分の一括計算
042	引数を使う	デフォルト引数・*args によるKPIフォーマット関数
043	戻り値を使う	タプルで不良率・良品率・サイクルタイムを同時に返す
044	売上金額を計算する関数を作る	良品売上・不良損失・純収益の5ライン一括計算と可視化
045	平均値を計算する関数を作る	日次不良率の平均・標準偏差・変動係数と週次推移グラフ
046	標準ライブラリをimportする	math・datetime・statistics のインポートと基本確認
047	mathモジュールを使う	発注量の切り上げ・不良率の 95% 信頼区間計算
048	datetimeモジュールを使う	発注日→納入予定日・シフト稼働時間・月末残日数の計算
049	try-exceptでエラーに対応する	CSV 読み込みの型エラー・列不足・値異常を安全に処理
050	データ処理で起こりやすいエラーを整理する	5種類のエラーパターンと対処法を体系的に整理

Python 環境の準備

import subprocess, sys

res = subprocess.run(["sw_vers", "-productVersion"], capture_output=True, text=True)
print(f"macOS : {res.stdout.strip()}")
print(f"Python: {sys.version}")

macOS : 26.3
Python: 3.13.1 (main, Dec  3 2024, 17:59:52) [Clang 16.0.0 (clang-1600.0.26.4)]

import numpy as np
import matplotlib
import matplotlib.pyplot as plt
import matplotlib.ticker as ticker

matplotlib.rcParams['font.family'] = 'Hiragino Maru Gothic Pro'
%config InlineBackend.figure_format = 'svg'
np.random.seed(42)

print("ライブラリ読み込み完了")
print(f"  NumPy     : {np.__version__}")
print(f"  Matplotlib: {matplotlib.__version__}")

ライブラリ読み込み完了
  NumPy     : 2.5.1
  Matplotlib: 3.11.0

架空データの作成

想定シナリオ: 自動車部品工場 / ハーネス製造ライン 品質管理チーム
期間: 2024年1月第3週（月〜金、5日間稼働）
管理対象: 5製品ライン（品名・単価・基準不良率が異なる）

# 5製品ライン × 5日間（2024-01-15〜19）の日次生産データ
lines = ["HA-01", "HA-02", "HB-01", "HB-02", "HC-01"]
line_names = ["ハーネス A-細", "ハーネス A-中", "ハーネス B-細", "ハーネス B-中", "ハーネス C-細"]
unit_prices = [1200, 1350, 980, 1100, 850]  # 円/個

day_labels = ["月(1/15)", "火(1/16)", "水(1/17)", "木(1/18)", "金(1/19)"]

# 日次生産数（seed=42）
base_prod = [450, 380, 520, 410, 480]
prod_data = np.array([[int(b + np.random.randint(-15, 16)) for _ in range(5)] for b in base_prod])

# 日次不良数（基準不良率 1.2〜2.5% ＋ ランダムノイズ）
dr_base = [0.020, 0.015, 0.025, 0.018, 0.012]
defect_data = np.array(
    [[max(1, int(round(prod_data[i][j] * dr_base[i] + np.random.randn() * 0.8))) for j in range(5)] for i in range(5)]
)

# 生産数一覧
print(f"{'ライン':<8} {'品名':<16} {'単価':>8}  {'月':>4} {'火':>4} {'水':>4} {'木':>4} {'金':>4}  {'週計':>6}")
print("=" * 65)
for i in range(len(lines)):
    row_str = "  ".join(f"{int(prod_data[i][j]):>4}" for j in range(5))
    print(f"{lines[i]:<8} {line_names[i]:<16} {unit_prices[i]:>8,}  {row_str}  {int(prod_data[i].sum()):>6,}")
print()
print(f"{'ライン':<8} {'':16} {'不良数':>8}  {'月':>4} {'火':>4} {'水':>4} {'木':>4} {'金':>4}  {'週計':>6}")
print("=" * 65)
for i in range(len(lines)):
    row_str = "  ".join(f"{int(defect_data[i][j]):>4}" for j in range(5))
    print(f"{lines[i]:<8} {line_names[i]:<16} {'':>8}  {row_str}  {int(defect_data[i].sum()):>6}")

ライン      品名                     単価     月    火    水    木    金      週計
=================================================================
HA-01    ハーネス A-細            1,200   441   454   463   449   445   2,252
HA-02    ハーネス A-中            1,350   372   393   385   371   390   1,911
HB-01    ハーネス B-細              980   523   527   515   515   528   2,608
HB-02    ハーネス B-中            1,100   415   398   402   418   397   2,030
HC-01    ハーネス C-細              850   486   485   466   488   476   2,401

ライン                            不良数     月    火    水    木    金      週計
=================================================================
HA-01    ハーネス A-細                      9     9     9     9     9      45
HA-02    ハーネス A-中                      5     6     6     6     6      29
HB-01    ハーネス B-細                     13    15    11    14    14      67
HB-02    ハーネス B-中                      7     8     8     7     6      36
HC-01    ハーネス C-細                      6     5     4     6     5      26

---

No.041：関数を定義する

実務での意味

不良率の計算式 defects / production * 100 を5ライン分コピーしていると、
計算ロジックを修正したとき5箇所すべてを直す必要があります。
関数（def）にまとめることで、ロジックを1か所で管理できるようになります。

分析・モデル化の考え方

関数は「DRY 原則（Don’t Repeat Yourself）」の実践手段です。
製造 KPI 計算では不良率・稼働率・サイクルタイムなどを
共通関数として定義しておくことで、ライン数や集計期間が変わっても
コードの変更が最小限に抑えられます。

Python で確認する

# 不良率を計算する関数を定義する
def defect_rate(defects, production):
    """不良率（%）を返す"""
    return defects / production * 100


# 単体での動作確認
print("=== 関数の動作確認 ===")
print(f"defect_rate(9, 450)  = {defect_rate(9, 450):.2f}%")
print(f"defect_rate(6, 381)  = {defect_rate(6, 381):.2f}%")
print()

# 5ライン一括計算（関数化の効果）
print(f"{'ライン':<8} {'週次生産':>8} {'週次不良':>8} {'不良率':>8}")
print("-" * 38)
for i in range(len(lines)):
    weekly_prod = int(prod_data[i].sum())
    weekly_defects = int(defect_data[i].sum())
    rate = defect_rate(weekly_defects, weekly_prod)
    print(f"{lines[i]:<8} {weekly_prod:>8,} {weekly_defects:>8} {rate:>7.2f}%")

=== 関数の動作確認 ===
defect_rate(9, 450)  = 2.00%
defect_rate(6, 381)  = 1.57%

ライン          週次生産     週次不良      不良率
--------------------------------------
HA-01       2,252       45    2.00%
HA-02       1,911       29    1.52%
HB-01       2,608       67    2.57%
HB-02       2,030       36    1.77%
HC-01       2,401       26    1.08%

結果の読み取り

defect_rate() を1つ定義することで、5ライン分の計算が同じ関数呼び出しで完結しました。
ラインが増えても for ループに1行追加するだけで対応できます。
この「定義は1回・呼び出しは何度でも」というのが関数化の最大のメリットです。
不良率計算式を * 100 から * 10000 に変更する必要が生じた場合も、
def defect_rate(...) の1行を直すだけで全ラインに反映されます。

---

No.042：引数を使う

実務での意味

「生産数・不良数・単位・精度」など、状況によって変える値が引数です。
デフォルト引数 を使えば「通常はこの設定、必要なときだけ変更」という
柔軟な関数が作れます。製造 KPI のレポート生成に活用できます。

分析・モデル化の考え方

引数には「位置引数」「キーワード引数」「デフォルト引数」「可変長引数（*args）」の4種類があります。
製造 KPI 関数では「単位・精度・ラベル」をデフォルト引数で設定しておくと、
標準レポートは1行で呼び出しつつ、特殊なレポートにも対応できます。

Python で確認する

# デフォルト引数・キーワード引数を使うKPIフォーマット関数
def format_kpi(label, value, unit="", precision=2, width=10):
    """KPI ラベルと値を整形して返す"""
    formatted = f"{value:.{precision}f}"
    return f"{label}: {formatted:>{width}}{unit}"


# デフォルト引数の動作確認
print(format_kpi("不良率", 2.02, unit="%"))  # デフォルト精度
print(format_kpi("週次生産", 2255, unit="個", precision=0))
print(format_kpi("稼働時間", 7.5, unit="h"))
print()


# *args で可変長引数：複数ラインの不良率から最悪ラインを特定
def worst_line_idx(*defect_rates):
    """最も不良率が高いラインのインデックスと値を返す"""
    max_rate = max(defect_rates)
    idx = list(defect_rates).index(max_rate)
    return idx, max_rate


weekly_rates = [defect_rate(int(defect_data[i].sum()), int(prod_data[i].sum())) for i in range(len(lines))]
worst_idx, worst_rate = worst_line_idx(*weekly_rates)
print(f"最高不良率ライン: {lines[worst_idx]}（{worst_rate:.2f}%）")
print()


# **kwargs で将来の拡張にも対応
def kpi_header(**metadata):
    """レポートヘッダーを生成する"""
    parts = [f"{k}={v}" for k, v in metadata.items()]
    return "[ " + " / ".join(parts) + " ]"


header = kpi_header(factory="ハーネス製造棟", week="2024-W03", shift="昼番")
print(header)

不良率:       2.02%
週次生産:       2255個
稼働時間:       7.50h

最高不良率ライン: HB-01（2.57%）

[ factory=ハーネス製造棟 / week=2024-W03 / shift=昼番 ]

結果の読み取り

format_kpi("不良率", 2.02, unit="%") は precision=2（デフォルト）で
「不良率: 2.02%」と出力されました。
*args を使うと worst_line_idx(*weekly_rates) のように
5つの数値を個別の引数として渡せます。
これにより、ライン数が変わってもコードを変更せずに最悪ラインを特定できます。
**kwargs はレポートのメタデータなど、項目数が不定の場合に便利です。

---

No.043：戻り値を使う

実務での意味

1回の計算で「不良率・良品率・サイクルタイム」の複数 KPI を同時に返したい場面は製造現場で多くあります。
Python の関数は タプルで複数の値を返せる ため、1回の呼び出しで
複数指標を一括取得できます。

分析・モデル化の考え方

戻り値をタプルで返すと「アンパッキング代入（a, b, c = func()）」が使え、
変数への格納が1行で済みます。
複数の関連指標はタプルで返すか、辞書（{}）で返すかを選択します。
辞書で返すと「結果の一部だけ使う」ときに result["defect_rate"] と
キーで取り出せるため、出力項目が多い場合は辞書が推奨されます。

Python で確認する

# 週次の主要 KPI を辞書にまとめて返す関数
def production_kpi(production_list, defect_list, unit_price, operated_hours_total):
    """
    週次主要 KPI を計算して辞書で返す

    Parameters
    ----------
    production_list      : list[int] - 日次生産数
    defect_list          : list[int] - 日次不良数
    unit_price           : int       - 単価（円/個）
    operated_hours_total : float     - 週次稼働時間（時間）

    Returns
    -------
    dict: total_prod, total_defects, good_units, defect_rate_pct,
          yield_rate_pct, cycle_time_sec, revenue
    """
    total_prod = sum(production_list)
    total_defects = sum(defect_list)
    good_units = total_prod - total_defects
    dr_pct = total_defects / total_prod * 100
    yield_pct = good_units / total_prod * 100
    cycle_sec = operated_hours_total * 3600 / total_prod
    revenue = good_units * unit_price
    return {
        "total_prod": total_prod,
        "total_defects": total_defects,
        "good_units": good_units,
        "defect_rate": dr_pct,
        "yield_rate": yield_pct,
        "cycle_time_s": cycle_sec,
        "revenue": revenue,
    }


# HA-01 の週次 KPI を計算
operated_week = 7.5 * 5  # 7.5h × 5日
kpi_ha01 = production_kpi(
    production_list=[int(x) for x in prod_data[0]],
    defect_list=[int(x) for x in defect_data[0]],
    unit_price=unit_prices[0],
    operated_hours_total=operated_week,
)

print("=== HA-01 週次 KPI ===")
print(f"  総生産数     : {kpi_ha01['total_prod']:,} 個")
print(f"  総不良数     : {kpi_ha01['total_defects']} 個")
print(f"  良品数       : {kpi_ha01['good_units']:,} 個")
print(f"  不良率       : {kpi_ha01['defect_rate']:.2f}%")
print(f"  良品率（歩留）: {kpi_ha01['yield_rate']:.2f}%")
print(f"  サイクルタイム: {kpi_ha01['cycle_time_s']:.1f} 秒/個")
print(f"  良品売上     : {kpi_ha01['revenue']:,.0f} 円")

=== HA-01 週次 KPI ===
  総生産数     : 2,252 個
  総不良数     : 45 個
  良品数       : 2,207 個
  不良率       : 2.00%
  良品率（歩留）: 98.00%
  サイクルタイム: 59.9 秒/個
  良品売上     : 2,648,400 円

結果の読み取り

辞書で戻り値を返しているので kpi_ha01["defect_rate"] のように任意の KPI だけを取り出せます。
サイクルタイム（cycle_time_s）は週次稼働時間と総生産数から自動計算されます。
たとえば {kpi_ha01['cycle_time_s']:.1f} 秒/個という数値は、
タクトタイム（計画値）との乖離を確認するための重要指標です。
関数に渡すデータを変えるだけで HB-01・HC-01 など全ラインに同じ分析が適用できます。

---

No.044：売上金額を計算する関数を作る

実務での意味

「良品売上 − 不良損失 = 純収益」という計算式は全ラインで共通です。
不良品の廃棄コスト率（penalty_rate）はラインによって異なりますが、
デフォルト引数として設定することで通常ラインは同じ式で処理できます。

分析・モデル化の考え方

コスト計算モデルとして「不良損失 = 不良数 × 単価 × (1 + 廃棄コスト率)」を使います。
廃棄コスト率は「材料費の損失」に加え「廃棄処理費・再検査費」を含めた係数です。
製造現場では penalty_rate=0.3（単価の30%追加）が一般的なベースラインです。

Python で確認する

# 良品売上・不良損失・純収益を計算する関数
def calc_revenue(production, defects, unit_price, penalty_rate=0.30):
    """
    週次の収益指標を計算する

    Parameters
    ----------
    production   : int   - 生産数
    defects      : int   - 不良数
    unit_price   : int   - 単価（円/個）
    penalty_rate : float - 不良品廃棄コスト率（デフォルト 0.30）

    Returns
    -------
    dict: good_units, revenue, loss, net_revenue, loss_ratio_pct
    """
    good_units = production - defects
    revenue = good_units * unit_price
    loss = defects * unit_price * (1 + penalty_rate)
    net_revenue = revenue - loss
    loss_ratio = loss / (revenue + loss) * 100
    return {
        "good_units": good_units,
        "revenue": revenue,
        "loss": loss,
        "net_revenue": net_revenue,
        "loss_ratio_pct": loss_ratio,
    }


# 全5ライン 週次 収益計算
results_044 = []
print(f"{'ライン':<8} {'良品売上':>12} {'不良損失':>10} {'純収益':>12} {'損失率':>8}")
print("=" * 56)
for i in range(len(lines)):
    prod_w = int(prod_data[i].sum())
    defect_w = int(defect_data[i].sum())
    r = calc_revenue(prod_w, defect_w, unit_prices[i])
    results_044.append(r)
    print(
        f"{lines[i]:<8} {r['revenue']:>12,.0f} {r['loss']:>10,.0f} "
        f"{r['net_revenue']:>12,.0f} {r['loss_ratio_pct']:>7.1f}%"
    )

total_rev = sum(r["revenue"] for r in results_044)
total_los = sum(r["loss"] for r in results_044)
total_net = sum(r["net_revenue"] for r in results_044)
print("=" * 56)
print(f"{'合計':<8} {total_rev:>12,.0f} {total_los:>10,.0f} {total_net:>12,.0f}")

ライン              良品売上       不良損失          純収益      損失率
========================================================
HA-01       2,648,400     70,200    2,578,200     2.6%
HA-02       2,540,700     50,895    2,489,805     2.0%
HB-01       2,490,180     85,358    2,404,822     3.3%
HB-02       2,193,400     51,480    2,141,920     2.3%
HC-01       2,018,750     28,730    1,990,020     1.4%
========================================================
合計         11,891,430    286,663   11,604,767

# 良品売上・不良損失・純収益のライン別棒グラフ
fig, ax = plt.subplots(figsize=(10, 5))

revenues = [r["revenue"] for r in results_044]
losses = [r["loss"] for r in results_044]
net_revenues = [r["net_revenue"] for r in results_044]

x = np.arange(len(lines))
width = 0.28

ax.bar(x - width, revenues, width, label="良品売上", color="#4C72B0", alpha=0.88)
ax.bar(x, losses, width, label="不良損失", color="#DD8452", alpha=0.88)
ax.bar(x + width, net_revenues, width, label="純収益", color="#55A868", alpha=0.88)

ax.set_title("製品ライン別 週次 良品売上・不良損失・純収益（2024年1月第3週）", fontsize=13, pad=10)
ax.set_xlabel("製品ライン", fontsize=11)
ax.set_ylabel("金額（万円）", fontsize=11)
ax.set_xticks(x)
ax.set_xticklabels(lines)
ax.yaxis.set_major_formatter(ticker.FuncFormatter(lambda v, _: f"{v/10000:.0f}万"))
ax.legend(fontsize=10)
ax.grid(axis="y", alpha=0.3)
plt.tight_layout()
plt.show()

結果の読み取り

calc_revenue() を1つ定義することで、5ラインすべての週次収益を
同じロジックで計算できました。
グラフから「良品売上が最大のライン」と「不良損失の比率が最も高いライン」を
視覚的に比較できます。
penalty_rate=0.30 を penalty_rate=0.50 に変更するだけで
「廃棄コスト見直し後の収益シミュレーション」が即座に実行できます。
これが関数化による感度分析の実践です。

---

No.045：平均値を計算する関数を作る

実務での意味

各ラインの日次不良率の「平均・最大・最小・標準偏差・変動係数」を計算することで、
製造プロセスの安定性（ばらつきの大きさ）を定量的に評価できます。
変動係数（CV = 標準偏差 / 平均） は単位が異なるラインを横並びで比較する際に有用です。

分析・モデル化の考え方

Python の標準ライブラリを使わずに sum() と四則演算だけで
平均・分散・標準偏差を実装することで、NumPy の mean() や std() が
内部で何をしているかを理解できます。
製造現場では「標準偏差が大きい = プロセスが不安定」という判断に使います。

Python で確認する

# 記述統計を計算する汎用関数（NumPy 不使用）
def calc_stats(data):
    """
    平均・最大・最小・標準偏差・変動係数を計算する

    Parameters
    ----------
    data : list[float] - 数値リスト

    Returns
    -------
    dict: mean, max_v, min_v, std, cv_pct
    """
    n = len(data)
    mean_v = sum(data) / n
    var = sum((x - mean_v) ** 2 for x in data) / n  # 母標準偏差
    std_v = var**0.5
    cv_pct = std_v / mean_v * 100  # 変動係数（%）
    return {
        "mean": mean_v,
        "max_v": max(data),
        "min_v": min(data),
        "std": std_v,
        "cv_pct": cv_pct,
    }


# 各ラインの日次不良率を計算してリストに格納
daily_defect_rates_045 = [[defect_data[i][j] / prod_data[i][j] * 100 for j in range(5)] for i in range(len(lines))]

# 統計サマリー表
print(f"{'ライン':<8} {'平均不良率':>10} {'最大':>8} {'最小':>8} {'標準偏差':>10} {'変動係数':>10}")
print("=" * 60)
for i, line in enumerate(lines):
    s = calc_stats(daily_defect_rates_045[i])
    print(
        f"{line:<8} {s['mean']:>9.2f}% {s['max_v']:>7.2f}% {s['min_v']:>7.2f}%"
        f" {s['std']:>9.3f}% {s['cv_pct']:>9.1f}%"
    )

# 全ライン合計の週次不良率
total_prod_all = int(prod_data.sum())
total_defects_all = int(defect_data.sum())
overall_dr = total_defects_all / total_prod_all * 100
print("=" * 60)
print(f"{'全体':<8} {overall_dr:>9.2f}%  (総生産 {total_prod_all:,}個 / 総不良 {total_defects_all}個)")

ライン           平均不良率       最大       最小       標準偏差       変動係数
============================================================
HA-01         2.00%    2.04%    1.94%     0.034%       1.7%
HA-02         1.52%    1.62%    1.34%     0.092%       6.1%
HB-01         2.57%    2.85%    2.14%     0.245%       9.5%
HB-02         1.77%    2.01%    1.51%     0.194%      11.0%
HC-01         1.08%    1.23%    0.86%     0.140%      13.0%
============================================================
全体            1.81%  (総生産 11,202個 / 総不良 203個)

# 日次不良率の推移グラフ（全ライン）
fig, ax = plt.subplots(figsize=(10, 5))
colors_045 = ["#4C72B0", "#DD8452", "#55A868", "#C44E52", "#8172B2"]

for i, line in enumerate(lines):
    ax.plot(
        day_labels, daily_defect_rates_045[i], marker="o", label=line, color=colors_045[i], linewidth=2, markersize=6
    )

# 警告ライン（2.0%）
ax.axhline(2.0, color="red", linewidth=1.5, linestyle="--", alpha=0.7, label="警告ライン（2.0%）")

ax.set_title("製品ライン別 日次不良率推移（2024年1月第3週）", fontsize=13, pad=10)
ax.set_xlabel("日付", fontsize=11)
ax.set_ylabel("不良率（%）", fontsize=11)
ax.legend(bbox_to_anchor=(1.02, 1), loc="upper left", fontsize=9)
ax.grid(alpha=0.3)
plt.tight_layout()
plt.show()

結果の読み取り

calc_stats() は Python の組み込み関数（sum, max, min）だけで実装した
汎用統計関数です。var ** 0.5 で標準偏差を算出しています。
変動係数（CV）が大きいラインほど日によるばらつきが大きく、
プロセスが不安定であることを示します。
グラフで警告ライン（2.0%）を超えた日付・ラインを一目で特定し、
要因分析の優先順位付けに活用できます。

---

No.046：標準ライブラリをimportする

実務での意味

Python には pip で追加インストールしなくても使える 標準ライブラリ が豊富にあります。
製造データ処理でよく使うのは math（数学）・datetime（日時）・statistics（統計）・
csv（CSV 読み書き）・os（ファイルパス）です。

分析・モデル化の考え方

import ライブラリ名 で標準ライブラリを読み込みます。
statistics モジュールは numpy なしで平均・分散・中央値を計算できます。
ただし、大量データの高速処理には numpy が必要です。
状況に応じて標準ライブラリと外部ライブラリを使い分けましょう。

Python で確認する

# 製造データ処理でよく使う標準ライブラリのimport
import math
import datetime
import statistics
import os
import sys

print("=== 標準ライブラリ確認 ===")
for name, mod in [("math", math), ("datetime", datetime), ("statistics", statistics), ("os", os), ("sys", sys)]:
    print(f"  {name:<12}: {mod.__name__}")
print()

# statistics モジュールで日次不良率を集計
all_daily_rates = [defect_data[i][j] / prod_data[i][j] * 100 for i in range(len(lines)) for j in range(5)]

print(f"statistics.mean()    : {statistics.mean(all_daily_rates):.3f}%")
print(f"statistics.median()  : {statistics.median(all_daily_rates):.3f}%")
print(f"statistics.stdev()   : {statistics.stdev(all_daily_rates):.3f}%   # 標本標準偏差")
print(f"statistics.variance(): {statistics.variance(all_daily_rates):.4f}")
print()
print(f"  [参考] No.045 の calc_stats() との比較")
s = calc_stats(all_daily_rates)
print(f"  calc_stats mean: {s['mean']:.3f}%  / std: {s['std']:.3f}%  (母標準偏差)")

=== 標準ライブラリ確認 ===


  math        : math
  datetime    : datetime
  statistics  : statistics
  os          : os
  sys         : sys

statistics.mean()    : 1.788%
statistics.median()  : 1.687%
statistics.stdev()   : 0.531%   # 標本標準偏差
statistics.variance(): 0.2819

  [参考] No.045 の calc_stats() との比較
  calc_stats mean: 1.788%  / std: 0.520%  (母標準偏差)

結果の読み取り

statistics.stdev() は標本標準偏差（分母 n-1）を使います。
一方 No.045 の calc_stats() の std は母標準偏差（分母 n）です。
サンプルが少ない場合、この差が統計的推測に影響します。
製造データは「今週の生産全体 = 母集団」と見なせる場合が多いため、
母標準偏差を使う場面もあります。
numpy の np.std() はデフォルトが母標準偏差（ddof=0）なので、
statistics.stdev() と差が出ることを覚えておきましょう。

---

No.047：mathモジュールを使う

実務での意味

製造現場での math モジュール活用例として代表的なのは
①発注数の 切り上げ計算（math.ceil）と
②統計的品質管理での 信頼区間計算（math.sqrt）です。
「発注数は少数にできない」という制約に対して math.ceil は必須です。

分析・モデル化の考え方

math.ceil(x) は x を超える最小の整数を返します。
発注数の計算で round() を使うと切り捨てられることがあり、
最低在庫を下回るリスクがあります。製造業では安全側（多め）に丸めるのが原則です。
95% 信頼区間（$\hat{p} \pm 1.96\sqrt{\hat{p}(1-\hat{p})/n}$）は
不良率の推定精度を示す統計的品質管理（SQC）の基本指標です。

Python で確認する

import math

# 発注量の切り上げ計算
min_stock_47 = 200  # 最低在庫数（個）
current_stock_47 = 143  # 現在庫数（個）
daily_usage_47 = 18.7  # 日次平均使用量（個/日）
lead_days_47 = 5  # 発注リードタイム（日）

# 発注期間中の消費量 + 最低在庫維持分
order_exact = (min_stock_47 - current_stock_47) + daily_usage_47 * lead_days_47
order_qty = math.ceil(order_exact)  # 切り上げで整数個

print(f"=== 発注量計算 ===")
print(f"  在庫不足分       : {min_stock_47 - current_stock_47} 個")
print(f"  リードタイム消費  : {daily_usage_47} × {lead_days_47} = {daily_usage_47 * lead_days_47:.1f} 個")
print(f"  理論発注数 (端数あり): {order_exact:.2f} 個")
print(f"  実際の発注数 (切上): {order_qty} 個  ← math.ceil() 適用")
print()

# 週次不良率の 95% 信頼区間
total_prod_47 = int(prod_data.sum())
total_defects_47 = int(defect_data.sum())
p_hat = total_defects_47 / total_prod_47
hw = 1.96 * math.sqrt(p_hat * (1 - p_hat) / total_prod_47)

print(f"=== 不良率 95% 信頼区間 ===")
print(f"  総生産数  : {total_prod_47:,} 個")
print(f"  総不良数  : {total_defects_47} 個")
print(f"  不良率推定値: {p_hat*100:.4f}%")
print(f"  95% CI      : [{(p_hat-hw)*100:.4f}%, {(p_hat+hw)*100:.4f}%]")
print()

# math 関数リファレンス
print("=== math 主要関数 ===")
print(f"  math.ceil(3.2)    : {math.ceil(3.2)}")
print(f"  math.floor(3.8)   : {math.floor(3.8)}")
print(f"  math.sqrt(2)      : {math.sqrt(2):.6f}")
print(f"  math.log(math.e)  : {math.log(math.e):.6f}  # 自然対数")
print(f"  math.log10(1000)  : {math.log10(1000):.6f}  # 常用対数")
print(f"  math.pi           : {math.pi:.6f}")

=== 発注量計算 ===
  在庫不足分       : 57 個
  リードタイム消費  : 18.7 × 5 = 93.5 個
  理論発注数 (端数あり): 150.50 個
  実際の発注数 (切上): 151 個  ← math.ceil() 適用

=== 不良率 95% 信頼区間 ===
  総生産数  : 11,202 個
  総不良数  : 203 個
  不良率推定値: 1.8122%
  95% CI      : [1.5652%, 2.0592%]

=== math 主要関数 ===
  math.ceil(3.2)    : 4
  math.floor(3.8)   : 3
  math.sqrt(2)      : 1.414214
  math.log(math.e)  : 1.000000  # 自然対数
  math.log10(1000)  : 3.000000  # 常用対数
  math.pi           : 3.141593

結果の読み取り

math.ceil(order_exact) で理論発注数を切り上げ、整数個の発注量を算出しました。
round() や int() では端数が切り捨てられて在庫不足になるリスクがあるため、
製造業での発注計算では math.ceil が安全 です。
不良率の 95% 信頼区間は「この範囲に真の不良率が含まれる可能性が 95%」という意味です。
生産数が多いほど CI の幅は狭くなり（推定精度が上がり）、
不良率の管理図（管理限界線）の算出にも同じ考え方が使われます。

---

No.048：datetimeモジュールを使う

実務での意味

製造現場での datetime の主な用途は
①納入予定日の計算（発注日＋リードタイム日数）
②シフト稼働時間の計算（終了時刻 − 開始時刻 − 休憩時間）
③月末・期末までの残日数確認（生産計画の残量計算）
の3つです。これらを手計算・カレンダー目視から自動化できます。

分析・モデル化の考え方

datetime.date と datetime.timedelta の2クラスを中心に
「日付の加減算」ができます。
timedelta(days=5) は「5日後」を計算するための「時間の差分オブジェクト」です。
strftime() でフォーマット文字列に変換することで、
レポートや通知メールの日付表示を自動生成できます。

Python で確認する

import datetime

# 発注日から納入予定日を計算する
order_date_48 = datetime.date(2024, 1, 15)  # 月曜日に発注
lead_days_48 = 5
delivery_date_48 = order_date_48 + datetime.timedelta(days=lead_days_48)

print("=== 発注・納入スケジュール ===")
print(f"  発注日      : {order_date_48.strftime('%Y-%m-%d (%a)')}")
print(f"  リードタイム: {lead_days_48} 日")
print(f"  納入予定日  : {delivery_date_48.strftime('%Y-%m-%d (%a)')}")
print()

# シフト稼働時間の計算
shift_start_48 = datetime.datetime(2024, 1, 15, 8, 0, 0)
shift_end_48 = datetime.datetime(2024, 1, 15, 17, 0, 0)
break_time_48 = datetime.timedelta(minutes=60)
worked_td = shift_end_48 - shift_start_48 - break_time_48
worked_hours_48 = worked_td.total_seconds() / 3600

print("=== シフト稼働時間 ===")
print(f"  シフト開始: {shift_start_48.strftime('%H:%M')}")
print(f"  シフト終了: {shift_end_48.strftime('%H:%M')}")
print(f"  休憩時間  : 60 分")
print(f"  実稼働時間: {worked_hours_48:.1f} 時間")
print()

# 月末・週末までの残日数
today_48 = datetime.date(2024, 1, 19)  # 第3週金曜日
month_end = datetime.date(2024, 1, 31)
remaining_days = (month_end - today_48).days

print("=== 月末までの残日数 ===")
print(f"  本日  : {today_48}")
print(f"  月末  : {month_end}")
print(f"  残日数: {remaining_days} 日")
print()

# 5日間の生産スケジュール日付を自動生成
print("=== 週次生産スケジュール ===")
weekday_map = {0: "月", 1: "火", 2: "水", 3: "木", 4: "金", 5: "土", 6: "日"}
for d in range(5):
    dt = order_date_48 + datetime.timedelta(days=d)
    prod_total = int(prod_data[:, d].sum())
    print(f"  {dt.strftime('%m/%d')} ({weekday_map[dt.weekday()]})  全ライン合計生産: {prod_total:,} 個")

=== 発注・納入スケジュール ===
  発注日      : 2024-01-15 (Mon)
  リードタイム: 5 日
  納入予定日  : 2024-01-20 (Sat)

=== シフト稼働時間 ===
  シフト開始: 08:00
  シフト終了: 17:00
  休憩時間  : 60 分
  実稼働時間: 8.0 時間

=== 月末までの残日数 ===
  本日  : 2024-01-19
  月末  : 2024-01-31
  残日数: 12 日

=== 週次生産スケジュール ===
  01/15 (月)  全ライン合計生産: 2,237 個
  01/16 (火)  全ライン合計生産: 2,257 個
  01/17 (水)  全ライン合計生産: 2,231 個
  01/18 (木)  全ライン合計生産: 2,241 個
  01/19 (金)  全ライン合計生産: 2,236 個

結果の読み取り

datetime.timedelta(days=5) を使うことで、発注日から5日後の納入予定日を
1行で計算できました。週末・月末をまたぐカレンダー計算も自動的に正確に処理されます。
total_seconds() / 3600 でシフト稼働時間を時間単位に変換しています。
datetime.timedelta は日・時・分・秒・マイクロ秒を扱えるので、
設備稼働ログのタイムスタンプ差分計算にも応用できます。

---

No.049：try-exceptでエラーに対応する

実務での意味

センサーからのデータやオペレータが手入力した CSV には
「数値欄に文字が入っている」「列が足りない」「負の数が入っている」などの
異常データが混入します。try-except がなければエラーが出た瞬間に処理全体が停止します。
エラーを処理しながら正常な行だけを集計し続ける のが製造データパイプラインの要件です。

分析・モデル化の考え方

try ブロック内でエラーが発生すると、対応する except ブロックに処理が移ります。
複数の except を書くことで、エラーの種類に応じた処理を分岐できます。
finally ブロックは成否に関わらず必ず実行されるため、
ファイルのクローズ処理などに使います（第6章で詳しく扱います）。

Python で確認する

# CSV の1行を安全に解析する関数
def safe_parse_row(row, row_num=0):
    """
    CSV 1行（リスト形式）を解析し、結果辞書またはエラー辞書を返す

    想定エラー
    ----------
    IndexError : 列数不足
    ValueError : 型変換失敗・値の異常
    """
    try:
        product_code = row[0]  # 品番（文字列）
        production = int(row[1])  # 生産数 → int 変換
        defects = int(row[2])  # 不良数 → int 変換
        unit_price = float(row[3])  # 単価   → float 変換

        # 業務ルールのバリデーション
        if production <= 0:
            raise ValueError(f"生産数は正の整数でなければなりません: {production}")
        if defects > production:
            raise ValueError(f"不良数({defects})が生産数({production})を超えています")

        dr = defects / production * 100
        return {
            "code": product_code,
            "production": production,
            "defects": defects,
            "unit_price": unit_price,
            "defect_rate": dr,
        }

    except IndexError:
        return {"error": f"行{row_num}: 列数不足（{len(row)} 列）", "row": row}
    except ValueError as e:
        return {"error": f"行{row_num}: 値エラー（{e}）", "row": row}


# テストデータ（正常・異常混在）
test_rows_049 = [
    ["HA-01", "450", "9", "1200"],  # ✅ 正常
    ["HA-02", "381", "6", "1350"],  # ✅ 正常
    ["HB-01", "525", "abc", "980"],  # ❌ ValueError（不良数が文字列）
    ["HB-02", "415"],  # ❌ IndexError（列不足）
    ["HC-01", "485", "520", "850"],  # ❌ ValueError（不良数 > 生産数）
    ["HC-01", "0", "0", "850"],  # ❌ ValueError（生産数 = 0）
]

print("=== CSV 行の解析結果 ===")
ok_rows, ng_rows = [], []
for i, row in enumerate(test_rows_049, start=1):
    result = safe_parse_row(row, row_num=i)
    if "error" in result:
        ng_rows.append(result)
        print(f"  ❌ {result['error']}")
    else:
        ok_rows.append(result)
        print(f"  ✅ 行{i}: {result['code']}  生産{result['production']:>4}個  " f"不良率{result['defect_rate']:.2f}%")

print()
print(f"正常処理: {len(ok_rows)} 行 / スキップ: {len(ng_rows)} 行")

=== CSV 行の解析結果 ===
  ✅ 行1: HA-01  生産 450個  不良率2.00%
  ✅ 行2: HA-02  生産 381個  不良率1.57%
  ❌ 行3: 値エラー（invalid literal for int() with base 10: 'abc'）
  ❌ 行4: 列数不足（2 列）
  ❌ 行5: 値エラー（不良数(520)が生産数(485)を超えています）
  ❌ 行6: 値エラー（生産数は正の整数でなければなりません: 0）

正常処理: 2 行 / スキップ: 4 行

結果の読み取り

safe_parse_row() は6行のテストデータを処理し、正常2行・エラー4行を正確に振り分けました。
エラーが発生した行は {"error": ..., "row": ...} として返されるため、
後でエラー一覧をレポートして手動修正を依頼できます。
except IndexError と except ValueError を分けて書いているので、
「列が足りない」と「値がおかしい」を区別したエラーログが残せます。
製造データパイプラインでは「エラーで止めない・エラーをログに残す」の2点が鉄則です。

---

No.050：データ処理で起こりやすいエラーを整理する

実務での意味

製造データ処理で起きるエラーには決まったパターンがあります。
5種類のエラー（ZeroDivisionError / KeyError / TypeError / ValueError / IndexError）を
体系的に理解し、それぞれに適切な対処法をコードとして用意しておくことで、
本番環境でのトラブル対応時間を大幅に短縮できます。

分析・モデル化の考え方

エラー処理は「防御的プログラミング」の基礎です。
「このデータに何が入ってくるか」を事前に想定し、
try-except と事前バリデーション（if 文での確認）を組み合わせることが
堅牢なデータパイプラインの設計原則です。

Python で確認する

# データ処理で頻出する5種類のエラーと対処法

error_catalog = [
    {
        "type": "ZeroDivisionError",
        "scene": "生産数0での不良率計算",
        "trigger": "defects / production * 100  (production=0)",
        "fix": "production > 0 を確認 or try-except",
    },
    {
        "type": "KeyError",
        "scene": "辞書に存在しないキーへのアクセス",
        "trigger": "part_info['supplier']  (キー未登録)",
        "fix": "part_info.get('supplier', '不明')",
    },
    {
        "type": "TypeError",
        "scene": "文字列と数値の演算",
        "trigger": "'326' + 15  (str + int)",
        "fix": "int('326') + 15  で型変換",
    },
    {
        "type": "ValueError",
        "scene": "変換できない文字列を数値に変換",
        "trigger": "int('12a')  (非数値文字列)",
        "fix": "try-except ValueError でスキップ",
    },
    {
        "type": "IndexError",
        "scene": "リストの範囲外アクセス",
        "trigger": "row[5]  (5列しかない CSV)",
        "fix": "len(row) > 5 を確認 or try-except",
    },
]

print(f"{'エラー種別':<22} {'発生場面':<28} {'対処方法'}")
print("=" * 80)
for e in error_catalog:
    print(f"{e['type']:<22} {e['scene']:<28} {e['fix']}")
print()

# 各エラーの try-except 実演
print("=== try-except 実演 ===")

# ZeroDivisionError
try:
    rate_050 = 3 / 0
except ZeroDivisionError:
    rate_050 = None
print(f"ZeroDivisionError → rate = {rate_050}  (None を設定して処理継続)")

# KeyError（get() で回避）
part_info_050 = {"code": "HA-01", "production": 450}
supplier_050 = part_info_050.get("supplier", "未登録")
print(f"KeyError 回避     → supplier = '{supplier_050}'  (get() のデフォルト値)")

# TypeError
try:
    result_050 = "326" + 15
except TypeError:
    result_050 = int("326") + 15
print(f"TypeError 回避    → result = {result_050}  (int() で変換して再計算)")

# ValueError
try:
    v_050 = int("12a")
except ValueError as e:
    v_050 = None
    e_msg_050 = str(e)
print(f"ValueError 回避   → v = {v_050}  ('{e_msg_050}' → スキップ)")

# IndexError
row_050 = ["HC-01", "480"]
try:
    defects_050 = int(row_050[2])
except IndexError:
    defects_050 = 0
print(f"IndexError 回避   → defects = {defects_050}  (列不足 → デフォルト0を設定)")

エラー種別                  発生場面                         対処方法
================================================================================
ZeroDivisionError      生産数0での不良率計算                  production > 0 を確認 or try-except
KeyError               辞書に存在しないキーへのアクセス             part_info.get('supplier', '不明')
TypeError              文字列と数値の演算                    int('326') + 15  で型変換
ValueError             変換できない文字列を数値に変換              try-except ValueError でスキップ
IndexError             リストの範囲外アクセス                  len(row) > 5 を確認 or try-except

=== try-except 実演 ===
ZeroDivisionError → rate = None  (None を設定して処理継続)
KeyError 回避     → supplier = '未登録'  (get() のデフォルト値)
TypeError 回避    → result = 341  (int() で変換して再計算)
ValueError 回避   → v = None  ('invalid literal for int() with base 10: '12a'' → スキップ)
IndexError 回避   → defects = 0  (列不足 → デフォルト0を設定)

結果の読み取り

5種類のエラーとその対処法を一覧で整理しました。
実際のデータパイプラインでは「エラーを返す」「デフォルト値を使う」「スキップしてログに残す」の
3つの選択肢から状況に応じて選びます。

対応方針	使いどころ
None を設定して処理継続	集計処理で欠損として扱える場合
デフォルト値（get() など）	キーや列が任意項目の場合
except でスキップ + ログ	異常行を除外して正常行だけ集計する場合
raise で再送出	呼び出し元にエラーを通知すべき重大な異常の場合

「どのエラーを黙殺し、どのエラーを通知するか」の設計が
製造データパイプラインの品質を左右します。

---

対象ノックを通して見える実務上の示唆

No.041〜050 で学んだ「関数・モジュール・例外処理」は、
製造 KPI システムの 設計品質 を決める基礎スキルです。

1. 関数化 = 「メンテナンスコストの削減」

5ラインぶん同じ計算を並べたコードは、仕様変更のたびに5箇所を修正する必要があります。
calc_revenue() のような関数1つに集約することで、
仕様変更を1箇所の修正で全ラインに反映できます。
製造現場では「ラインが増える / KPI の定義が変わる」ことが頻繁にあるため、
関数化は変化への耐性を高める設計です。

2. 標準ライブラリ = 「インストール不要の生産性向上ツール」

math.ceil による発注数の切り上げや
datetime.timedelta による納入日計算は、
追加ライブラリなしで即座に使えます。
「ツールのインストール申請が通らない」という製造 IT 環境でも
標準ライブラリで多くの業務自動化が実現できます。

3. 例外処理 = 「データパイプラインの品質保証」

センサーデータ・手入力 CSV には必ず異常値が混入します。
try-except なしのシステムは、異常1件で処理全体が停止します。
safe_parse_row() のような関数を実装することで、
「99% の正常データを止めずに処理し続け、
異常データだけをログに蓄積して後から確認する」という
製造データパイプラインの理想的な動作が実現できます。

実務導入する場合に必要なこと

本章で学んだ内容を製造現場のデータシステムに導入する際の手順を示します。

Step 1: KPI 関数ライブラリの作成（約1日）

defect_rate(), calc_revenue(), calc_stats() など本章で作成した関数を
kpi_functions.py として1ファイルにまとめます。
他の notebook・スクリプトから from kpi_functions import calc_revenue で再利用できます。

Step 2: datetime によるスケジュール自動化（約半日）

発注日・納入予定日・月末残日数の計算を datetime 関数で実装し、
毎日の定常業務（発注アラート・スケジュール確認）を自動化します。

Step 3: 安全な CSV 読み込みパイプラインの構築（約1日）

safe_parse_row() を使って「異常行をスキップしながら正常行だけを集計する」
読み込み関数を構築します。
第6章（ファイル操作と CSV）では実際の CSV ファイルを読み込んでこのパイプラインを完成させます。

Step 4: エラーログの出力（約半日）

except ブロックで検出した異常行を
error_log.csv などのファイルに書き出す仕組みを追加します。
これにより「どの行が・どのエラーで・なぜスキップされたか」を
データ担当者が後から確認できるようになります。

まとめ

本章（No.041〜050）で学んだ内容を整理します。

No.	スキル	製造現場での活用
041	関数を定義する	不良率計算の共通関数化・5ライン一括計算
042	引数を使う	デフォルト引数でKPIフォーマット、*argsで最悪ライン特定
043	戻り値を使う	辞書で複数KPI（不良率・良品率・サイクルタイム・売上）を返す
044	売上金額を計算する関数を作る	5ラインの週次収益一括計算・棒グラフ可視化
045	平均値を計算する関数を作る	日次不良率の記述統計・変動係数・週次推移グラフ
046	標準ライブラリをimportする	math・datetime・statistics のインポートと使い分け
047	mathモジュールを使う	発注数の切り上げ・不良率の 95% 信頼区間計算
048	datetimeモジュールを使う	発注日→納入予定日・シフト稼働時間・残日数計算
049	try-exceptでエラーに対応する	CSV 異常データを処理しながら正常行だけ集計
050	データ処理で起こりやすいエラーを整理する	5種類のエラーパターンと対処法の体系的整理

第6章（No.051〜060）ではファイル操作と CSV モジュールを使って、
実際のファイルから本章の関数・例外処理を組み合わせた
本格的なデータパイプラインを構築します。

法人向けのご相談

---

数理工房 では、製造業・DX 推進担当者向けの Python 研修・データ分析支援を提供しています。

こんなお悩みはありませんか？

• 「本章の KPI 関数を自社の製品ライン・指標に合わせてカスタマイズしたい」
• 「CSV で止まるデータパイプラインを堅牢な処理フローに作り直したい」
• 「発注計算・スケジュール管理を Python で自動化したい」
• 「社内研修で実際の現場データを使って Python を教えてほしい」

提供サービス

サービス	概要
製造業向け Python 研修	現場データを使った実践型研修（オンライン/対面）
KPI 自動集計システム開発	不良率・稼働率・収益計算の自動化パイプライン
データクレンジング基盤構築	センサー・CSV データの安全な前処理フロー
DX 推進コンサルティング	課題整理から実装・定着支援まで一貫対応

---

📩 お問い合わせ: surikobo.co.jp/contact
まずはお気軽にご相談ください。