/*
Inventory Management Database Schema
*/
CREATE TABLE parts (
part_id INT PRIMARY KEY AUTO_INCREMENT,
sku VARCHAR(50) UNIQUE NOT NULL,
name VARCHAR(100) NOT NULL,
description TEXT,
category_id INT NOT NULL,
supplier_id INT NOT NULL,
unit_cost DECIMAL(10,2) NOT NULL,
unit_price DECIMAL(10,2) NOT NULL,
min_stock_level INT NOT NULL,
max_stock_level INT NOT NULL,
lead_time_days INT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (category_id) REFERENCES categories(category_id),
FOREIGN KEY (supplier_id) REFERENCES suppliers(supplier_id)
);
CREATE TABLE inventory (
inventory_id INT PRIMARY KEY AUTO_INCREMENT,
part_id INT NOT NULL,
quantity_on_hand INT NOT NULL,
quantity_allocated INT DEFAULT 0,
last_counted_date DATE,
location VARCHAR(50),
FOREIGN KEY (part_id) REFERENCES parts(part_id)
);
CREATE TABLE sales_history (
sale_id INT PRIMARY KEY AUTO_INCREMENT,
part_id INT NOT NULL,
sale_date DATE NOT NULL,
quantity INT NOT NULL,
unit_price DECIMAL(10,2) NOT NULL,
customer_id INT,
FOREIGN KEY (part_id) REFERENCES parts(part_id),
FOREIGN KEY (customer_id) REFERENCES customers(customer_id)
);
CREATE TABLE categories (
category_id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(50) NOT NULL,
description TEXT
);
CREATE TABLE suppliers (
supplier_id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(100) NOT NULL,
contact_person VARCHAR(100),
email VARCHAR(100),
phone VARCHAR(20),
lead_time_days INT
);
CREATE TABLE customers (
customer_id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(100) NOT NULL,
email VARCHAR(100),
phone VARCHAR(20)
);
CREATE TABLE forecast_results (
forecast_id INT PRIMARY KEY AUTO_INCREMENT,
part_id INT NOT NULL,
forecast_date DATE NOT NULL,
forecast_model VARCHAR(20) NOT NULL, -- 'ARIMA' or 'LSTM'
forecast_period INT NOT NULL, -- in months
forecast_quantity DECIMAL(10,2) NOT NULL,
confidence_interval_lower DECIMAL(10,2),
confidence_interval_upper DECIMAL(10,2),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (part_id) REFERENCES parts(part_id)
);
CREATE TABLE alerts (
alert_id INT PRIMARY KEY AUTO_INCREMENT,
part_id INT NOT NULL,
alert_type VARCHAR(20) NOT NULL, -- 'low_stock', 'forecast', etc.
alert_message TEXT NOT NULL,
alert_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
resolved BOOLEAN DEFAULT FALSE,
resolved_date TIMESTAMP NULL,
FOREIGN KEY (part_id) REFERENCES parts(part_id)
);
CREATE TABLE alert_recipients (
recipient_id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(100) NOT NULL,
email VARCHAR(100) NOT NULL,
phone VARCHAR(20),
receive_email BOOLEAN DEFAULT TRUE,
receive_sms BOOLEAN DEFAULT FALSE,
alert_types VARCHAR(100) -- comma-separated list of alert types to receive
);
"""
Inventory Forecasting Script using ARIMA and LSTM
"""
import pandas as pd
import numpy as np
from statsmodels.tsa.arima.model import ARIMA
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_absolute_error
import sqlalchemy
import smtplib
from email.mime.text import MIMEText
import warnings
warnings.filterwarnings('ignore')
# Database connection
def get_db_connection():
engine = sqlalchemy.create_engine('mysql+pymysql://user:password@localhost/inventory_db')
return engine.connect()
# Fetch sales data for a specific part
def fetch_sales_data(part_id, months=24):
conn = get_db_connection()
query = f"""
SELECT sale_date, quantity
FROM sales_history
WHERE part_id = {part_id}
AND sale_date >= DATE_SUB(CURRENT_DATE, INTERVAL {months} MONTH)
ORDER BY sale_date
"""
df = pd.read_sql(query, conn)
conn.close()
# Resample to monthly data if needed
df['sale_date'] = pd.to_datetime(df['sale_date'])
df.set_index('sale_date', inplace=True)
monthly_df = df.resample('M').sum()
return monthly_df
# ARIMA Forecasting
def arima_forecast(data, periods=6):
# Fit ARIMA model
model = ARIMA(data, order=(1,1,1))
model_fit = model.fit()
# Make forecast
forecast = model_fit.forecast(steps=periods)
return forecast
# LSTM Forecasting
def lstm_forecast(data, periods=6, look_back=6):
# Normalize data
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(data.values.reshape(-1,1))
# Prepare data for LSTM
def create_dataset(dataset, look_back=1):
X, Y = [], []
for i in range(len(dataset)-look_back-1):
a = dataset[i:(i+look_back), 0]
X.append(a)
Y.append(dataset[i + look_back, 0])
return np.array(X), np.array(Y)
X, Y = create_dataset(scaled_data, look_back)
X = np.reshape(X, (X.shape[0], X.shape[1], 1))
# Build LSTM model
model = Sequential()
model.add(LSTM(50, return_sequences=True, input_shape=(look_back, 1)))
model.add(LSTM(50))
model.add(Dense(1))
model.compile(loss='mean_squared_error', optimizer='adam')
model.fit(X, Y, epochs=20, batch_size=1, verbose=0)
# Make forecast
forecast_input = scaled_data[-look_back:].reshape(1, look_back, 1)
forecast = []
for _ in range(periods):
next_step = model.predict(forecast_input)
forecast.append(next_step[0,0])
forecast_input = np.append(forecast_input[:,1:,:], [[next_step]], axis=1)
forecast = scaler.inverse_transform(np.array(forecast).reshape(-1,1))
return forecast.flatten()
# Save forecast to database
def save_forecast(part_id, forecast, model_name, forecast_date):
conn = get_db_connection()
for i, value in enumerate(forecast):
query = f"""
INSERT INTO forecast_results
(part_id, forecast_date, forecast_model, forecast_period, forecast_quantity)
VALUES ({part_id}, DATE_ADD('{forecast_date}', INTERVAL {i+1} MONTH),
'{model_name}', {i+1}, {value})
"""
conn.execute(query)
conn.close()
# Check stock levels and send alerts
def check_stock_and_alert(part_id, forecast):
conn = get_db_connection()
# Get current inventory and threshold
query = f"""
SELECT p.part_id, p.name, p.min_stock_level, i.quantity_on_hand
FROM parts p
JOIN inventory i ON p.part_id = i.part_id
WHERE p.part_id = {part_id}
"""
inventory = pd.read_sql(query, conn).iloc[0]
# Get forecast for next month
monthly_forecast = forecast[0]
# Check if projected stock will be below threshold
projected_stock = inventory['quantity_on_hand'] - monthly_forecast
if projected_stock < inventory['min_stock_level']:
# Create alert
alert_msg = f"Projected stock for {inventory['name']} will be below threshold ({projected_stock}/{inventory['min_stock_level']})"
query = f"""
INSERT INTO alerts (part_id, alert_type, alert_message)
VALUES ({part_id}, 'low_stock', '{alert_msg}')
"""
conn.execute(query)
# Send email/SMS to recipients
send_alerts(part_id, alert_msg)
conn.close()
# Send email/SMS alerts
def send_alerts(part_id, message):
conn = get_db_connection()
# Get recipients who should receive this type of alert
query = f"""
SELECT * FROM alert_recipients
WHERE alert_types LIKE '%low_stock%'
"""
recipients = pd.read_sql(query, conn)
# Send alerts
for _, recipient in recipients.iterrows():
if recipient['receive_email']:
send_email(recipient['email'], "Inventory Alert", message)
if recipient['receive_sms']:
send_sms(recipient['phone'], message)
conn.close()
def send_email(to_email, subject, body):
# Configure your email settings
smtp_server = "smtp.example.com"
smtp_port = 587
smtp_user = "alerts@example.com"
smtp_password = "password"
msg = MIMEText(body)
msg['Subject'] = subject
msg['From'] = smtp_user
msg['To'] = to_email
try:
server = smtplib.SMTP(smtp_server, smtp_port)
server.starttls()
server.login(smtp_user, smtp_password)
server.sendmail(smtp_user, [to_email], msg.as_string())
server.quit()
except Exception as e:
print(f"Failed to send email: {e}")
def send_sms(phone_number, message):
# Implement your SMS gateway integration here
pass
# Main forecasting workflow
def run_forecast_for_part(part_id):
# Get historical sales data
sales_data = fetch_sales_data(part_id)
if len(sales_data) < 6:
print(f"Not enough data for part {part_id}")
return
# Run ARIMA forecast
arima_result = arima_forecast(sales_data['quantity'])
# Run LSTM forecast
lstm_result = lstm_forecast(sales_data['quantity'])
# Save forecasts
today = pd.to_datetime('today').strftime('%Y-%m-%d')
save_forecast(part_id, arima_result, 'ARIMA', today)
save_forecast(part_id, lstm_result, 'LSTM', today)
# Check stock levels and send alerts
check_stock_and_alert(part_id, arima_result)
# Example usage
if __name__ == "__main__":
# Get list of parts that need forecasting (e.g., all active parts)
conn = get_db_connection()
parts = pd.read_sql("SELECT part_id FROM parts WHERE active = TRUE", conn)
conn.close()
# Run forecast for each part
for part_id in parts['part_id']:
try:
run_forecast_for_part(part_id)
print(f"Forecast completed for part {part_id}")
except Exception as e:
print(f"Error forecasting for part {part_id}: {e}")
