# Inventory Management: Track Stock Levels

# Inventory Management: Track Stock Levels & Prevent Overselling

## Problem

E-commerce businesses face critical challenges in inventory management:
- **Overselling**: Selling products that are out of stock, leading to unfulfilled orders
- **Stock Discrepancies**: Manual tracking causes inaccuracies between actual and recorded inventory
- **Real-time Visibility**: Multiple sales channels create confusion about true stock levels
- **Inefficient Reordering**: No automated alerts for low stock situations
- **Lost Revenue**: Inability to fulfill orders damages customer trust and revenue
- **Operational Chaos**: Without proper tracking, warehouse operations become disorganized

These issues compound during peak seasons, flash sales, or when managing multiple warehouses and sales channels simultaneously.

## Solution

Implement a comprehensive inventory management system with:

1. **Real-time Stock Tracking**: Maintain accurate, up-to-date inventory counts across all locations
2. **Overselling Prevention**: Lock inventory when orders are placed, preventing double-selling
3. **Multi-channel Synchronization**: Centralize inventory across website, marketplace, and physical stores
4. **Automated Alerts**: Trigger notifications when stock falls below thresholds
5. **Transaction Logging**: Maintain audit trails for all inventory movements
6. **Reservation System**: Reserve stock for pending orders without immediate deduction
7. **Reorder Management**: Automate purchase orders based on minimum stock levels
8. **Analytics Dashboard**: Visualize inventory health and trends

## Code

### 1. Core Inventory Model & Database Schema

```python
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
import json

class TransactionType(Enum):
    PURCHASE = "purchase"
    SALE = "sale"
    RETURN = "return"
    ADJUSTMENT = "adjustment"
    RESTOCK = "restock"
    DAMAGE = "damage"

@dataclass
class InventoryItem:
    sku: str
    name: str
    quantity: int
    reserved: int
    warehouse_id: str
    reorder_level: int
    reorder_quantity: int
    unit_cost: float
    last_updated: datetime

@dataclass
class Transaction:
    transaction_id: str
    sku: str
    transaction_type: TransactionType
    quantity: int
    timestamp: datetime
    reference_id: Optional[str]
    notes: str

class InventoryDatabase:
    """Simulates database operations for inventory"""
    
    def __init__(self):
        self.inventory: Dict[str, InventoryItem] = {}
        self.transactions: List[Transaction] = []
        self.reservations: Dict[str, int] = {}  # order_id -> quantity
    
    def add_product(self, item: InventoryItem):
        """Add new product to inventory"""
        self.inventory[item.sku] = item
        print(f"✓ Product added: {item.sku} - {item.name}")
    
    def get_available_stock(self, sku: str) -> int:
        """Get available stock (total - reserved)"""
        if sku not in self.inventory:
            return 0
        item = self.inventory[sku]
        return item.quantity - item.reserved
    
    def log_transaction(self, transaction: Transaction):
        """Log all inventory movements"""
        self.transactions.append(transaction)
    
    def get_transaction_history(self, sku: str, days: int = 30) -> List[Transaction]:
        """Retrieve transaction history for audit trail"""
        cutoff_date = datetime.now() - timedelta(days=days)
        return [t for t in self.transactions 
                if t.sku == sku and t.timestamp >= cutoff_date]
```

### 2. Overselling Prevention System

