ในอุตสาหกรรมค้าปลีก การจัดการสินค้าคงคลังให้เหมาะสมเป็นความท้าทายสำคัญ เราเคยประสบปัญหา ConnectionError: timeout ขณะเรียก API คาดการณ์ ในช่วง Peak Season ทำให้ระบบหยุดทำงาน 48 ชั่วโมง และสินค้าขาดสต็อกจนเสียโอกาสทางธุรกิจไปหลายล้านบาท บทความนี้จะสอนวิธีสร้างระบบคาดการณ์สินค้าคงคลังด้วย Time Series + LLM ที่เชื่อถือได้และรวดเร็ว

ทำไมต้องใช้ Time Series + LLM สำหรับ Inventory Forecasting

การคาดการณ์สินค้าคงคลังแบบดั้งเดิมใช้เพียงสถิติ แต่ไม่สามารถวิเคราะห์ปัจจัยภายนอก เช่น ฤดูกาล เทศกาล หรือเหตุการณ์พิเศษได้ การผสมผสาน Time Series Model กับ LLM ช่วยให้:

สถาปัตยกรรมระบบ Inventory AI

"""
ระบบ Inventory Forecasting ด้วย Time Series + LLM
สถาปัตยกรรม: Data Pipeline → Time Series Model → LLM Analysis
"""

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import requests
import json
from typing import Dict, List, Optional

