LoginSignup
4
5

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@BNR-Longin株式会社BinaryTech | 日本品質 × ベトナム開発、最高のバランスを。

💥倉庫崩壊を阻止！🚚失敗しないWMS開発の極意 | 第6回: 多チャネル販売とEコマース統合

Last updated at Posted at 2025-05-15

はじめに

現代の小売業では、オンライン（Shopify、Amazon）とオフライン（実店舗）の販売チャネルを統合する多チャネル戦略が不可欠です。倉庫管理システム（WMS）は、これらのチャネル間で在庫を一元管理し、データ同期を確保する必要があります。特に、OMO（Online-Merge-Offline）モデルでは、オンラインとオフラインの在庫をシームレスに連携させ、顧客体験を向上させます。第6回では、WMSの多チャネル管理とEコマースプラットフォーム（例：Shopify）およびPOS（Point of Sale）システムとの統合に焦点を当て、データ同期、API設計、衝突回避について、具体的なPython（Flask）とPostgreSQLのコード例とともに解説します。さらに、実際の運用での教訓も共有します。

多チャネル販売とOMOの課題

多チャネル販売では、オンライン（Eコマース）とオフライン（店舗）の在庫を同期させることが求められます。OMOモデルでは、顧客がオンラインで注文し、店舗で受け取る（BOPIS: Buy Online, Pick Up In-Store）といったシナリオをサポートする必要があります。主な課題は以下の通りです：

  • データ同期の遅延：オンライン販売が店舗の在庫に即座に反映されない場合、オーバーセルが発生。
  • 在庫管理の複雑さ：各チャネルで異なる在庫割り当てや優先順位を管理。
  • POS統合：店舗のPOSシステムとWMSの在庫データをリアルタイムで同期。
  • 衝突管理：同一商品がオンラインとオフラインで同時に販売される場合のデータ不整合。

筆者の経験では、ある小売企業でデータ同期が10分遅延した結果、店舗とオンラインで同一商品が重複販売され、月間50件の返金対応が発生しました。このような失敗を避けるため、EコマースとPOSの統合が重要です。

技術要件

多チャネル販売をサポートするWMSの技術要件は以下の通りです：

  • API統合：Eコマースプラットフォーム（例：Shopify API）とPOSシステムに対応。
  • データ同期：リアルタイムまたは準リアルタイムでの在庫・注文データ同期。
  • POS機能：店舗での販売を記録し、WMSに即座に反映。
  • 衝突回避：楽観的ロックやトランザクションでデータ不整合を防止。
  • 高可用性：ピーク時の負荷（例：セール期間）に対応。

統合アーキテクチャ

多チャネル統合の一般的なアーキテクチャは以下の通りです：

  1. Eコマース統合：ShopifyのWebhookで注文データを受信、WMSで在庫を更新。
  2. POS統合：店舗のPOS端末からAPI経由で販売データを送信、WMSに反映。
  3. キュー：非同期処理（例：Celery）で同期タスクを管理し、ピーク時の負荷を軽減。
  4. データベース：第2回のスキーマ（products, stocks）を活用し、チャネルごとの在庫を追跡。
  5. 監視：同期エラーを検出し、アラートを送信。

バックエンド実装（Python/Flask）

以下は、Shopifyの注文WebhookとPOS販売データを処理し、在庫を更新するFlaskベースのAPIエンドポイントの例です。第2回のデータベーススキーマを前提としています。

from flask import Flask, request, jsonify
from http import HTTPStatus
import psycopg2
from psycopg2.extras import RealDictCursor
from celery import Celery

app = Flask(__name__)
celery = Celery(app.name, broker='redis://localhost:6379/0')

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

# 在庫更新タスク（非同期）
@celery.task
def update_inventory(sku, quantity, location_code, channel):
    try:
        conn = get_db_connection()
        cursor = conn.cursor(cursor_factory=RealDictCursor)
        
        cursor.execute("SELECT id FROM products WHERE sku = %s", (sku,))
        product = cursor.fetchone()
        if not product:
            return {'error': f'商品 {sku} が見つかりません'}
        
        cursor.execute("SELECT id FROM locations WHERE code = %s", (location_code,))
        location = cursor.fetchone()
        if not location:
            return {'error': f'ロケーション {location_code} が見つかりません'}
        
        cursor.execute(
            """
            SELECT quantity FROM stocks
            WHERE product_id = %s AND location_id = %s
            FOR UPDATE
            """,
            (product['id'], location['id'])
        )
        stock = cursor.fetchone()
        
        new_quantity = (stock['quantity'] if stock else 0) - quantity
        if new_quantity < 0:
            return {'error': f'{channel} で在庫が不足しています'}
        
        cursor.execute(
            """
            UPDATE stocks
            SET quantity = %s
            WHERE product_id = %s AND location_id = %s
            """,
            (new_quantity, product['id'], location['id'])
        )
        
        cursor.execute(
            """
            INSERT INTO transactions (product_id, location_id, quantity, type, channel)
            VALUES (%s, %s, %s, %s, %s)
            """,
            (product['id'], location['id'], -quantity, 'OUT', channel)
        )
        
        conn.commit()
        return {'message': f'{channel} の在庫を更新しました', 'sku': sku, 'quantity': new_quantity}
    
    except Exception as e:
        conn.rollback()
        return {'error': str(e)}
    
    finally:
        cursor.close()
        conn.close()