```python
import uuid
from threading import Lock

class OversellPreventionEngine:
    """Prevents overselling through reservation and locking mechanisms"""
    
    def __init__(self, db: InventoryDatabase):
        self.db = db
        self.lock = Lock()  # Thread-safe operations
    
    def check_availability(self, sku: str, quantity: int) -> Dict:
        """Check if product is available for purchase"""
        with self.lock:
            if sku not in self.db.inventory:
                return {
                    "available": False,
                    "reason": "Product not found",
                    "available_quantity": 0
                }
            
            item = self.db.inventory[sku]
            available = self.db.get_available_stock(sku)
            
            return {
                "available": available >= quantity,
                "available_quantity": available,
                "requested_quantity": quantity,
                "total_stock": item.quantity,
                "reserved_stock": item.reserved,
                "reason": "OK" if available >= quantity else "Insufficient stock"
            }
    
    def reserve_stock(self, order_id: str, sku: str, quantity: int) -> Dict:
        """Reserve stock for an order (prevents overselling)"""
        with self.lock:
            availability = self.check_availability(sku, quantity)
            
            if not availability["available"]:
                return {
                    "success": False,
                    "order_id": order_id,
                    "reason": availability["reason"],
                    "reservation_id": None
                }
            
            # Reserve the stock
            self.db.inventory[sku].reserved += quantity
            self.db.reservations[order_id] = quantity
            
            # Log transaction
            transaction = Transaction(
                transaction_id=str(uuid.uuid4()),
                sku=sku,
                transaction_type=TransactionType.PURCHASE,
                quantity=quantity,
                timestamp=datetime.now(),
                reference_id=order_id,
                notes=f"Stock reserved for order {order_id}"
            )
            self.db.log_transaction(transaction)
            
            return {
                "success": True,
                "order_id": order_id,
                "sku": sku,
                "reserved_quantity": quantity,
                "reservation_id": str(uuid.uuid4()),
                "timestamp": datetime.now().isoformat()
            }
    
    def confirm_sale(self, order_id: str, sku: str) -> Dict:
        """Confirm sale and deduct from actual inventory"""
        with self.lock:
            if order_id not in self.db.reservations:
                return {"success": False, "reason": "Reservation not found"}
            
            quantity = self.db.reservations[order_id]
            item = self.db.inventory[sku]
            
            # Deduct from inventory
            item.quantity -= quantity
            item.reserved -= quantity
            item.last_updated = datetime.now()
            
            # Log transaction
            transaction = Transaction(
                transaction_id=str(uuid.uuid4()),
                sku=sku,
                transaction_type=TransactionType.SALE,
                quantity=quantity,
                timestamp=datetime.now(),
                reference_id=order_id,
                notes=f"Sale confirmed for order {order_id}"
            )
            self.db.log_transaction(transaction)
            
            del self.db.reservations[order_id]
            
            return {
                "success": True,
                "order_id": order_id,
                "sku": sku,
                "quantity_sold": quantity,
                "remaining_stock": item.quantity
            }
    
    def cancel_reservation(self, order_id: str, sku: str) -> Dict:
        """Cancel reservation and release stock"""
        with self.lock:
            if order_id not in self.db.reservations:
                return {"success": False, "reason": "Reservation not found"}
            
            quantity = self.db.reservations[order_id]
            self.db.inventory[sku].reserved -= quantity
            
            transaction = Transaction(
                transaction_id=str(uuid.uuid4()),
                sku=sku,
                transaction_type=TransactionType.RETURN,
                quantity=quantity,
                timestamp=datetime.now(),
                reference_id=order_id,
                notes=f"Reservation cancelled for order {order_id}"
            )
            self.db.log_transaction(transaction)
            
            del self.db.reservations[order_id]
            
            return {
                "success": True,
                "order_id": order_id,
                "released_quantity": quantity
            }
```

### 3. Stock Level Monitoring & Alerts

