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

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

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

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

@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