# Shopify Webhookエンドポイント
@app.route('/api/webhook/shopify/order', methods=['POST'])
def shopify_order_webhook():
    data = request.get_json()
    order_items = data.get('line_items', [])
    
    if not order_items:
        return jsonify({'error': '注文アイテムがありません'}), HTTPStatus.BAD_REQUEST
    
    # 非同期で在庫更新
    for item in order_items:
        sku = item.get('sku')
        quantity = item.get('quantity')
        location_code = 'A-01-01'  # 仮のロケーション（実際はロジックで決定）
        update_inventory.delay(sku, quantity, location_code, 'shopify')
    
    return jsonify({'message': '注文を受信しました'}), HTTPStatus.OK

# POS販売エンドポイント
@app.route('/api/pos/sale', methods=['POST'])
def pos_sale():
    data = request.get_json()
    items = data.get('items', [])  # [{sku, quantity}, ...]
    store_code = data.get('store_code', 'STORE001')
    
    if not items:
        return jsonify({'error': '販売アイテムがありません'}), HTTPStatus.BAD_REQUEST
    
    # 非同期で在庫更新
    for item in items:
        sku = item.get('sku')
        quantity = item.get('quantity')
        location_code = store_code  # 店舗をロケーションとして扱う
        update_inventory.delay(sku, quantity, location_code, 'pos')
    
    return jsonify({'message': 'POS販売を記録しました'}), HTTPStatus.OK

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

コードのポイント

  1. データ同期：Shopify WebhookとPOS販売データを共通のupdate_inventoryタスクで処理。
  2. 非同期処理：Celeryで在庫更新をキューイングし、ピーク時の負荷を軽減。
  3. データ整合性：FOR UPDATEロックで同時更新による衝突を防止。
  4. チャネル追跡：トランザクションにchannel（例：shopify, pos）を記録。

APIの使用例

Shopify Webhook（注文データ）：

curl -X POST http://localhost:5000/api/webhook/shopify/order \
-H "Content-Type: application/json" \
-d '{"line_items": [{"sku": "TSHIRT001", "quantity": 2}, {"sku": "JEANS001", "quantity": 1}]}'

POS販売（店舗販売データ）：

curl -X POST http://localhost:5000/api/pos/sale \
-H "Content-Type: application/json" \
-d '{"store_code": "STORE001", "items": [{"sku": "TSHIRT001", "quantity": 1}, {"sku": "JEANS001", "quantity": 2}]}'

レスポンス例：

{
    "message": "POS販売を記録しました"
}

POSモジュールの実装（React）

以下は、店舗スタッフ向けの簡易POSインターフェースをReactで構築した例です。販売データを記録し、WMSに送信します。

import React, { useState } from 'react';

const POSForm = () => {
    const [storeCode, setStoreCode] = useState('STORE001');
    const [items, setItems] = useState([]);
    const [sku, setSku] = useState('');
    const [quantity, setQuantity] = useState('');
    const [message, setMessage] = useState('');

    // アイテム追加
    const handleAddItem = () => {
        if (sku && quantity) {
            setItems([...items, { sku, quantity: parseInt(quantity) }]);
            setSku('');
            setQuantity('');
        }
    };

    // 販売送信
    const handleSale = async () => {
        try {
            const response = await fetch('http://localhost:5000/api/pos/sale', {
                method: 'POST',
                headers: { 'Content-Type': 'application/json' },
                body: JSON.stringify({ store_code: storeCode, items })
            });
            const data = await response.json();
            if (response.ok) {
                setMessage(data.message);
                setItems([]);
            } else {
                setMessage(data.error);
            }
        } catch (error) {
            setMessage('エラーが発生しました');
        }
    };

    return (
        <div style={{ padding: '20px', maxWidth: '400px', margin: 'auto' }}>
            <h2>POS販売</h2>
            <div>
                <label>店舗コード:</label>
                <input
                    type="text"
                    value={storeCode}
                    onChange={(e) => setStoreCode(e.target.value)}
                    style={{ width: '100%', padding: '10px', fontSize: '16px' }}
                />
            </div>
            <div>
                <label>SKU:</label>
                <input
                    type="text"
                    value={sku}
                    onChange={(e) => setSku(e.target.value)}
                    style={{ width: '100%', padding: '10px', fontSize: '16px' }}
                />
            </div>
            <div>
                <label>数量:</label>
                <input
                    type="number"
                    value={quantity}
                    onChange={(e) => setQuantity(e.target.value)}
                    style={{ width: '100%', padding: '10px', fontSize: '16px' }}
                />
            </div>
            <button
                onClick={handleAddItem}
                style={{ width: '100%', padding: '15px', fontSize: '18px', margin: '10px 0' }}
            >
                アイテム追加
            </button>
            <h3>販売アイテム</h3>
            <ul>
                {items.map((item, index) => (
                    <li key={index}>SKU: {item.sku}, 数量: {item.quantity}</li>
                ))}
            </ul>
            <button
                onClick={handleSale}
                style={{ width: '100%', padding: '15px', fontSize: '18px', margin: '10px 0' }}
            >
                販売送信
            </button>
            {message && <p style={{ color: message.includes('エラー') ? 'red' : 'green' }}>{message}</p>}
        </div>
    );
};