```python
from enum import Enum as AlertLevel

class AlertSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"

@dataclass
class Alert:
    alert_id: str
    sku: str
    severity: AlertSeverity
    message: str
    timestamp: datetime
    resolved: bool = False

class StockMonitor:
    """Monitor stock levels and trigger alerts"""
    
    def __init__(self, db: InventoryDatabase):
        self.db = db
        self.alerts: List[Alert] = []
    
    def check_stock_levels(self) -> List[Alert]:
        """Check all products for low stock conditions"""
        new_alerts = []
        
        for sku, item in self.db.inventory.items():
            available = self.db.get_available_stock(sku)
            
            # Critical: Below reorder level
            if available <= item.reorder_level:
                alert = Alert(
                    alert_id=str(uuid.uuid4()),
                    sku=sku,
                    severity=AlertSeverity.CRITICAL,
                    message=f"Stock critically low: {available} units (reorder level: {item.reorder_level})",
                    timestamp=datetime.now()
                )
                new_alerts.append(alert)
            
            # Warning: Below 50% of reorder level
            elif available <= item.reorder_level * 1.5:
                alert = Alert(
                    alert_id=str(uuid.uuid4()),
                    sku=sku,
                    severity=AlertSeverity.WARNING,
                    message=f"Stock running low: {available} units",
                    timestamp=datetime.now()
                )
                new_alerts.append(alert)
            
            # Info: Out of stock
            if item.quantity == 0:
                alert = Alert(
                    alert_id=str(uuid.uuid4()),
                    sku=sku,
                    severity=AlertSeverity.CRITICAL,
                    message=f"Product out of stock: {item.name}",
                    timestamp=datetime.now()
                )
                new_alerts.append(alert)
        
        self.alerts.extend(new_alerts)
        return new_alerts
    
    def get_active_alerts(self) -> List[Alert]:
        """Get unresolved alerts"""
        return [a for a in self.alerts if not a.resolved]
    
    def resolve_alert(self, alert_id: str):
        """Mark alert as resolved"""
        for alert in self.alerts:
            if alert.alert_id == alert_id:
                alert.resolved = True
                break
```

### 4. Automated Reordering System

```python
@dataclass
class PurchaseOrder:
    po_id: str
    sku: str
    quantity: int
    supplier_id: str
    status: str  # pending, ordered, received
    created_date: datetime
    expected_delivery: datetime

class ReorderManager:
    """Automate reordering based on stock levels"""
    
    def __init__(self, db: InventoryDatabase):
        self.db = db
        self.purchase_orders: List[PurchaseOrder] = []
    
    def generate_reorder_suggestions(self) -> List[Dict]:
        """Generate reorder suggestions for low stock items"""
        suggestions = []
        
        for sku, item in self.db.inventory.items():
            available = self.db.get_available_stock(sku)
            
            if available <= item.reorder_level:
                # Calculate order quantity
                order_qty = item.reorder_quantity
                
                # Check if already ordered
                pending_orders = [po for po in self.purchase_orders 
                                if po.sku == sku and po.status == "pending"]
                
                if not pending_orders:
                    suggestions.append({
                        "sku": sku,
                        "product_name": item.name,
                        "current_stock": available,
                        "reorder_level": item.reorder_level,
                        "suggested_quantity": order_qty,
                        "unit_cost": item.unit_cost,
                        "total_cost": order_qty * item.unit_cost,
                        "urgency": "CRITICAL" if available == 0 else "HIGH"
                    })
        
        return suggestions
    
    def create_purchase_order(self, sku: str, quantity: int, 
                            supplier_id: str, lead_time_days: int = 7) -> Dict:
        """Create a purchase order"""
        po = PurchaseOrder(
            po_id=f"PO-{str(uuid.uuid4())[:8]}",
            sku=sku,
            quantity=quantity,
            supplier_id=supplier_id,
            status="pending",
            created_date=datetime.now(),
            expected_delivery=datetime.now() + timedelta(days=lead_time_days)
        )
        
        self.purchase_orders.append(po)
        
        return {
            "po_id": po.po_id,
            "sku": sku,
            "quantity": quantity,
            "status": "pending",
            "expected_delivery": po.expected_delivery.isoformat()
        }
    
    def receive_purchase_order(self, po_id: str) -> Dict:
        """Receive goods from purchase order"""
        po = next((p for p in self.purchase_orders if p.po_id == po_id), None)
        
        if not po:
            return {"success": False, "reason": "PO not found"}
        
        # Update inventory
        item = self.db.inventory[po.sku]
        item.quantity += po.quantity
        item.last_updated = datetime.now()
        
        # Log transaction
        transaction = Transaction(
            transaction_id=str(uuid.uuid4()),
            sku=po.sku,
            transaction_type=TransactionType.RESTOCK,
            quantity=po.quantity,
            timestamp=datetime.now(),
            reference_id=po.po_id,
            notes=f"Received PO {po.po_id} from supplier {po.supplier_id}"
        )
        self.db.log_transaction(transaction)
        
        po.status = "received"
        
        return {
            "success": True,
            "po_id": po_id,
            "sku": po.sku,
            "quantity_received": po.quantity,
            "new_stock_level": item.quantity
        }
```

