📊 WMSパワーアップ！倉庫運用の効率と利益を最大化 | 第5回: 棚卸の自動化と在庫精度向上

## はじめに
在庫の不正確さは、倉庫運用の最大の敵です。誤った在庫データは、注文遅延や販売機会の損失を引き起こします。**WMS**（倉庫管理システム）は、**バーコード**や**RFID**を活用した**棚卸**を自動化し、**在庫精度**を劇的に向上させます。本シリーズの第5回では、**WMS**を使ったサイクルカウントやフルカウントを効率化し、リアルタイムで在庫を監視する方法を解説します。Flask APIで**棚卸**データを処理、Reactと`quaggaJS`でモバイルスキャンアプリを構築、Pythonで在庫差異を検出するコードを紹介します。目標は、棚卸時間を70%削減し、在庫精度を90%から99%に引き上げることです。

## なぜ棚卸が重要か
不正確な在庫は、以下のような問題を引き起こします：
- **注文ミス**：在庫があるはずの商品が欠品、月間100件のキャンセル。
- **コスト増**：手動棚卸で作業員が1日10時間費やす。
- **顧客不満**：欠品による遅延でクレームが月間30件。
- **過剰在庫**：不正確なデータで発注ミス、月間50万円の無駄。

筆者の経験では、ある倉庫が年1回のフルカウントのみ実施した結果、在庫精度が80%にとどまり、売上機会損失が月間200万円発生しました。**WMS**によるサイクルカウントと**自動化**を導入後、精度が99%に向上し、損失がほぼゼロになりました。

## 技術要件
**棚卸**を最適化するための技術要件は以下の通りです：
- **データ収集**：**バーコード**または**RFID**で在庫をスキャン。
- **リアルタイム処理**：モバイルデバイスでWMSと同期。
- **差異検出**：WMSデータと実在庫の不一致を自動検出。
- **データベース**：`PostgreSQL`で在庫と棚卸ログを管理。
- **スケーラビリティ**：1日数千SKUを処理。

## 棚卸アーキテクチャ
以下は、**WMS**を使った棚卸のアーキテクチャ概要です：
1. **データベース**：`PostgreSQL`（`stocks`, `locations`, `inventory_logs`テーブル）。
2. **バックエンド**：`Flask`で棚卸データ処理APIを提供。
3. **スクリプト**：`Python`で在庫差異を検出。
4. **フロントエンド**：`React`と`quaggaJS`でモバイルスキャンアプリを構築。
5. **デバイス**：**バーコード**スキャナーまたは**RFID**リーダー。

## 棚卸データ処理API
以下は、モバイルデバイスからのスキャンデータを処理するFlask APIです。

```python
from flask import Flask, request, jsonify
from http import HTTPStatus
import psycopg2
from psycopg2.extras import RealDictCursor
from datetime import datetime

app = Flask(__name__)

def get_db_connection():
    return psycopg2.connect(
        dbname="wms_db",
        user="postgres",
        password="password",
        host="localhost",
        port="5432"
    )

@app.route('/api/inventory/scan', methods=['POST'])
def process_inventory_scan():
    data = request.get_json()
    scan = data.get('scan', {})  # {sku, location_code, quantity, scan_type}
    
    if not all([scan.get('sku'), scan.get('location_code'), scan.get('quantity'), scan.get('scan_type')]):
        return jsonify({'error': '必要なフィールドが不足しています'}), HTTPStatus.BAD_REQUEST
    
    try:
        conn = get_db_connection()
        cursor = conn.cursor(cursor_factory=RealDictCursor)
        
        # 商品特定
        cursor.execute(
            """
            SELECT id FROM products
            WHERE sku = %s
            """,
            (scan['sku'],)
        )
        product = cursor.fetchone()
        if not product:
            return jsonify({'error': '商品が見つかりません'}), HTTPStatus.NOT_FOUND
        
        # ロケーション特定
        cursor.execute(
            """
            SELECT id FROM locations
            WHERE code = %s
            """,
            (scan['location_code'],)
        )
        location = cursor.fetchone()
        if not location:
            return jsonify({'error': 'ロケーションが見つかりません'}), HTTPStatus.NOT_FOUND
        
        # スキャンデータ記録
        cursor.execute(
            """
            INSERT INTO inventory_logs (product_id, location_id, quantity, scan_type, scanned_at)
            VALUES (%s, %s, %s, %s, %s)
            RETURNING id
            """,
            (
                product['id'], location['id'], scan['quantity'],
                scan['scan_type'], datetime.now()
            )
        )
        log_id = cursor.fetchone()['id']
        
        # 在庫差異チェック
        cursor.execute(
            """
            SELECT quantity FROM stocks
            WHERE product_id = %s AND location_id = %s
            """,
            (product['id'], location['id'])
        )
        stock = cursor.fetchone()
        discrepancy = (stock['quantity'] - scan['quantity']) if stock else scan['quantity']
        
        if discrepancy != 0:
            cursor.execute(
                """
                INSERT INTO discrepancies (inventory_log_id, discrepancy, reported_at)
                VALUES (%s, %s, %s)
                """,
                (log_id, discrepancy, datetime.now())
            )
        
        conn.commit()
        cursor.close()
        conn.close()
        
        return jsonify({
            'message': 'スキャンデータを処理しました',
            'log_id': log_id,
            'discrepancy': discrepancy
        }), HTTPStatus.OK
    
    except Exception as e:
        conn.rollback()
        return jsonify({'error': str(e)}), HTTPStatus.INTERNAL_SERVER_ERROR

if __name__ == '__main__':
    app.run(debug=True)
```