export default POSForm;

コードのポイント

  1. ユーザビリティ：シンプルなフォームで店舗スタッフ dễ dàng nhập SKU và số lượng.
  2. データ同期：販売データを即座にWMSに送信し、在庫を更新。
  3. エラーハンドリング：無効なSKUや在庫不足を検出し、エラーメッセージを表示。
  4. モバイルフレンドリー：大きなボタンと入力欄で、タブレットやPOS端末に対応。

データ衝突の回避

多チャネル環境では、同一商品がオンラインとオフラインで同時に販売される可能性があります。衝突を防ぐための戦略は以下の通りです：

  1. 楽観的ロック：在庫更新時にバージョンチェックを行う。 
    UPDATE stocks
SET quantity = %s, version = version + 1
WHERE product_id = %s AND location_id = %s AND version = %s
  2. バッファ在庫：各チャネルに仮想的な在庫割り当てを設定。 
    CREATE TABLE channel_buffers (
    id SERIAL PRIMARY KEY,
    channel_name VARCHAR(50) NOT NULL,
    product_id INTEGER REFERENCES products(id),
    buffer_quantity INTEGER NOT NULL CHECK (buffer_quantity >= 0)
);
  3. トランザクションログ：すべての在庫変更をtransactionsテーブルに記録し、衝突の原因を追跡。
  4. 同期頻度制限：Shopify APIやPOSデータの同期を5秒ごとに制限し、過負荷を防止。

以下は、衝突を検出するクエリ例：

SELECT t1.sku, t1.created_at, t1.quantity, t1.channel
FROM transactions t1
JOIN transactions t2 ON t1.product_id = t2.product_id
WHERE t1.channel != t2.channel
AND ABS(EXTRACT(EPOCH FROM (t1.created_at - t2.created_at))) < 5
AND t1.type = 'OUT' AND t2.type = 'OUT';

実際のユースケース

以下は、多チャネル統合とOMOが倉庫業務にどのように役立つかの例です：

  1. OMO小売：
    • 課題：オンライン注文が店舗在庫に反映されず、BOPIS失敗率が20%。
    • 解決策：POSとShopify APIを統合し、リアルタイム同期を導入。
    • 結果：BOPIS成功率が95%に向上。
  2. 店舗販売：
    • 課題：店舗のPOSがWMSと非同期で、オーバーセルが月間100件。
    • 解決策：データ同期を5秒間隔に設定し、楽観的ロックを適用。
    • 結果：オーバーセルがゼロに。
  3. ピーク時対応：
    • 課題：セール期間にShopify APIが制限に達し、同期停止。
    • 解決策：Celeryでタスクをキューイングし、同期頻度を制限。
    • 結果：同期エラーが90%削減。

実践のポイント

  • データ同期の最適化：リアルタイム同期はコストが高いため、POSデータは即時、Eコマースデータは5秒間隔で同期。
  • API制限の考慮：Shopify APIのレート制限（例：2リクエスト/秒）を遵守し、バックオフ戦略を導入。
  • ユーザビリティ：POSインターフェースは店舗スタッフが1分以内に操作できるように設計。
  • エラーログ：同期エラーや衝突をSentryで監視し、Slackでアラート。
  • フィードバック収集：店舗スタッフとEコマースチームに同期遅延や在庫割り当ての課題を確認。

学びのポイント

多チャネルは顧客体験の鍵：OMOモデルでは、データ同期の遅延が顧客信頼を損ないます。筆者のプロジェクトでは、POSとWMSの同期を軽視した結果、店舗で在庫切れが頻発し、顧客クレームが月間200件に達しました。Webhook、Celery、楽観的ロックを導入することで、同期の信頼性が99.9%に向上し、クレームがほぼゼロになりました。店舗スタッフとEコマースチームのフィードバックを定期的に収集し、ユーザビリティを高めることが成功の鍵です。

次回予告

次回は、棚卸と在庫精度の向上に焦点を当てます。定期棚卸の自動化や、在庫差異を最小化する手法について、PythonとPostgreSQLのコード例とともに解説します。

4
5
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
4
5

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?