### 5. Analytics & Reporting Dashboard

```python
from collections import defaultdict

class InventoryAnalytics:
    """Generate insights and reports on inventory health"""
    
    def __init__(self, db: InventoryDatabase):
        self.db = db
    
    def get_inventory_summary(self) -> Dict:
        """Overall inventory health snapshot"""
        total_items = len(self.db.inventory)
        total_quantity = sum(item.quantity for item in self.db.inventory.values())
        total_reserved = sum(item.reserved for item in self.db.inventory.values())
        total_value = sum(item.quantity * item.unit_cost 
                         for item in self.db.inventory.values())
        
        out_of_stock = sum(1 for item in self.db.inventory.values() 
                          if item.quantity == 0)
        low_stock = sum(1 for item in self.db.inventory.values() 
                       if 0 < item.quantity <= item.reorder_level)
        
        return {
            "total_products": total_items,
            "total_quantity": total_quantity,
            "total_reserved": total_reserved,
            "available_quantity": total_quantity - total_reserved,
            "total_inventory_value": f"${total_value:,.2f}",
            "out_of_stock_count": out_of_stock,
            "low_stock_count": low_stock,
            "healthy_stock_count": total_items - out_of_stock - low_stock
        }
    
    def get_product_performance(self, sku: str, days: int = 30) -> Dict:
        """Analyze product sales and stock movement"""
        item = self.db.inventory.get(sku)
        if not item:
            return {"error": "Product not found"}
        
        transactions = self.db.get_transaction_history(sku, days)
        
        sales = sum(t.quantity for t in transactions 
                   if t.transaction_type == TransactionType.SALE)
        returns = sum(t.quantity for t in transactions 
                     if t.transaction_type == TransactionType.RETURN)
        restocks = sum(t.quantity for t in transactions 
                      if t.transaction_type == TransactionType.RESTOCK)
        
        return {
            "sku": sku,
            "product_name": item.name,
            "current_stock": item.quantity,
            "reserved": item.reserved,
            "available": self.db.get_available_stock(sku),
            "period_days": days,
            "sales_volume": sales,
            "returns_volume": returns,
            "restocks_volume": restocks,
            "net_movement": sales - returns + restocks,
            "turnover_rate": sales / item.quantity if item.quantity > 0 else 0,
            "inventory_value": item.quantity * item.unit_cost
        }
    
    def get_slow_moving_products(self, threshold_days: int = 90) -> List[Dict]:
        """Identify products with low sales velocity"""
        slow_movers = []
        
        for sku, item in self.db.inventory.items():
            transactions = self.db.get_transaction_history(sku, threshold_days)
            sales = sum(t.quantity for t in transactions 
                       if t.transaction_type == TransactionType.SALE)
            
            if sales == 0 and item.quantity > 0:
                slow_movers.append({
                    "sku": sku,
                    "product_name": item.name,
                    "stock_quantity": item.quantity,
                    "inventory_value": item.quantity * item.unit_cost,
                    "days_without_sales": threshold_days,
                    "recommendation": "Consider discount or discontinuation"
                })
        
        return sorted(slow_movers, 
                     key=lambda x: x["inventory_value"], 
                     reverse=True)
    
    def forecast_stockout(self) -> List[Dict]:
        """Predict which products will run out soon"""
        forecasts = []
        
        for sku, item in self.db.inventory.items():
            # Get last 30 days of sales
            transactions = self.db.get_transaction_history(sku, 30)
            sales = sum(t.quantity for t in transactions 
                       if t.transaction_type == TransactionType.SALE)
            
            if sales > 0:
                daily_rate = sales / 30
                days_until_stockout = item.quantity / daily_rate if daily_rate > 0 else float('inf')
                
                if days_until_stockout < 14:  # Less than 2 weeks
                    forecasts.append({
                        "sku": sku,
                        "product_name": item.name,
                        "current_stock": item.quantity,
                        "daily_sales_rate": round(daily_rate, 2),
                        "estimated_days_until_stockout": roun