### コードのポイント
1. **スキャンデータ**：**バーコード**または**RFID**からのSKU、ロケーション、数量を処理。
2. **差異検出**：WMS在庫とスキャンデータの不一致を記録。
3. **データ整合性**：棚卸ログと差異を`inventory_logs`と`discrepancies`に保存。
4. **エラーハンドリング**：無効なSKUやロケーションを検出。

### APIの使用例
リクエスト：
```bash
curl -X POST http://localhost:5000/api/inventory/scan \
-H "Content-Type: application/json" \
-d '{"scan": {"sku": "TSHIRT001", "location_code": "A-01-01", "quantity": 50, "scan_type": "BARCODE"}}'
```

レスポンス例：
```json
{
    "message": "スキャンデータを処理しました",
    "log_id": 2001,
    "discrepancy": -5
}
```

## 在庫差異検出スクリプト
以下は、在庫差異を分析する`Python`スクリプトです。

```python
import pandas as pd
import psycopg2

# データベース接続
def get_db_connection():
    return psycopg2.connect(
        dbname="wms_db",
        user="postgres",
        password="password",
        host="localhost",
        port="5432"
    )

# 在庫差異分析
def analyze_discrepancies():
    conn = get_db_connection()
    query = """
        SELECT il.id, p.sku, l.code as location_code, il.quantity as scanned_quantity,
               COALESCE(s.quantity, 0) as wms_quantity,
               COALESCE(s.quantity, 0) - il.quantity as discrepancy
        FROM inventory_logs il
        JOIN products p ON il.product_id = p.id
        JOIN locations l ON il.location_id = l.id
        LEFT JOIN stocks s ON il.product_id = s.product_id AND il.location_id = s.location_id
        WHERE il.scanned_at >= CURRENT_DATE - INTERVAL '7 days'
    """
    df = pd.read_sql(query, conn)
    conn.close()
    
    # 差異のあるレコードを抽出
    discrepancies = df[df['discrepancy'] != 0]
    
    # 集計
    summary = discrepancies.groupby('sku').agg({
        'discrepancy': ['sum', 'count'],
        'location_code': 'first'
    }).reset_index()
    
    # 結果をCSVに保存
    summary.to_csv('discrepancy_report.csv', index=False)
    return summary

if __name__ == '__main__':
    report = analyze_discrepancies()
    print(report)
```

### コードのポイント
1. **差異検出**：過去7日の棚卸ログとWMS在庫を比較。
2. **集計**：SKUごとの差異合計と頻度を計算。
3. **出力**：CSVレポートを生成、管理者向け。
4. **拡張性**：数千SKUを効率的に処理。

### レポートの活用
- **大きな差異**：SKU TSHIRT001で-50個→即時調査。
- **頻発差異**：特定のロケーション（例：A-01-01）で複数SKUに差異→作業員トレーニング。
- **傾向分析**：差異が多いSKUを特定、発注やピッキングプロセスを改善。

## モバイルスキャンアプリ
以下は、`React`と`quaggaJS`でバーコードスキャンを行うモバイルアプリの例です。