class InventoryForecastingSystem:
    """ระบบคาดการณ์สินค้าคงคลังแบบครบวงจร"""
    
    def __init__(self, api_key: str):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def fetch_sales_data(self, product_id: str, days: int = 90) -> pd.DataFrame:
        """
        ดึงข้อมูลยอดขายย้อนหลัง
        ใช้ API จากระบบ POS หรือ ERP
        """
        # ตัวอย่าง: ดึงข้อมูลจาก API
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days)
        
        # สมมติว่ามี API สำหรับดึงข้อมูลยอดขาย
        data = {
            "product_id": product_id,
            "start_date": start_date.isoformat(),
            "end_date": end_date.isoformat()
        }
        
        # สำหรับ Demo ใช้ข้อมูลสมมติ
        dates = pd.date_range(start=start_date, end=end_date, freq='D')
        np.random.seed(42)
        base_demand = 50
        trend = np.linspace(0, 10, len(dates))
        seasonal = 15 * np.sin(2 * np.pi * np.arange(len(dates)) / 30)
        noise = np.random.normal(0, 5, len(dates))
        
        sales_data = pd.DataFrame({
            'date': dates,
            'product_id': product_id,
            'quantity_sold': np.maximum(0, base_demand + trend + seasonal + noise).astype(int),
            'revenue': np.maximum(0, base_demand + trend + seasonal + noise).astype(int) * 299
        })
        
        return sales_data
    
    def calculate_time_series_features(self, df: pd.DataFrame) -> Dict:
        """คำนวณคุณลักษณะจากอนุกรมเวลา"""
        quantity = df['quantity_sold'].values
        
        features = {
            'mean_daily_sales': float(np.mean(quantity)),
            'std_daily_sales': float(np.std(quantity)),
            'min_daily_sales': int(np.min(quantity)),
            'max_daily_sales': int(np.max(quantity)),
            'trend_coefficient': float(np.polyfit(range(len(quantity)), quantity, 1)[0]),
            'seasonality_amplitude': float(15),  # จากการคำนวณข้างต้น
            'recent_7day_avg': float(np.mean(quantity[-7:])),
            'recent_30day_avg': float(np.mean(quantity[-30:])),
        }
        
        # คำนวณ Growth Rate
        if np.mean(quantity[:30]) > 0:
            features['growth_rate'] = float(
                (np.mean(quantity[-30:]) - np.mean(quantity[:30])) / np.mean(quantity[:30])
            )
        else:
            features['growth_rate'] = 0.0
            
        return features
    
    def forecast_demand(self, features: Dict, forecast_days: int = 30) -> Dict:
        """คาดการณ์ความต้องการสินค้าล่วงหน้า"""
        
        base_forecast = features['recent_7day_avg']
        trend_adj = features['trend_coefficient']
        seasonality = features['seasonality_amplitude']
        
        daily_forecasts = []
        for day in range(forecast_days):
            day_in_cycle = day % 30
            seasonal_factor = seasonality * np.sin(2 * np.pi * day_in_cycle / 30)
            forecast = base_forecast + (trend_adj * day) + seasonal_factor
            daily_forecasts.append({
                'day': day + 1,
                'forecasted_demand': int(max(0, forecast)),
                'confidence_lower': int(max(0, forecast - 2 * features['std_daily_sales'])),
                'confidence_upper': int(forecast + 2 * features['std_daily_sales'])
            })
        
        return {
            'total_forecast_30days': sum(d['forecasted_demand'] for d in daily_forecasts),
            'daily_forecasts': daily_forecasts,
            'features': features
        }
    
    def analyze_with_llm(self, features: Dict, forecast: Dict, product_name: str) -> str:
        """
        ใช้ LLM วิเคราะห์ข้อมูลและสร้างรายงาน
        ใช้ HolySheep API สำหรับ DeepSeek V3.2 ที่ราคาถูกและเร็ว
        """
        prompt = f"""วิเคราะห์ข้อมูลการคาดการณ์สินค้าคงคลังสำหรับ: {product_name}

ข้อมูลสถิติ:
- ยอดขายเฉลี่ยต่อวัน: {features['mean_daily_sales']:.1f} ชิ้น
- ความผันผวน (Std): {features['std_daily_sales']:.1f}
- แนวโน้ม (Trend): {features['trend_coefficient']:.2f} ต่อวัน
- อัตราการเติบโต: {features['growth_rate']*100:.1f}%
- ค่าเฉลี่ย 7 วันล่าสุด: {features['recent_7day_avg']:.1f} ชิ้น
- ค่าเฉลี่ย 30 วันล่าสุด: {features['recent_30day_avg']:.1f} ชิ้น

การคาดการณ์ 30 วันข้างหน้า:
- ความต้องการรวม: {forecast['total_forecast_30days']} ชิ้น
- ความต้องการเฉลี่ยต่อวัน: {forecast['total_forecast_30days']/30:.1f} ชิ้น

กรุณาวิเคราะห์:
1. แนวโน้มการขาย (ขาขึ้น/ขาลง/คงที่)
2. ปัจจัยที่ต้องจับตา (Seasonality, Promotion ที่ใกล้มา)
3. คำแนะนำสำหรับการสั่งซื้อสินค้า
4. คำเตือนความเสี่ยง (ถ้ามี)
5. ข้อความสรุปสำหรับผู้บริหาร 3 บรรทัด

ตอบเป็นภาษาไทย ใช้รายงานที่เป็นมืออาชีพ"""
        
        payload = {
            "model": "deepseek-v3.2",
            "messages": [
                {"role": "system", "content": "คุณเป็นผู้เชี่ยวชาญด้านการคาดการณ์สินค้าคงคลังและการวิเคราะห์ธุรกิจค้าปลีก"},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.3,
            "max_tokens": 800
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=self.headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            result = response.json()
            return result['choices'][0]['message']['content']
        except requests.exceptions.Timeout:
            raise ConnectionError("LLM API Timeout - ลองใช้ model ที่เล็กกว่า")
        except requests.exceptions.RequestException as e:
            if response.status_code == 401:
                raise PermissionError("API Key ไม่ถูกต้อง - ตรวจสอบ HolySheep API Key")
            raise Exception(f"API Error: {str(e)}")
    
    def generate_reorder_recommendation(self, forecast: Dict, current_stock: int, 
                                        lead_time_days: int = 7, safety_stock: int = 50) -> Dict:
        """สร้างคำแนะนำการสั่งซื้อสินค้าใหม่"""
        
        daily_avg_demand = forecast['total_forecast_30days'] / 30
        demand_during_leadtime = daily_avg_demand * lead_time_days
        total_needed = demand_during_leadtime + safety_stock
        recommended_order = max(0, total_needed - current_stock)
        
        # คำนวณ reorder point
        reorder_point = (daily_avg_demand * lead_time_days) + safety_stock
        
        return {
            "current_stock": current_stock,
            "reorder_point": int(reorder_point),
            "safety_stock": safety_stock,
            "recommended_order_quantity": int(recommended_order),
            "days_until_reorder": int(current_stock / daily_avg_demand) if daily_avg_demand > 0 else 999,
            "urgency": "สูง" if current_stock < reorder_point else "ปกติ",
            "estimated_cost": int(recommended_order * 299)  # ราคาต่อหน่วย
        }



ตัวอย่างการใช้งาน

if __name__ == "__main__":
    # ตั้งค่า API Key จาก HolySheep
    API_KEY = "YOUR_HOLYSHEEP_API_KEY"
    
    system = InventoryForecastingSystem(API_KEY)
    
    # ดึงข้อมูลยอดขายสินค้า SKU-12345
    product_id = "SKU-12345"
    product_name = "เสื้อยืด Cotton Premium สีดำ"
    sales_data = system.fetch_sales_data(product_id, days=90)
    
    print(f"📊 ดึงข้อมูลยอดขาย {len(sales_data)} วัน สำเร็จ")
    
    # คำนวณคุณลักษณะ Time Series
    features = system.calculate_time_series_features(sales_data)
    print(f"📈 คุณลักษณะ: ยอดขายเฉลี่ย {features['mean_daily_sales']:.1f} ชิ้น/วัน")
    
    # คาดการณ์ความต้องการ 30 วัน
    forecast = system.forecast_demand(features, forecast_days=30)
    print(f"🔮 คาดการณ์ 30 วัน: {forecast['total_forecast_30days']} ชิ้น")
    
    # วิเคราะห์ด้วย LLM
    analysis = system.analyze_with_llm(features, forecast, product_name)
    print(f"\n📝 ผลวิเคราะห์จาก AI:\n{analysis}")
    
    # คำแนะนำการสั่งซื้อ
    current_stock = 200
    recommendation = system.generate_reorder_recommendation(
        forecast, current_stock, lead_time_days=7, safety_stock=50
    )
    print(f"\n🛒 คำแนะนำสั่งซื้อ: {recommendation['recommended_order_quantity']} ชิ้น")
    print(f"   ความเร่งด่วน: {recommendation['urgency']}")

การใช้ LLM ตอบคำถามเกี่ยวกับ Inventory แบบ Interactive

"""
Chatbot สำหรับตอบคำถามเกี่ยวกับสินค้าคงคลัง
ใช้ RAG (Retrieval Augmented Generation) กับข้อมูล Inventory
"""

import json
from datetime import datetime

class InventoryChatbot:
    """Chatbot สำหรับถามตอบเกี่ยวกับสินค้าคงคลัง"""
    
    def __init__(self, api_key: str):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        self.inventory_context = {}
    
    def update_inventory_context(self, inventory_data: Dict):
        """อัปเดตข้อมูลสินค้าคงคลังล่าสุด"""
        self.inventory_context = {
            "timestamp": datetime.now().isoformat(),
            "total_products": len(inventory_data),
            "low_stock_items": [],
            "top_selling": []
        }
        
        # หาสินค้าใกล้หมด
        for item_id, item in inventory_data.items():
            if item.get('stock', 0) < item.get('reorder_point', 0):
                self.inventory_context['low_stock_items'].append({
                    'product_id': item_id,
                    'name': item.get('name', 'Unknown'),
                    'current_stock': item.get('stock', 0),
                    'reorder_point': item.get('reorder_point', 0)
                })
    
    def ask(self, question: str, context: Dict) -> str:
        """ถามคำถามเกี่ยวกับสินค้าคงคลัง"""
        
        # สร้าง Context สำหรับ RAG
        inventory_summary = json.dumps(context, indent=2, ensure_ascii=False)
        
        prompt = f"""คุณเป็นผู้ช่วยฝ่ายจัดการสินค้าคงคลัง กรุณาตอบคำถามต่อไปนี้ตามข้อมูลที่ให้มา

ข้อมูลสินค้าคงคลังปัจจุบัน:
{inventory_summary}

คำถาม: {question}

กฎการตอบ:
1. ตอบเป็นภาษาไทยเท่านั้น
2. ถ้าตอบไม่ได้ ให้บอกว่า "ไม่มีข้อมูลในระบบ"
3. ถ้าเกี่ยวกับการสั่งซื้อ ให้แนะนำ quantity ที่เหมาะสม
4. ถ้าสินค้าใกล้หมด ให้แจ้งเตือนความเร่งด่วน
5. ตอบกระชับ ใช้ตารางถ้ามีหลายรายการ"""

        payload = {
            "model": "deepseek-v3.2",
            "messages": [
                {"role": "system", "content": "ผู้ช่วยจัดการสินค้าคงคลังที่เป็นมืออาชีพ"},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.2,
            "max_tokens": 500
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code == 401:
            raise PermissionError("❌ API Key ไม่ถูกต้อง กรุณาตรวจสอบที่ https://www.holysheep.ai/register")
        
        response.raise_for_status()
        return response.json()['choices'][0]['message']['content']
    
    def generate_daily_report(self, inventory_data: Dict) -> str:
        """สร้างรายงานประจำวันอัตโนมัติ"""
        
        total_value = sum(
            item.get('stock', 0) * item.get('unit_cost', 0) 
            for item in inventory_data.values()
        )
        low_stock_count = sum(
            1 for item in inventory_data.values() 
            if item.get('stock', 0) < item.get('reorder_point', 0)
        )
        
        report_prompt = f"""สร้างรายงานสินค้าคงคลังประจำวัน {datetime.now().strftime('%d/%m/%Y')}

สรุป:
- จำนวน SKU ทั้งหมด: {len(inventory_data)}
- มูลค่าสินค้าคงคลังรวม: ฿{total_value:,.0f}
- สินค้าใกล้หมด (ต่ำกว่า Reorder Point): {low_stock_count} รายการ

รายละเอียดสินค้าใกล้หมด:
{json.dumps(self.inventory_context.get('low_stock_items', [])[:5], indent=2, ensure_ascii=False)}

กรุณาสร้างรายงานที่ประกอบด้วย:
1. สรุปสถานะโดยรวม (ดี/ปานกลาง/ต้องแก้ไข)
2. รายการสินค้าเร่งด่วนที่ต้องสั่งซื้อ
3. คำแนะนำสำหรับวันนี้
4. ความเสี่ยงที่ต้องจับตามอง

ตอบเป็นภาษาไทย กระชับ เข้าใจง่าย"""

        payload = {
            "model": "gpt-4.1",
            "messages": [
                {"role": "user", "content": report_prompt}
            ],
            "temperature": 0.3,
            "max_tokens": 600
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload,
            timeout=45
        )
        
        return response.json()['choices'][0]['message']['content']



ตัวอย่างการใช้งาน Chatbot

def demo_inventory_chatbot():
    API_KEY = "YOUR_HOLYSHEEP_API_KEY"
    chatbot = InventoryChatbot(API_KEY)
    
    # ข้อมูลสินค้าคงคลังตัวอย่าง
    inventory = {
        "SKU-001": {"name": "เสื้อยืด Cotton Premium", "stock": 45, "reorder_point": 100, "unit_cost": 150},
        "SKU-002": {"name": "กางเกงยีนส์ Slim Fit", "stock": 200, "reorder_point": 80, "unit_cost": 450},
        "SKU-003": {"name": "รองเท้าสนีคเกอร์ Running", "stock": 12, "reorder_point": 50, "unit_cost": 890},
        "SKU-004": {"name": "กระเป๋า Backpack Urban", "stock": 75, "reorder_point": 60, "unit_cost": 650},
    }
    
    # อัปเดต Context
    chatbot.update_inventory_context(inventory)
    
    # ถามคำถาม
    questions = [
        "มีสินค้าตัวไหนที่ต้องสั่งซื้อด่วนบ้าง?",
        "ถ้าต้องการสั่งสินค้า SKU-003 ควรสั่งกี่ชิ้น?",
        "สรุปสถานะสินค้าคงคลังวันนี้เป็นอย่างไร?"
    ]
    
    for q in questions:
        print(f"\n❓ คำถาม: {q}")
        answer = chatbot.ask(q, inventory)
        print(f"💬 คำตอบ: {answer}")
    
    # สร้างรายงานประจำวัน
    print("\n" + "="*50)
    print("📋 รายงานประจำวัน")
    print("="*50)
    report = chatbot.generate_daily_report(inventory)
    print(report)


if __name__ == "__main__":
    demo_inventory_chatbot()

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. ConnectionError: Timeout ขณะเรียก LLM API

# ❌ วิธีผิด: ไม่มีการจัดการ Timeout
response = requests.post(url, json=payload)  # ค้างถ้า server ตอบช้า


✅ วิธีถูก: เพ


แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง