smartstock.py

# -*- coding: utf-8 -*-
"""SmartStock.ipynb

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/drive/1MhfsNfM1s8yX1_TYYa3wm5qFxPcBXiNl
"""

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from sklearn.linear_model import LinearRegression
from typing import Dict, List, Optional
import json
import warnings
warnings.filterwarnings('ignore')

class InventoryManagementSystem:
    def __init__(self):
        """Initialize the Inventory Management System with empty data structures"""
        # Core inventory data
        self.inventory = pd.DataFrame(columns=[
            'product_id', 'product_name', 'category', 'quantity',
            'unit_price', 'reorder_point', 'supplier_id', 'min_order_qty'
        ])

        # Sales history
        self.sales_history = pd.DataFrame(columns=[
            'transaction_id', 'product_id', 'quantity', 'sale_date', 'sale_price'
        ])

        # Supplier information
        self.suppliers = pd.DataFrame(columns=[
            'supplier_id', 'name', 'lead_time_days', 'reliability_score'
        ])

        # Purchase orders
        self.purchase_orders = pd.DataFrame(columns=[
            'po_id', 'product_id', 'supplier_id', 'quantity',
            'order_date', 'expected_delivery', 'status'
        ])

    def add_product(self, product_data: Dict) -> bool:
        """
        Add a new product to inventory

        Args:
            product_data (dict): Product information including name, category, etc.
        Returns:
            bool: Success status
        """
        try:
            new_product = pd.DataFrame([product_data])
            self.inventory = pd.concat([self.inventory, new_product], ignore_index=True)
            return True
        except Exception as e:
            print(f"Error adding product: {e}")
            return False

    def record_sale(self, sale_data: Dict) -> bool:
        """
        Record a new sale transaction

        Args:
            sale_data (dict): Sale transaction details
        Returns:
            bool: Success status
        """
        try:
            # Record the sale
            new_sale = pd.DataFrame([sale_data])
            self.sales_history = pd.concat([self.sales_history, new_sale], ignore_index=True)

            # Update inventory
            product_id = sale_data['product_id']
            quantity = sale_data['quantity']
            self.inventory.loc[self.inventory['product_id'] == product_id, 'quantity'] -= quantity

            # Check if reorder is needed
            self._check_reorder_point(product_id)
            return True
        except Exception as e:
            print(f"Error recording sale: {e}")
            return False

    def predict_demand(self, product_id: int, days_ahead: int = 30) -> float:
        """
        Predict demand for a product using simple linear regression

        Args:
            product_id (int): Product identifier
            days_ahead (int): Number of days to forecast
        Returns:
            float: Predicted demand
        """
        try:
            # Get historical sales data
            product_sales = self.sales_history[
                self.sales_history['product_id'] == product_id
            ].copy()

            if len(product_sales) < 5:  # Need minimum data points
                return None

            # Prepare data for prediction
            product_sales['sale_date'] = pd.to_datetime(product_sales['sale_date'])
            daily_sales = product_sales.groupby('sale_date')['quantity'].sum().reset_index()

            # Create features (days since first sale)
            first_date = daily_sales['sale_date'].min()
            daily_sales['days_since_start'] = (daily_sales['sale_date'] - first_date).dt.days

            # Train model
            model = LinearRegression()
            X = daily_sales['days_since_start'].values.reshape(-1, 1)
            y = daily_sales['quantity'].values
            model.fit(X, y)

            # Predict future demand
            future_days = np.array(range(
                daily_sales['days_since_start'].max() + 1,
                daily_sales['days_since_start'].max() + days_ahead + 1
            )).reshape(-1, 1)

            predictions = model.predict(future_days)
            return max(0, predictions.mean())  # Ensure non-negative prediction

        except Exception as e:
            print(f"Error predicting demand: {e}")
            return None

    def _check_reorder_point(self, product_id: int) -> None:
        """
        Check if product needs reordering and create purchase order if necessary

        Args:
            product_id (int): Product identifier
        """
        try:
            product = self.inventory[self.inventory['product_id'] == product_id].iloc[0]

            if product['quantity'] <= product['reorder_point']:
                # Calculate optimal order quantity
                predicted_demand = self.predict_demand(product_id)
                if predicted_demand is None:
                    order_quantity = product['min_order_qty']
                else:
                    order_quantity = max(
                        product['min_order_qty'],
                        int(predicted_demand * 1.2)  # 20% buffer
                    )

                # Create purchase order
                po_data = {
                    'po_id': len(self.purchase_orders) + 1,
                    'product_id': product_id,
                    'supplier_id': product['supplier_id'],
                    'quantity': order_quantity,
                    'order_date': datetime.now().strftime('%Y-%m-%d'),
                    'expected_delivery': (datetime.now() + timedelta(days=7)).strftime('%Y-%m-%d'),
                    'status': 'pending'
                }
                new_po = pd.DataFrame([po_data])
                self.purchase_orders = pd.concat([self.purchase_orders, new_po], ignore_index=True)

        except Exception as e:
            print(f"Error checking reorder point: {e}")

    def generate_report(self) -> Dict:
        """
        Generate a summary report of inventory status

        Returns:
            dict: Summary statistics and alerts
        """
        try:
            report = {
                'total_products': len(self.inventory),
                'low_stock_items': len(self.inventory[
                    self.inventory['quantity'] <= self.inventory['reorder_point']
                ]),
                'pending_orders': len(self.purchase_orders[
                    self.purchase_orders['status'] == 'pending'
                ]),
                'total_inventory_value': (
                    self.inventory['quantity'] * self.inventory['unit_price']
                ).sum(),
                'alerts': []
            }

            # Generate alerts for low stock items
            low_stock = self.inventory[
                self.inventory['quantity'] <= self.inventory['reorder_point']
            ]
            for _, product in low_stock.iterrows():
                report['alerts'].append(
                    f"Low stock alert: {product['product_name']} "
                    f"(Quantity: {product['quantity']})"
                )

            return report

        except Exception as e:
            print(f"Error generating report: {e}")
            return None

# Example usage and demonstration
def demo_system():
    # Initialize system
    ims = InventoryManagementSystem()

    # Add sample product
    product = {
        'product_id': 1,
        'product_name': 'Sample Product',
        'category': 'Electronics',
        'quantity': 100,
        'unit_price': 29.99,
        'reorder_point': 20,
        'supplier_id': 1,
        'min_order_qty': 50
    }
    ims.add_product(product)

    # Record some sample sales
    for i in range(5):
        sale = {
            'transaction_id': i + 1,
            'product_id': 1,
            'quantity': 10,
            'sale_date': (datetime.now() - timedelta(days=i)).strftime('%Y-%m-%d'),
            'sale_price': 29.99
        }
        ims.record_sale(sale)

    # Generate and print report
    report = ims.generate_report()
    print(json.dumps(report, indent=2))

    return ims

if __name__ == "__main__":
    demo_system()