```jsx
import React, { useEffect, useState } from 'react';
import Quagga from 'quagga';
import axios from 'axios';

const InventoryScanApp = () => {
    const [scanData, setScanData] = useState({ sku: '', location_code: '', quantity: 0 });
    const [result, setResult] = useState(null);

    useEffect(() => {
        Quagga.init({
            inputStream: { name: 'Live', type: 'LiveStream', target: document.querySelector('#scanner') },
            decoder: { readers: ['ean_reader', 'code_128_reader'] }
        }, err => {
            if (err) {
                console.error('Quaggaエラー:', err);
                return;
            }
            Quagga.start();
        });

        Quagga.onDetected(data => {
            setScanData(prev => ({ ...prev, sku: data.codeResult.code }));
        });

        return () => Quagga.stop();
    }, []);

    const handleSubmit = async () => {
        try {
            const response = await axios.post('http://localhost:5000/api/inventory/scan', {
                scan: { ...scanData, scan_type: 'BARCODE' }
            });
            setResult(response.data);
        } catch (error) {
            setResult({ error: error.message });
        }
    };

    return (
        <div>
            <h1>在庫スキャンアプリ</h1>
            <div id="scanner" style={{ width: '100%', height: '300px' }} />
            <div>
                <label>SKU: </label>
                <input
                    value={scanData.sku}
                    onChange={e => setScanData({ ...scanData, sku: e.target.value })}
                />
            </div>
            <div>
                <label>ロケーションコード: </label>
                <input
                    value={scanData.location_code}
                    onChange={e => setScanData({ ...scanData, location_code: e.target.value })}
                />
            </div>
            <div>
                <label>数量: </label>
                <input
                    type="number"
                    value={scanData.quantity}
                    onChange={e => setScanData({ ...scanData, quantity: parseInt(e.target.value) })}
                />
            </div>
            <button onClick={handleSubmit}>スキャンデータ送信</button>
            {result && (
                <div>
                    <p>結果: {result.message}</p>
                    {result.discrepancy !== 0 && <p>差異: {result.discrepancy}個</p>}
                </div>
            )}
        </div>
    );
};

export default InventoryScanApp;
```

### アプリのポイント
1. **バーコードスキャン**：`quaggaJS`でリアルタイムスキャン。
2. **ユーザビリティ**：作業員がSKU、ロケーション、数量を簡単に入力。
3. **リアルタイム**：スキャンデータをAPIに即送信、差異を即表示。
4. **拡張性**：**RFID**リーダーにも対応可能。

## 実際のユースケース
以下は、**棚卸**自動化が倉庫業務にどう役立つかの例です：
1. **時間削減**：
   - 課題：フルカウントに1週間。
   - 解決策：サイクルカウントとモバイルアプリで毎日処理。
   - 結果：棚卸時間が70%削減（1週から2日）。
2. **在庫精度向上**：
   - 課題：精度90%、月間50件の欠品。
   - 解決策：**バーコード**スキャンと差異検出。
   - 結果：精度が99%に向上、欠品ゼロ。
3. **コスト削減**：
   - 課題：手動棚卸で人件費が月間30万円。
   - 解決策：**自動化**で作業員を半減。
   - 結果：人件費が50%削減。

## 実践のポイント
- **データ品質**：スキャンデータの正確性を確保（例：バーコードの読み取り精度）。
- **フィードバック**：作業員にアプリの使い勝手を聞き、UIを改善。
- **サイクルカウント**：高価値SKUを週次、低価値SKUを月次でカウント。
- **RFID活用**：大量在庫（例：1000SKU以上）の倉庫でRFIDを優先。
- **定期分析**：差異レポートを月次でレビュー、原因を特定。

## 学びのポイント
**在庫精度は信頼の基盤**：正確な**棚卸**がなければ、どんな**WMS**も機能しません。筆者のプロジェクトでは、手動棚卸により在庫差異が月間200件発生し、売上損失が100万円でした。**バーコード**スキャンとサイクルカウントを導入後、差異が10件に減り、顧客信頼度が20%向上しました。鍵は、リアルタイムのデータ収集と継続的な改善です。

## 次回予告
次回は、**容量管理**に焦点を当てます。**WMS**を活用して倉庫スペースを最適化し、需要予測を行う方法を、具体的なコード例で解説します。