# import pandas as pd # import numpy as np # from datetime import timedelta # import matplotlib # matplotlib.use('Agg') # Use a non-interactive backend # import matplotlib.pyplot as plt # from sklearn.linear_model import LinearRegression # from transformers import pipeline # import gradio as gr # import os # # Initialize a Hugging Face text-generation pipeline using an open-source model # text_generator = pipeline("text-generation", model="distilgpt2") # # ---------------------------- # # Utility Functions # # ---------------------------- # def load_sample_data(): # # Sample CSV data as a multi-line string # data = """Date,Product_ID,Category,Sales_Quantity,Price,Promotion_Flag,Holiday_Flag,Market_Trend_Score,Current_Stock,Reorder_Threshold,Lead_Time_Days,Supplier_ID,Supplier_Reliability_Score,Demand_Forecast_Next_30_Days,Risk_Type,Severity_Level,Reason,Expected_Delay_Days # 2023-01-01,1078,Jeans,74,44.4,1,1,92,842,141,10,238,67,1712,No Risk,Low,Port Strikes,5 # 2023-01-02,1041,Jackets,366,52.64,0,1,85,947,100,12,201,86,908,Shipment Delay,High,Port Strikes,8 # 2023-01-03,1063,Jeans,440,28.46,1,0,51,354,274,12,240,68,1025,No Risk,Medium,Port Strikes,6 # 2023-01-04,1060,Jackets,133,95.96,0,0,74,803,241,9,234,71,1889,No Risk,Low,Port Strikes,1 # 2023-01-05,1016,Shirts,113,20.8,0,1,74,342,77,8,229,93,1424,No Risk,High,No Issue,7 # 2023-01-06,1019,Jeans,346,99.89,0,1,96,163,101,2,248,81,684,No Risk,Low,High Demand,4 # 2023-01-07,1072,T-Shirts,251,149.16,1,0,50,396,278,2,297,70,697,No Risk,Low,Port Strikes,7""" # from io import StringIO # df = pd.read_csv(StringIO(data), parse_dates=["Date"]) # return df # # ---------------------------- # # Core Feature Functions # # ---------------------------- # def demand_forecasting_explainability(df, product_category): # df_cat = df[df["Category"] == product_category].copy() # if df_cat.empty: # return 0, "No data available for this category." # # Convert date to ordinal for regression modeling # df_cat["Date_Ordinal"] = df_cat["Date"].apply(lambda x: x.toordinal()) # X = df_cat[["Date_Ordinal"]] # y = df_cat["Sales_Quantity"] # model = LinearRegression() # model.fit(X, y) # last_date = df_cat["Date"].max() # future_dates = [last_date + timedelta(days=i) for i in range(1, 31)] # future_ordinals = np.array([d.toordinal() for d in future_dates]).reshape(-1, 1) # predictions = model.predict(future_ordinals) # forecast = int(np.mean(predictions)) # prompt = (f"Explain why demand for {product_category} is forecasted at {forecast} units " # "in the next 30 days, considering promotion, holiday effects, and market trends.") # explanation = text_generator(prompt, max_length=60, do_sample=True, temperature=0.7)[0]["generated_text"] # return forecast, explanation # def inventory_optimization(df): # recommendations = {} # for _, row in df.iterrows(): # product = row["Category"] # current_stock = row["Current_Stock"] # reorder_threshold = row["Reorder_Threshold"] # lead_time = row["Lead_Time_Days"] # forecast = row["Demand_Forecast_Next_30_Days"] # # Recommend reorder if current stock is below forecast or below the reorder threshold # if current_stock < forecast or current_stock < reorder_threshold: # rec_stock = int(forecast * 1.2) # Add 20% safety stock # recommendations[row["Product_ID"]] = { # "Category": product, # "Current_Stock": current_stock, # "Recommended_Stock": rec_stock, # "Lead_Time_Days": lead_time # } # return recommendations # def risk_alert_system(df): # alerts = {} # for _, row in df.iterrows(): # expected_delay = row["Expected_Delay_Days"] # risk_type = row["Risk_Type"] # # Alert if there's a risk flagged or an expected delay > 7 days # if risk_type != "No Risk" or expected_delay > 7: # alerts[row["Product_ID"]] = { # "Category": row["Category"], # "Risk_Type": risk_type, # "Expected_Delay_Days": expected_delay, # "Explanation": (f"Risk alert: {risk_type} with an expected delay of {expected_delay} days. " # f"Reason: {row['Reason']}.") # } # return alerts # def scenario_simulator(df, product_category, demand_increase_percent): # baseline_forecast, _ = demand_forecasting_explainability(df, product_category) # adjusted_forecast = int(baseline_forecast * (1 + demand_increase_percent / 100.0)) # prompt = (f"Given a {demand_increase_percent}% increase in demand for {product_category}, explain " # f"how the forecast changes from {baseline_forecast} to {adjusted_forecast} units.") # scenario_explanation = text_generator(prompt, max_length=60, do_sample=True, temperature=0.7)[0]["generated_text"] # return adjusted_forecast, scenario_explanation # def generate_inventory_plot(inventory_recs): # if not inventory_recs: # return None # product_ids = [str(pid) for pid in inventory_recs.keys()] # current_stock = [rec["Current_Stock"] for rec in inventory_recs.values()] # recommended_stock = [rec["Recommended_Stock"] for rec in inventory_recs.values()] # x = np.arange(len(product_ids)) # width = 0.35 # fig, ax = plt.subplots() # ax.bar(x - width/2, current_stock, width, label="Current Stock") # ax.bar(x + width/2, recommended_stock, width, label="Recommended Stock") # ax.set_ylabel("Stock Levels") # ax.set_title("Inventory Optimization") # ax.set_xticks(x) # ax.set_xticklabels(product_ids) # ax.legend() # os.makedirs("temp", exist_ok=True) # plot_filename = "temp/inventory_optimization.png" # plt.savefig(plot_filename) # plt.close(fig) # return plot_filename # # ---------------------------- # # Main Processing Function # # ---------------------------- # def process_file(file_obj): # # If no file is provided, use sample data. # if file_obj is None: # df = load_sample_data() # else: # try: # df = pd.read_csv(file_obj, parse_dates=["Date"]) # except Exception as e: # return f"Error reading CSV: {str(e)}", None # # For demonstration, focus on the "Jeans" category. # category = "Jeans" # forecast, explanation = demand_forecasting_explainability(df, category) # inv_recs = inventory_optimization(df) # risk_alerts = risk_alert_system(df) # adjusted_forecast, scenario_explanation = scenario_simulator(df, category, 10) # plot_file = generate_inventory_plot(inv_recs) # # Prepare text results # inv_text = "Inventory Recommendations:\n" # for pid, rec in inv_recs.items(): # inv_text += (f"Product ID {pid}: Current Stock {rec['Current_Stock']}, " # f"Recommended Stock {rec['Recommended_Stock']}, Lead Time {rec['Lead_Time_Days']} days.\n") # risk_text = "Risk Alerts:\n" # for pid, alert in risk_alerts.items(): # risk_text += (f"Product ID {pid}: {alert['Risk_Type']} (Delay: {alert['Expected_Delay_Days']} days) - " # f"{alert['Explanation']}\n") # output_text = (f"--- Demand Forecasting for {category} ---\nForecast (next 30 days): {forecast}\n" # f"Explanation: {explanation}\n\n") # output_text += inv_text + "\n" + risk_text + "\n" # output_text += (f"--- Scenario Simulation for {category} (10% Demand Increase) ---\n" # f"Adjusted Forecast: {adjusted_forecast}\nScenario Explanation: {scenario_explanation}\n") # return output_text, plot_file # # ---------------------------- # # Gradio Interface Setup # # ---------------------------- # iface = gr.Interface( # fn=process_file, # inputs=gr.File(label="Upload CSV File", file_types=['.csv']), # outputs=[ # gr.Textbox(label="Results", lines=20), # gr.Image(label="Inventory Optimization Plot") # ], # title="XAI for Supply Chain Optimization Prototype", # description="Upload a CSV file with supply chain data or leave empty to use sample data." # ) # if __name__ == "__main__": # iface.launch() # CODE 2 # import pandas as pd # import numpy as np # from datetime import timedelta # import matplotlib # matplotlib.use('Agg') # Non-interactive backend # import matplotlib.pyplot as plt # from sklearn.linear_model import LinearRegression # from transformers import pipeline # import gradio as gr # import os # # Initialize a Hugging Face text-generation pipeline using an open-source model # text_generator = pipeline("text-generation", model="distilgpt2") # # ---------------------------- # # Utility Functions # # ---------------------------- # def load_sample_data(): # # Sample CSV data as a multi-line string # data = """Date,Product_ID,Category,Sales_Quantity,Price,Promotion_Flag,Holiday_Flag,Market_Trend_Score,Current_Stock,Reorder_Threshold,Lead_Time_Days,Supplier_ID,Supplier_Reliability_Score,Demand_Forecast_Next_30_Days,Risk_Type,Severity_Level,Reason,Expected_Delay_Days # 2023-01-01,1078,Jeans,74,44.4,1,1,92,842,141,10,238,67,1712,No Risk,Low,Port Strikes,5 # 2023-01-02,1041,Jackets,366,52.64,0,1,85,947,100,12,201,86,908,Shipment Delay,High,Port Strikes,8 # 2023-01-03,1063,Jeans,440,28.46,1,0,51,354,274,12,240,68,1025,No Risk,Medium,Port Strikes,6 # 2023-01-04,1060,Jackets,133,95.96,0,0,74,803,241,9,234,71,1889,No Risk,Low,Port Strikes,1 # 2023-01-05,1016,Shirts,113,20.8,0,1,74,342,77,8,229,93,1424,No Risk,High,No Issue,7 # 2023-01-06,1019,Jeans,346,99.89,0,1,96,163,101,2,248,81,684,No Risk,Low,High Demand,4 # 2023-01-07,1072,T-Shirts,251,149.16,1,0,50,396,278,2,297,70,697,No Risk,Low,Port Strikes,7""" # from io import StringIO # df = pd.read_csv(StringIO(data), parse_dates=["Date"]) # return df # # ---------------------------- # # Core Feature Functions # # ---------------------------- # def demand_forecasting_explainability(df, product_category): # df_cat = df[df["Category"] == product_category].copy() # if df_cat.empty: # return 0, "No data available for this category." # # Convert date to ordinal for regression modeling # df_cat["Date_Ordinal"] = df_cat["Date"].apply(lambda x: x.toordinal()) # X = df_cat[["Date_Ordinal"]] # y = df_cat["Sales_Quantity"] # model = LinearRegression() # model.fit(X, y) # last_date = df_cat["Date"].max() # future_dates = [last_date + timedelta(days=i) for i in range(1, 31)] # future_ordinals = np.array([d.toordinal() for d in future_dates]).reshape(-1, 1) # predictions = model.predict(future_ordinals) # forecast = int(np.mean(predictions)) # prompt = (f"Explain why demand for {product_category} is forecasted at {forecast} units " # "in the next 30 days, considering promotion, holiday effects, and market trends. " # "Provide a brief supply chain tip.") # explanation = text_generator(prompt, max_length=60, do_sample=True, temperature=0.7)[0]["generated_text"] # return forecast, explanation # def inventory_optimization(df): # recommendations = {} # for _, row in df.iterrows(): # product = row["Category"] # current_stock = row["Current_Stock"] # reorder_threshold = row["Reorder_Threshold"] # lead_time = row["Lead_Time_Days"] # forecast = row["Demand_Forecast_Next_30_Days"] # # Recommend reorder if current stock is below forecast or below the reorder threshold # if current_stock < forecast or current_stock < reorder_threshold: # rec_stock = int(forecast * 1.2) # Add 20% safety stock # recommendations[row["Product_ID"]] = { # "Category": product, # "Current_Stock": current_stock, # "Recommended_Stock": rec_stock, # "Lead_Time_Days": lead_time # } # return recommendations # def risk_alert_system(df): # alerts = {} # for _, row in df.iterrows(): # expected_delay = row["Expected_Delay_Days"] # risk_type = row["Risk_Type"] # if risk_type != "No Risk" or expected_delay > 7: # alerts[row["Product_ID"]] = { # "Category": row["Category"], # "Risk_Type": risk_type, # "Expected_Delay_Days": expected_delay, # "Explanation": (f"Risk alert: {risk_type} with an expected delay of {expected_delay} days. " # f"Reason: {row['Reason']}.") # } # return alerts # def scenario_simulator(df, product_category, demand_increase_percent): # baseline_forecast, _ = demand_forecasting_explainability(df, product_category) # adjusted_forecast = int(baseline_forecast * (1 + demand_increase_percent / 100.0)) # prompt = (f"Given a {demand_increase_percent}% increase in demand for {product_category}, explain " # f"how the forecast changes from {baseline_forecast} to {adjusted_forecast} units. Provide a supply chain tip.") # scenario_explanation = text_generator(prompt, max_length=60, do_sample=True, temperature=0.7)[0]["generated_text"] # return adjusted_forecast, scenario_explanation # # ---------------------------- # # Plotting Functions for Each Feature # # ---------------------------- # def generate_demand_forecast_plot(df, product_category): # df_cat = df[df["Category"] == product_category].copy() # if df_cat.empty: # return None # df_cat["Date_Ordinal"] = df_cat["Date"].apply(lambda x: x.toordinal()) # X = df_cat[["Date_Ordinal"]] # y = df_cat["Sales_Quantity"] # model = LinearRegression() # model.fit(X, y) # df_cat["Predicted"] = model.predict(X) # last_date = df_cat["Date"].max() # future_dates = [last_date + timedelta(days=i) for i in range(1, 31)] # future_ordinals = np.array([d.toordinal() for d in future_dates]).reshape(-1, 1) # future_predictions = model.predict(future_ordinals) # fig, ax = plt.subplots(figsize=(6,4)) # ax.scatter(df_cat["Date"], df_cat["Sales_Quantity"], label="Historical Sales", color="blue") # ax.plot(df_cat["Date"], df_cat["Predicted"], label="Trend", color="green", linestyle="--") # ax.plot(future_dates, future_predictions, label="Forecast", color="red", linestyle="-") # ax.set_title("Demand Forecasting") # ax.set_xlabel("Date") # ax.set_ylabel("Sales Quantity") # ax.legend() # os.makedirs("temp", exist_ok=True) # demand_plot_file = "temp/demand_forecasting.png" # plt.savefig(demand_plot_file) # plt.close(fig) # return demand_plot_file # def generate_inventory_plot(inventory_recs): # if not inventory_recs: # return None # product_ids = [str(pid) for pid in inventory_recs.keys()] # current_stock = [rec["Current_Stock"] for rec in inventory_recs.values()] # recommended_stock = [rec["Recommended_Stock"] for rec in inventory_recs.values()] # x = np.arange(len(product_ids)) # width = 0.35 # fig, ax = plt.subplots(figsize=(6,4)) # ax.bar(x - width/2, current_stock, width, label="Current Stock", color="blue") # ax.bar(x + width/2, recommended_stock, width, label="Recommended Stock", color="red", alpha=0.7) # ax.set_ylabel("Stock Levels") # ax.set_title("Inventory Optimization") # ax.set_xticks(x) # ax.set_xticklabels(product_ids) # ax.legend() # os.makedirs("temp", exist_ok=True) # inv_plot_file = "temp/inventory_optimization.png" # plt.savefig(inv_plot_file) # plt.close(fig) # return inv_plot_file # def generate_risk_alert_plot(df): # alerts = risk_alert_system(df) # fig, ax = plt.subplots(figsize=(6,4)) # if not alerts: # ax.text(0.5, 0.5, "No Risk Alerts", horizontalalignment="center", verticalalignment="center", transform=ax.transAxes) # ax.axis('off') # else: # product_ids = list(alerts.keys()) # expected_delays = [alerts[pid]["Expected_Delay_Days"] for pid in product_ids] # ax.bar(product_ids, expected_delays, color="orange") # ax.set_xlabel("Product ID") # ax.set_ylabel("Expected Delay (Days)") # ax.set_title("Risk Alerts") # os.makedirs("temp", exist_ok=True) # risk_plot_file = "temp/risk_alerts.png" # plt.savefig(risk_plot_file) # plt.close(fig) # return risk_plot_file # def generate_scenario_simulation_plot(baseline_forecast, demand_increase_percent): # percents = np.linspace(0, demand_increase_percent, num=20) # forecasts = [int(baseline_forecast * (1 + p/100)) for p in percents] # fig, ax = plt.subplots(figsize=(6,4)) # ax.plot(percents, forecasts, marker="o", color="purple") # ax.set_xlabel("Demand Increase (%)") # ax.set_ylabel("Forecasted Demand") # ax.set_title("Scenario Simulation") # os.makedirs("temp", exist_ok=True) # scenario_plot_file = "temp/scenario_simulation.png" # plt.savefig(scenario_plot_file) # plt.close(fig) # return scenario_plot_file # # ---------------------------- # # Main Processing Function # # ---------------------------- # def process_file(file_obj): # # Load data: if no file provided, use sample data. # if file_obj is None: # df = load_sample_data() # else: # try: # df = pd.read_csv(file_obj, parse_dates=["Date"]) # except Exception as e: # return f"Error reading CSV: {str(e)}", None, None, None, None # # For demonstration, focus on the "Jeans" category. # category = "Jeans" # # 1. Demand Forecasting with Explainability # forecast, explanation = demand_forecasting_explainability(df, category) # demand_text = (f"--- Demand Forecasting for {category} ---\n" # f"Forecast (next 30 days): {forecast} units\n" # f"Explanation: {explanation}\n" # "Tip: Regularly monitor seasonal patterns and promotions to update forecasts.\n") # # 2. Inventory Optimization Dashboard # inv_recs = inventory_optimization(df) # inv_text = "Inventory Recommendations:\n" # for pid, rec in inv_recs.items(): # inv_text += (f"Product ID {pid}: Current Stock {rec['Current_Stock']}, " # f"Recommended Stock {rec['Recommended_Stock']}, Lead Time {rec['Lead_Time_Days']} days.\n") # inv_text += "Tip: Reorder promptly if stock is below forecast to avoid stockouts.\n" # # 3. Risk Alert System # risk_dict = risk_alert_system(df) # risk_text = "Risk Alerts:\n" # if risk_dict: # for pid, alert in risk_dict.items(): # risk_text += (f"Product ID {pid}: {alert['Risk_Type']} (Delay: {alert['Expected_Delay_Days']} days) - " # f"{alert['Explanation']}\n") # risk_text += "Tip: Investigate flagged risks immediately and consider alternative suppliers.\n" # else: # risk_text += "No risk alerts detected.\n" # # 4. Scenario Simulator # adjusted_forecast, scenario_explanation = scenario_simulator(df, category, 10) # scenario_text = (f"--- Scenario Simulation for {category} (10% Demand Increase) ---\n" # f"Adjusted Forecast: {adjusted_forecast} units\n" # f"Explanation: {scenario_explanation}\n" # "Tip: Use scenario simulations to prepare for demand fluctuations.\n") # overall_text = demand_text + "\n" + inv_text + "\n" + risk_text + "\n" + scenario_text # # Generate plots for each feature # demand_plot = generate_demand_forecast_plot(df, category) # inv_plot = generate_inventory_plot(inv_recs) # risk_plot = generate_risk_alert_plot(df) # scenario_plot = generate_scenario_simulation_plot(forecast, 10) # return overall_text, demand_plot, inv_plot, risk_plot, scenario_plot # # ---------------------------- # # Gradio Interface Setup # # ---------------------------- # iface = gr.Interface( # fn=process_file, # inputs=gr.File(label="Upload CSV File (or leave empty for sample data)", file_types=['.csv']), # outputs=[ # gr.Textbox(label="Results Summary", lines=20), # gr.Image(label="Demand Forecasting Plot"), # gr.Image(label="Inventory Optimization Plot"), # gr.Image(label="Risk Alerts Plot"), # gr.Image(label="Scenario Simulation Plot") # ], # title="XAI for Supply Chain Optimization Prototype", # description="Upload a CSV file with your supply chain data (or leave empty for sample data) to view detailed graphs, summaries, tips, and conclusions for each core feature." # ) # if __name__ == "__main__": # iface.launch() # CODE 3 import pandas as pd import numpy as np from datetime import timedelta import matplotlib matplotlib.use('Agg') # Non-interactive backend import matplotlib.pyplot as plt from sklearn.linear_model import LinearRegression from transformers import pipeline import gradio as gr import os from pytrends.request import TrendReq # Initialize a text-generation pipeline using a strong open-source model text_generator = pipeline("text-generation", model="EleutherAI/gpt-neo-125M") # ---------------------------- # Google Trends Integration # ---------------------------- def get_pytrends_insights(keyword): try: pytrends = TrendReq(hl='en-US', tz=360) pytrends.build_payload([keyword], timeframe='now 7-d') trends = pytrends.interest_over_time() if trends.empty: return f"No Google Trends data available for '{keyword}'." mean_interest = trends[keyword].mean() max_interest = trends[keyword].max() trend_text = (f"Google Trends for '{keyword}' shows an average interest of {mean_interest:.1f} " f"and a peak of {max_interest:.1f} in the past 7 days.") return trend_text except Exception as e: return f"Error fetching Google Trends data: {str(e)}" # ---------------------------- # Utility Functions # ---------------------------- def load_sample_data(): # Sample CSV data as a multi-line string (textile industry) data = """Date,Product_ID,Category,Sales_Quantity,Price,Promotion_Flag,Holiday_Flag,Market_Trend_Score,Current_Stock,Reorder_Threshold,Lead_Time_Days,Supplier_ID,Supplier_Reliability_Score,Demand_Forecast_Next_30_Days,Risk_Type,Severity_Level,Reason,Expected_Delay_Days 2023-01-01,1078,Jeans,74,44.4,1,1,92,842,141,10,238,67,1712,No Risk,Low,Port Strikes,5 2023-01-02,1041,Jackets,366,52.64,0,1,85,947,100,12,201,86,908,Shipment Delay,High,Port Strikes,8 2023-01-03,1063,Jeans,440,28.46,1,0,51,354,274,12,240,68,1025,No Risk,Medium,Port Strikes,6 2023-01-04,1060,Jackets,133,95.96,0,0,74,803,241,9,234,71,1889,No Risk,Low,Port Strikes,1""" from io import StringIO df = pd.read_csv(StringIO(data), parse_dates=["Date"]) return df # ---------------------------- # Core Feature Functions (ReAcT-style) # ---------------------------- def demand_forecasting_explainability(df, product_category): df_cat = df[df["Category"] == product_category].copy() if df_cat.empty: return 0, "No data available for the selected category." # Convert dates to ordinals for regression df_cat["Date_Ordinal"] = df_cat["Date"].apply(lambda x: x.toordinal()) X = df_cat[["Date_Ordinal"]] y = df_cat["Sales_Quantity"] model = LinearRegression() model.fit(X, y) last_date = df_cat["Date"].max() future_dates = [last_date + timedelta(days=i) for i in range(1, 31)] future_ordinals = np.array([d.toordinal() for d in future_dates]).reshape(-1, 1) predictions = model.predict(future_ordinals) forecast = int(np.mean(predictions)) # Fetch Google Trends insights for the selected product category trends_info = get_pytrends_insights(product_category) prompt = (f"You are a supply chain optimization AI agent with advanced ReAcT reasoning. Based on the following Google Trends data: {trends_info} " f"explain step-by-step why demand for {product_category} is forecasted at {forecast} units over the next 30 days. " "Consider factors such as promotions, holidays, and market trends. Provide detailed insights and actionable recommendations.") explanation = text_generator(prompt, max_length=120, do_sample=True, temperature=0.7)[0]["generated_text"] return forecast, explanation def inventory_optimization(df): recommendations = {} for _, row in df.iterrows(): product = row["Category"] current_stock = row["Current_Stock"] reorder_threshold = row["Reorder_Threshold"] lead_time = row["Lead_Time_Days"] forecast = row["Demand_Forecast_Next_30_Days"] # Recommend reorder if current stock is below forecast or below the reorder threshold if current_stock < forecast or current_stock < reorder_threshold: rec_stock = int(forecast * 1.2) # Add 20% safety stock recommendations[row["Product_ID"]] = { "Category": product, "Current_Stock": current_stock, "Recommended_Stock": rec_stock, "Lead_Time_Days": lead_time } return recommendations def risk_alert_system(df): alerts = {} for _, row in df.iterrows(): expected_delay = row["Expected_Delay_Days"] risk_type = row["Risk_Type"] if risk_type != "No Risk" or expected_delay > 7: alerts[row["Product_ID"]] = { "Category": row["Category"], "Risk_Type": risk_type, "Expected_Delay_Days": expected_delay, "Explanation": (f"Risk alert: {risk_type} with an expected delay of {expected_delay} days. " f"Reason: {row['Reason']}.") } return alerts def scenario_simulator(df, product_category, demand_increase_percent): baseline_forecast, _ = demand_forecasting_explainability(df, product_category) adjusted_forecast = int(baseline_forecast * (1 + demand_increase_percent / 100.0)) prompt = (f"You are a supply chain optimization AI agent with ReAcT reasoning. Explain, using a chain-of-thought approach, " f"how a {demand_increase_percent}% increase in demand for {product_category} changes the forecast from {baseline_forecast} to {adjusted_forecast} units. " "Provide actionable recommendations for adjusting inventory and mitigating risks.") scenario_explanation = text_generator(prompt, max_length=120, do_sample=True, temperature=0.7)[0]["generated_text"] return adjusted_forecast, scenario_explanation def inventory_insights(recommendations): if not recommendations: return "No inventory adjustments are needed; current stock levels are optimal." recs_str = "; ".join([f"Product {pid}: current {rec['Current_Stock']}, recommended {rec['Recommended_Stock']}" for pid, rec in recommendations.items()]) prompt = (f"Given these inventory recommendations: {recs_str}, provide a detailed chain-of-thought analysis with actionable tips " "to optimize inventory management and prevent stockouts.") insights = text_generator(prompt, max_length=120, do_sample=True, temperature=0.7)[0]["generated_text"] return insights def risk_insights(alerts): if not alerts: return "No risk alerts detected; the supply chain is stable." alerts_str = "; ".join([f"Product {pid}: {alert['Risk_Type']} (delay {alert['Expected_Delay_Days']} days)" for pid, alert in alerts.items()]) prompt = (f"Given the following risk alerts: {alerts_str}, provide a detailed chain-of-thought analysis with strategies " "to mitigate these risks and improve supplier reliability.") insights = text_generator(prompt, max_length=120, do_sample=True, temperature=0.7)[0]["generated_text"] return insights # ---------------------------- # Plotting Functions for Each Feature # ---------------------------- def generate_demand_forecast_plot(df, product_category): df_cat = df[df["Category"] == product_category].copy() if df_cat.empty: return None df_cat["Date_Ordinal"] = df_cat["Date"].apply(lambda x: x.toordinal()) X = df_cat[["Date_Ordinal"]] y = df_cat["Sales_Quantity"] model = LinearRegression() model.fit(X, y) df_cat["Predicted"] = model.predict(X) last_date = df_cat["Date"].max() future_dates = [last_date + timedelta(days=i) for i in range(1, 31)] future_ordinals = np.array([d.toordinal() for d in future_dates]).reshape(-1, 1) future_predictions = model.predict(future_ordinals) fig, ax = plt.subplots(figsize=(6,4)) ax.scatter(df_cat["Date"], df_cat["Sales_Quantity"], label="Historical Sales", color="blue") ax.plot(df_cat["Date"], df_cat["Predicted"], label="Trend", color="green", linestyle="--") ax.plot(future_dates, future_predictions, label="Forecast", color="red", linestyle="-") ax.set_title("Demand Forecasting") ax.set_xlabel("Date") ax.set_ylabel("Sales Quantity") ax.legend() os.makedirs("temp", exist_ok=True) demand_plot_file = "temp/demand_forecasting.png" plt.savefig(demand_plot_file) plt.close(fig) return demand_plot_file def generate_inventory_plot(inventory_recs): if not inventory_recs: return None product_ids = [str(pid) for pid in inventory_recs.keys()] current_stock = [rec["Current_Stock"] for rec in inventory_recs.values()] recommended_stock = [rec["Recommended_Stock"] for rec in inventory_recs.values()] x = np.arange(len(product_ids)) width = 0.35 fig, ax = plt.subplots(figsize=(6,4)) ax.bar(x - width/2, current_stock, width, label="Current Stock", color="blue") ax.bar(x + width/2, recommended_stock, width, label="Recommended Stock", color="red", alpha=0.7) ax.set_ylabel("Stock Levels") ax.set_title("Inventory Optimization") ax.set_xticks(x) ax.set_xticklabels(product_ids) ax.legend() os.makedirs("temp", exist_ok=True) inv_plot_file = "temp/inventory_optimization.png" plt.savefig(inv_plot_file) plt.close(fig) return inv_plot_file def generate_risk_alert_plot(df): alerts = risk_alert_system(df) fig, ax = plt.subplots(figsize=(6,4)) if not alerts: ax.text(0.5, 0.5, "No Risk Alerts", horizontalalignment="center", verticalalignment="center", transform=ax.transAxes) ax.axis('off') else: product_ids = list(alerts.keys()) expected_delays = [alerts[pid]["Expected_Delay_Days"] for pid in product_ids] ax.bar(product_ids, expected_delays, color="orange") ax.set_xlabel("Product ID") ax.set_ylabel("Expected Delay (Days)") ax.set_title("Risk Alerts") os.makedirs("temp", exist_ok=True) risk_plot_file = "temp/risk_alerts.png" plt.savefig(risk_plot_file) plt.close(fig) return risk_plot_file def generate_scenario_simulation_plot(baseline_forecast, demand_increase_percent): percents = np.linspace(0, demand_increase_percent, num=20) forecasts = [int(baseline_forecast * (1 + p/100)) for p in percents] fig, ax = plt.subplots(figsize=(6,4)) ax.plot(percents, forecasts, marker="o", color="purple") ax.set_xlabel("Demand Increase (%)") ax.set_ylabel("Forecasted Demand") ax.set_title("Scenario Simulation") os.makedirs("temp", exist_ok=True) scenario_plot_file = "temp/scenario_simulation.png" plt.savefig(scenario_plot_file) plt.close(fig) return scenario_plot_file # ---------------------------- # Main Processing Function (Dynamic) # ---------------------------- def process_file(file_obj, product_category, demand_increase_percent): # Load data: if no file provided, use sample data. if file_obj is None: df = load_sample_data() else: try: df = pd.read_csv(file_obj, parse_dates=["Date"]) except Exception as e: return f"Error reading CSV: {str(e)}", None, None, None, None # Dynamically adjust the product category based on available data available_categories = df["Category"].unique().tolist() if product_category not in available_categories: product_category = available_categories[0] # 1. Demand Forecasting with Explainability forecast, demand_explanation = demand_forecasting_explainability(df, product_category) demand_text = (f"--- Demand Forecasting for {product_category} ---\n" f"Forecast (next 30 days): {forecast} units\n" f"ReAcT Insights: {demand_explanation}\n" "Tip: Regularly monitor seasonal patterns, promotions, and market trends to update forecasts.\n") # 2. Inventory Optimization Dashboard inv_recs = inventory_optimization(df) inv_text = "Inventory Recommendations:\n" for pid, rec in inv_recs.items(): inv_text += (f"Product ID {pid}: Current Stock {rec['Current_Stock']}, " f"Recommended Stock {rec['Recommended_Stock']}, Lead Time {rec['Lead_Time_Days']} days.\n") inv_text += "Tip: Reorder promptly if stock is below forecast to avoid stockouts.\n" inv_insight_text = inventory_insights(inv_recs) # 3. Risk Alert System risk_dict = risk_alert_system(df) risk_text = "Risk Alerts:\n" if risk_dict: for pid, alert in risk_dict.items(): risk_text += (f"Product ID {pid}: {alert['Risk_Type']} (Delay: {alert['Expected_Delay_Days']} days) - " f"{alert['Explanation']}\n") risk_text += "Tip: Investigate flagged risks immediately and consider alternative suppliers.\n" else: risk_text += "No risk alerts detected.\n" risk_insight_text = risk_insights(risk_dict) # 4. Scenario Simulator adjusted_forecast, scenario_explanation = scenario_simulator(df, product_category, demand_increase_percent) scenario_text = (f"--- Scenario Simulation for {product_category} (Increase: {demand_increase_percent}%) ---\n" f"Adjusted Forecast: {adjusted_forecast} units\n" f"ReAcT Insights: {scenario_explanation}\n" "Tip: Use scenario simulations to prepare for demand fluctuations and adjust inventory accordingly.\n") # Get Google Trends insights separately trends_insights = get_pytrends_insights(product_category) overall_text = (demand_text + "\n" + inv_text + "\n" + inv_insight_text + "\n" + risk_text + "\n" + risk_insight_text + "\n" + scenario_text + "\n" + "Google Trends Insights: " + trends_insights) # Generate plots for each feature demand_plot = generate_demand_forecast_plot(df, product_category) inv_plot = generate_inventory_plot(inv_recs) risk_plot = generate_risk_alert_plot(df) scenario_plot = generate_scenario_simulation_plot(forecast, demand_increase_percent) return overall_text, demand_plot, inv_plot, risk_plot, scenario_plot # ---------------------------- # Gradio Interface Setup (Dynamic Inputs) # ---------------------------- iface = gr.Interface( fn=process_file, inputs=[ gr.File(label="Upload CSV File (or leave empty for sample data)", file_types=['.csv']), gr.Textbox(label="Product Category (e.g., Jeans, Jackets)", value="Jeans"), gr.Slider(label="Demand Increase (%)", minimum=0, maximum=50, step=1, value=10) ], outputs=[ gr.Textbox(label="Results Summary", lines=20), gr.Image(label="Demand Forecasting Plot"), gr.Image(label="Inventory Optimization Plot"), gr.Image(label="Risk Alerts Plot"), gr.Image(label="Scenario Simulation Plot") ], title="XAI for Textile Supply Chain Optimization - Dynamic ReAcT AI Agent", description=("Upload your CSV file (or leave empty to use sample data) and specify the product category and desired " "demand increase percentage. The agent will generate detailed chain-of-thought insights, graphs, and " "actionable tips – enhanced with real-time Google Trends data – to help optimize your supply chain decisions.") ) if __name__ == "__main__": iface.launch() # CODE 4 # import pandas as pd # import numpy as np # from datetime import timedelta # import matplotlib # matplotlib.use('Agg') # Non-interactive backend # import matplotlib.pyplot as plt # from sklearn.linear_model import LinearRegression # from transformers import pipeline # import gradio as gr # import os # from pytrends.request import TrendReq # # ---------------------------- # # Google Trends Integration # # ---------------------------- # def get_pytrends_insights(keyword): # try: # pytrends = TrendReq(hl='en-US', tz=360) # pytrends.build_payload([keyword], timeframe='now 7-d') # trends = pytrends.interest_over_time() # if trends.empty: # return f"No Google Trends data available for '{keyword}'." # mean_interest = trends[keyword].mean() # max_interest = trends[keyword].max() # trend_text = (f"Google Trends for '{keyword}' shows an average interest of {mean_interest:.1f} " # f"and a peak of {max_interest:.1f} in the past 7 days.") # return trend_text # except Exception as e: # return f"Error fetching Google Trends data: {str(e)}" # # ---------------------------- # # Utility Functions # # ---------------------------- # def load_sample_data(): # # Sample CSV data as a multi-line string for the textile industry # data = """Date,Product_ID,Category,Sales_Quantity,Price,Promotion_Flag,Holiday_Flag,Market_Trend_Score,Current_Stock,Reorder_Threshold,Lead_Time_Days,Supplier_ID,Supplier_Reliability_Score,Demand_Forecast_Next_30_Days,Risk_Type,Severity_Level,Reason,Expected_Delay_Days # 2023-01-01,1078,Jeans,74,44.4,1,1,92,842,141,10,238,67,1712,No Risk,Low,Port Strikes,5 # 2023-01-02,1041,Jackets,366,52.64,0,1,85,947,100,12,201,86,908,Shipment Delay,High,Port Strikes,8 # 2023-01-03,1063,Jeans,440,28.46,1,0,51,354,274,12,240,68,1025,No Risk,Medium,Port Strikes,6 # 2023-01-04,1060,Jackets,133,95.96,0,0,74,803,241,9,234,71,1889,No Risk,Low,Port Strikes,1""" # from io import StringIO # df = pd.read_csv(StringIO(data), parse_dates=["Date"]) # return df # # ---------------------------- # # Core Feature Functions (ReAcT-style) # # ---------------------------- # def demand_forecasting_explainability(df, product_category): # df_cat = df[df["Category"] == product_category].copy() # if df_cat.empty: # return 0, "No data available for the selected category." # # Convert dates to ordinals for regression modeling # df_cat["Date_Ordinal"] = df_cat["Date"].apply(lambda x: x.toordinal()) # X = df_cat[["Date_Ordinal"]] # y = df_cat["Sales_Quantity"] # model = LinearRegression() # model.fit(X, y) # last_date = df_cat["Date"].max() # future_dates = [last_date + timedelta(days=i) for i in range(1, 31)] # future_ordinals = np.array([d.toordinal() for d in future_dates]).reshape(-1, 1) # predictions = model.predict(future_ordinals) # forecast = int(np.mean(predictions)) # # Get Google Trends data to enrich the explanation # trends_info = get_pytrends_insights(product_category) # # Use a chain-of-thought prompt to mimic ReAcT reasoning (IBM Granite style) # prompt = (f"You are an advanced supply chain optimization AI agent (simulating IBM Granite’s XAI capabilities) with ReAcT reasoning. " # f"Based on the following Google Trends data: {trends_info} explain step-by-step why demand for {product_category} is forecasted " # f"at {forecast} units over the next 30 days. Consider factors such as promotions, holidays, and market trends. Provide detailed insights " # "and actionable recommendations.") # explanation = text_generator(prompt, max_length=120, do_sample=True, temperature=0.7)[0]["generated_text"] # return forecast, explanation # def inventory_optimization(df): # recommendations = {} # for _, row in df.iterrows(): # product = row["Category"] # current_stock = row["Current_Stock"] # reorder_threshold = row["Reorder_Threshold"] # lead_time = row["Lead_Time_Days"] # forecast = row["Demand_Forecast_Next_30_Days"] # # Recommend reorder if current stock is below forecast or below the reorder threshold # if current_stock < forecast or current_stock < reorder_threshold: # rec_stock = int(forecast * 1.2) # add 20% safety stock # recommendations[row["Product_ID"]] = { # "Category": product, # "Current_Stock": current_stock, # "Recommended_Stock": rec_stock, # "Lead_Time_Days": lead_time # } # return recommendations # def risk_alert_system(df): # alerts = {} # for _, row in df.iterrows(): # expected_delay = row["Expected_Delay_Days"] # risk_type = row["Risk_Type"] # if risk_type != "No Risk" or expected_delay > 7: # alerts[row["Product_ID"]] = { # "Category": row["Category"], # "Risk_Type": risk_type, # "Expected_Delay_Days": expected_delay, # "Explanation": (f"Risk alert: {risk_type} with an expected delay of {expected_delay} days. " # f"Reason: {row['Reason']}.") # } # return alerts # def scenario_simulator(df, product_category, demand_increase_percent): # baseline_forecast, _ = demand_forecasting_explainability(df, product_category) # adjusted_forecast = int(baseline_forecast * (1 + demand_increase_percent / 100.0)) # prompt = (f"You are a ReAcT AI agent for supply chain optimization. Using a chain-of-thought approach, explain " # f"how a {demand_increase_percent}% increase in demand for {product_category} changes the forecast " # f"from {baseline_forecast} to {adjusted_forecast} units. Provide actionable recommendations for inventory adjustments " # "and risk mitigation.") # scenario_explanation = text_generator(prompt, max_length=120, do_sample=True, temperature=0.7)[0]["generated_text"] # return adjusted_forecast, scenario_explanation # def inventory_insights(recommendations): # if not recommendations: # return "No inventory adjustments are needed; current stock levels are optimal." # recs_str = "; ".join([f"Product {pid}: current {rec['Current_Stock']}, recommended {rec['Recommended_Stock']}" # for pid, rec in recommendations.items()]) # prompt = (f"Given these inventory recommendations: {recs_str}, provide a detailed chain-of-thought analysis with actionable tips " # "to optimize inventory management and prevent stockouts.") # insights = text_generator(prompt, max_length=120, do_sample=True, temperature=0.7)[0]["generated_text"] # return insights # def risk_insights(alerts): # if not alerts: # return "No risk alerts detected; the supply chain is stable." # alerts_str = "; ".join([f"Product {pid}: {alert['Risk_Type']} (delay {alert['Expected_Delay_Days']} days)" # for pid, alert in alerts.items()]) # prompt = (f"Given the following risk alerts: {alerts_str}, provide a detailed chain-of-thought analysis with strategies " # "to mitigate these risks and improve supplier reliability.") # insights = text_generator(prompt, max_length=120, do_sample=True, temperature=0.7)[0]["generated_text"] # return insights # # ---------------------------- # # Plotting Functions for Each Feature # # ---------------------------- # def generate_demand_forecast_plot(df, product_category): # df_cat = df[df["Category"] == product_category].copy() # if df_cat.empty: # return None # df_cat["Date_Ordinal"] = df_cat["Date"].apply(lambda x: x.toordinal()) # X = df_cat[["Date_Ordinal"]] # y = df_cat["Sales_Quantity"] # model = LinearRegression() # model.fit(X, y) # df_cat["Predicted"] = model.predict(X) # last_date = df_cat["Date"].max() # future_dates = [last_date + timedelta(days=i) for i in range(1, 31)] # future_ordinals = np.array([d.toordinal() for d in future_dates]).reshape(-1, 1) # future_predictions = model.predict(future_ordinals) # fig, ax = plt.subplots(figsize=(6,4)) # ax.scatter(df_cat["Date"], df_cat["Sales_Quantity"], label="Historical Sales", color="blue") # ax.plot(df_cat["Date"], df_cat["Predicted"], label="Trend", color="green", linestyle="--") # ax.plot(future_dates, future_predictions, label="Forecast", color="red", linestyle="-") # ax.set_title("Demand Forecasting") # ax.set_xlabel("Date") # ax.set_ylabel("Sales Quantity") # ax.legend() # os.makedirs("temp", exist_ok=True) # demand_plot_file = "temp/demand_forecasting.png" # plt.savefig(demand_plot_file) # plt.close(fig) # return demand_plot_file # def generate_inventory_plot(inventory_recs): # if not inventory_recs: # return None # product_ids = [str(pid) for pid in inventory_recs.keys()] # current_stock = [rec["Current_Stock"] for rec in inventory_recs.values()] # recommended_stock = [rec["Recommended_Stock"] for rec in inventory_recs.values()] # x = np.arange(len(product_ids)) # width = 0.35 # fig, ax = plt.subplots(figsize=(6,4)) # ax.bar(x - width/2, current_stock, width, label="Current Stock", color="blue") # ax.bar(x + width/2, recommended_stock, width, label="Recommended Stock", color="red", alpha=0.7) # ax.set_ylabel("Stock Levels") # ax.set_title("Inventory Optimization") # ax.set_xticks(x) # ax.set_xticklabels(product_ids) # ax.legend() # os.makedirs("temp", exist_ok=True) # inv_plot_file = "temp/inventory_optimization.png" # plt.savefig(inv_plot_file) # plt.close(fig) # return inv_plot_file # def generate_risk_alert_plot(df): # alerts = risk_alert_system(df) # fig, ax = plt.subplots(figsize=(6,4)) # if not alerts: # ax.text(0.5, 0.5, "No Risk Alerts", horizontalalignment="center", verticalalignment="center", transform=ax.transAxes) # ax.axis('off') # else: # product_ids = list(alerts.keys()) # expected_delays = [alerts[pid]["Expected_Delay_Days"] for pid in product_ids] # ax.bar(product_ids, expected_delays, color="orange") # ax.set_xlabel("Product ID") # ax.set_ylabel("Expected Delay (Days)") # ax.set_title("Risk Alerts") # os.makedirs("temp", exist_ok=True) # risk_plot_file = "temp/risk_alerts.png" # plt.savefig(risk_plot_file) # plt.close(fig) # return risk_plot_file # def generate_scenario_simulation_plot(baseline_forecast, demand_increase_percent): # percents = np.linspace(0, demand_increase_percent, num=20) # forecasts = [int(baseline_forecast * (1 + p/100)) for p in percents] # fig, ax = plt.subplots(figsize=(6,4)) # ax.plot(percents, forecasts, marker="o", color="purple") # ax.set_xlabel("Demand Increase (%)") # ax.set_ylabel("Forecasted Demand") # ax.set_title("Scenario Simulation") # os.makedirs("temp", exist_ok=True) # scenario_plot_file = "temp/scenario_simulation.png" # plt.savefig(scenario_plot_file) # plt.close(fig) # return scenario_plot_file # # ---------------------------- # # Main Processing Function (Dynamic) # # ---------------------------- # def process_file(file_obj, product_category, demand_increase_percent): # # Load data: if no file provided, use sample data. # if file_obj is None: # df = load_sample_data() # else: # try: # df = pd.read_csv(file_obj, parse_dates=["Date"]) # except Exception as e: # return f"Error reading CSV: {str(e)}", None, None, None, None # # Dynamically adjust product category if not found # available_categories = df["Category"].unique().tolist() # if product_category not in available_categories: # product_category = available_categories[0] # # 1. Demand Forecasting with Explainability # forecast, demand_explanation = demand_forecasting_explainability(df, product_category) # demand_text = (f"--- Demand Forecasting for {product_category} ---\n" # f"Forecast (next 30 days): {forecast} units\n" # f"ReAcT Insights: {demand_explanation}\n" # "Tip: Regularly monitor seasonal patterns, promotions, and market trends to update forecasts.\n") # # 2. Inventory Optimization Dashboard # inv_recs = inventory_optimization(df) # inv_text = "Inventory Recommendations:\n" # for pid, rec in inv_recs.items(): # inv_text += (f"Product ID {pid}: Current Stock {rec['Current_Stock']}, " # f"Recommended Stock {rec['Recommended_Stock']}, Lead Time {rec['Lead_Time_Days']} days.\n") # inv_text += "Tip: Reorder promptly if stock is below forecast to avoid stockouts.\n" # inv_insight_text = inventory_insights(inv_recs) # # 3. Risk Alert System # risk_dict = risk_alert_system(df) # risk_text = "Risk Alerts:\n" # if risk_dict: # for pid, alert in risk_dict.items(): # risk_text += (f"Product ID {pid}: {alert['Risk_Type']} (Delay: {alert['Expected_Delay_Days']} days) - " # f"{alert['Explanation']}\n") # risk_text += "Tip: Investigate flagged risks immediately and consider alternative suppliers.\n" # else: # risk_text += "No risk alerts detected.\n" # risk_insight_text = risk_insights(risk_dict) # # 4. Scenario Simulator # adjusted_forecast, scenario_explanation = scenario_simulator(df, product_category, demand_increase_percent) # scenario_text = (f"--- Scenario Simulation for {product_category} (Increase: {demand_increase_percent}%) ---\n" # f"Adjusted Forecast: {adjusted_forecast} units\n" # f"ReAcT Insights: {scenario_explanation}\n" # "Tip: Use scenario simulations to prepare for demand fluctuations and adjust inventory accordingly.\n") # # Get additional Google Trends insights # trends_insights = get_pytrends_insights(product_category) # overall_text = (demand_text + "\n" + inv_text + "\n" + inv_insight_text + "\n" + # risk_text + "\n" + risk_insight_text + "\n" + scenario_text + "\n" + # "Google Trends Insights: " + trends_insights) # # Generate plots for each feature # demand_plot = generate_demand_forecast_plot(df, product_category) # inv_plot = generate_inventory_plot(inv_recs) # risk_plot = generate_risk_alert_plot(df) # scenario_plot = generate_scenario_simulation_plot(forecast, demand_increase_percent) # return overall_text, demand_plot, inv_plot, risk_plot, scenario_plot # # ---------------------------- # # Gradio Interface Setup (Dynamic Inputs) # # ---------------------------- # iface = gr.Interface( # fn=process_file, # inputs=[ # gr.File(label="Upload CSV File (or leave empty for sample data)", file_types=['.csv']), # gr.Textbox(label="Product Category (e.g., Jeans, Jackets)", value="Jeans"), # gr.Slider(label="Demand Increase (%)", minimum=0, maximum=50, step=1, value=10) # ], # outputs=[ # gr.Textbox(label="Results Summary", lines=20), # gr.Image(label="Demand Forecasting Plot"), # gr.Image(label="Inventory Optimization Plot"), # gr.Image(label="Risk Alerts Plot"), # gr.Image(label="Scenario Simulation Plot") # ], # title="XAI for Supply Chain Optimization with IBM Granite – ReAcT AI Agent Prototype", # description=("Upload your CSV file (or leave empty to use sample data) and specify the product category and desired " # "demand increase percentage. The agent uses dynamic data integration, pytrends insights, and ReAcT chain-of-thought reasoning " # "to generate actionable insights, detailed graphs, and recommendations for supply chain optimization.") # ) # if __name__ == "__main__": # iface.launch()