import streamlit as st
import pandas as pd
import os
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from langchain_openai import ChatOpenAI
from langchain_experimental.agents import create_pandas_dataframe_agent
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from dotenv import load_dotenv
import matplotlib.pyplot as plt
import base64
from io import BytesIO
import re
import seaborn as sns
# ✅ Load Environment Variables
load_dotenv()
openai_api_key = os.getenv("OPENAI_API_KEY")
# ✅ First command MUST be st.set_page_config()
st.set_page_config(page_title="AI Store Replacement & KPI Projection", layout="wide")
# ✅ Disable Right-Click
# ✅ Hide iframe using CSS
st.markdown(
"""
""",
unsafe_allow_html=True
)
# ✅ Disable Right-Click & Prevent Inspect Element
st.markdown(
"""
""",
unsafe_allow_html=True
)
st.markdown(
"""
""",
unsafe_allow_html=True
)
if not openai_api_key:
st.error("❌ OpenAI API Key is missing! Set it in your environment variables.")
# ✅ Load KPI Excel File
folder_path = os.getcwd()
kpi_file = os.path.join(folder_path, "kpi.xlsx")
df_kpi = pd.read_excel(kpi_file, engine="openpyxl")
supported_malls = [
"IBN BATTUTA MALL",
"DUBAI HILLS MALL",
"AL GHURAIR CENTRE DUBAI",
"DEIRA CITY CENTRE DUBAI",
"ARABIAN CENTRE DUBAI"
]
# ✅ Load Competitor Mapping Files (ISO-8859-1 Encoding)
mall_files = {
"Deira City Centre": "Competitor Mapping Doc Final1(DEIRA CITY CENTRE).csv",
"Arabian Centre": "Competitor Mapping Doc Final1(ARABIAN CENTRE).csv",
"Al Ghurair Centre": "Competitor Mapping Doc Final1(AL GHURAIR CENTRE).csv",
"Ibn Battuta Mall": "Competitor Mapping Doc Final1(IBN BATTUTA MALL).csv",
"Dubai Hills": "Competitor Mapping Doc Final1(DUBAI HILLS).csv"
}
mall_data = {}
for mall, file_name in mall_files.items():
file_path = os.path.join(folder_path, file_name)
try:
mall_data[mall] = pd.read_csv(file_path, encoding="ISO-8859-1")
except Exception as e:
st.error(f"⚠️ Error loading {file_name}: {e}")
# ✅ Title
st.title("🔄 AI-Powered Store Replacement & KPI Projection")
# ✅ 1️⃣ Sample Data Preview
st.subheader("📊 Sample Data from KPI Sheet")
st.dataframe(df_kpi.sample(5))
# ✅ 2️⃣ Select Mall & Closing Store
selected_mall = st.selectbox("🏬 Select a Mall:", supported_malls)
if selected_mall:
store_to_replace = st.selectbox("🚪 Select the store to replace:", df_kpi[df_kpi["Mall"] == selected_mall]["Store_Brand"].unique())
if store_to_replace:
# ✅ 3️⃣ AI-Based Category Identification
llm = ChatOpenAI(model="gpt-3.5-turbo", api_key=openai_api_key)
category_prompt = PromptTemplate.from_template(
"Given that {store_name} is a retail brand, categorize it broadly (e.g., Fashion, Electronics, Footwear, Food, Home Decor)."
)
category_chain = LLMChain(llm=llm, prompt=category_prompt)
store_category = category_chain.run(store_name=store_to_replace)
st.write(f"🔍 **Identified Category:** {store_category}")
# ✅ 4️⃣ Find Best Replacement Stores (Cosine Similarity)
df_kpi_filtered = df_kpi.drop(columns=["Store_Brand", "Mall", "Location", "City", "Country"]).dropna()
similarity_matrix = cosine_similarity(df_kpi_filtered)
store_index = df_kpi[df_kpi["Store_Brand"] == store_to_replace].index[0]
similarity_scores = list(enumerate(similarity_matrix[store_index]))
# ✅ Select top 3 unique stores
recommended_stores = []
used_stores = set(df_kpi[df_kpi["Mall"] == selected_mall]["Store_Brand"].unique())
for i in sorted(similarity_scores, key=lambda x: x[1], reverse=True):
potential_store = df_kpi.iloc[i[0]]["Store_Brand"]
if potential_store not in used_stores and potential_store not in recommended_stores:
recommended_stores.append(potential_store)
if len(recommended_stores) == 3:
break
if recommended_stores:
st.success("🔄 **Recommended Replacement Stores:**")
for replacement_store in recommended_stores:
store_data = df_kpi[df_kpi["Store_Brand"] == replacement_store]
if not store_data.empty:
# ✅ Extract Numeric Justification from KPI Data
sales = store_data["Annual_Sales"].values[0]
footfall = store_data["Footfall"].values[0]
margin = store_data["Gross_Margin"].values[0]
conversion_rate = store_data["Footfall_Conversion_Percent"].values[0] * 100 # ✅ Multiply by 100
# ✅ AI-Based Justification
justification_prompt = PromptTemplate.from_template(
"Analyze the historical data trends and customer demographics of {mall_name} from your knowledge"
"Explain why {replacement_store} is the best replacement for {store_to_replace}, considering these factors: "
"- Customer spending habits at {mall_name}. "
"- Historical footfall patterns. "
"- Expected revenue growth for {replacement_store}. "
"- How it aligns with customer demographics and preferences. "
"Use these KPIs for prediction: Sales: {sales}, Footfall: {footfall}, Margin: {margin}, Conversion Rate: {conversion_rate}%."
)
justification_chain = LLMChain(llm=llm, prompt=justification_prompt)
justification = justification_chain.run(
replacement_store=replacement_store,
store_to_replace=store_to_replace,
mall_name=selected_mall,
sales=f"{sales:,.2f} AED",
footfall=f"{footfall:,.0f} visitors",
margin=f"{margin:,.2f} AED",
conversion_rate=f"{conversion_rate:.2f}%"
)
st.write(f"📌 **{replacement_store}**")
st.write(f"🤖 **AI Justification:** {justification}")
# ✅ Display Projected KPIs
st.write(f"📊 **Projected KPIs for {replacement_store}:**")
st.write(f"💰 **Projected Sales:** AED {sales:,.2f}")
st.write(f"👣 **Projected Footfall:** {footfall:,.0f} visitors")
st.write(f"📈 **Projected Margin:** AED {margin:,.2f}")
st.write(f"🔄 **Projected Conversion Rate:** {conversion_rate:.2f}%")
# ✅ Fetch Competitors (Prefer Sheet Over AI)
competitor_list = []
if selected_mall in mall_data:
mall_df = mall_data[selected_mall]
if "Brand Name" in mall_df.columns:
competitors = mall_df[mall_df["Brand Name"].str.strip().str.lower() == replacement_store.lower()]
competitor_list = competitors.iloc[:, 2:].dropna(axis=1).values.flatten().tolist()
competitor_list = [c.strip() for c in competitor_list if isinstance(c, str)]
if competitor_list:
st.subheader(f"🔎 Competitors of {replacement_store} (from Sheet)")
for competitor in competitor_list:
st.write(f"- {competitor}")
else:
competitor_ai_prompt = PromptTemplate.from_template(
"Generate a list of 5 competing brands for {store_name} in the retail industry."
)
competitor_ai_chain = LLMChain(llm=llm, prompt=competitor_ai_prompt)
competitor_list = competitor_ai_chain.run(store_name=replacement_store).split("\n")
st.subheader(f"🤖 AI-Suggested Competitors for {replacement_store}")
for competitor in competitor_list:
st.write(f"- {competitor}")
# ✅ AI Chatbot with System Prompt
st.subheader("💬 Chat with WaysAhead AI Engine")
system_prompt = """
You are an advanced AI specializing in retail analytics. You help users:
1. Find similar stores based on category and sales performance.
2. Generate insights from sales data.
3. Create visualizations like bar charts, line charts, and heatmaps.
4. If a visualization request is made, generate clean Python code without markdown formatting or any text just give code.
"""
agent = create_pandas_dataframe_agent(
ChatOpenAI(model="gpt-3.5-turbo", api_key=openai_api_key),
df_kpi,
verbose=True,
agent_type="openai-tools",
allow_dangerous_code=True,
handle_parsing_errors=True,
system_prompt=system_prompt
)
# ✅ User Input
user_query = st.text_input("🔍 Ask AI for similar stores or analytics:")
if user_query:
try:
response = agent.run(user_query)
# ✅ Detect if AI Generated Python Code for a Chart
if "import matplotlib.pyplot" in response or "sns.heatmap" in response:
st.write("🛠 AI is generating a chart based on your request...")
# ✅ Extract Python Code Using Regex
code_match = re.search(r"```python\n(.*?)```", response, re.DOTALL)
if code_match:
ai_generated_code = code_match.group(1) # Extracted Python code
else:
ai_generated_code = response # Use full response if no markdown formatting
# ✅ Remove Any Remaining Triple Quotes or Formatting Issues
ai_generated_code = ai_generated_code.replace("```", "").strip()
# ✅ Create a dictionary for exec() and ensure df_kpi is accessible
exec_globals = {"df": df_kpi, "plt": plt, "sns": sns}
# ✅ Execute the AI-generated Python Code
try:
exec(ai_generated_code, exec_globals)
# ✅ Capture Matplotlib Figure
fig = plt.gcf()
buf = BytesIO()
fig.savefig(buf, format="png")
buf.seek(0)
# ✅ Display Image in Streamlit
st.image(buf, caption="📊 AI-Generated Chart", use_column_width=True)
except Exception as e:
st.error(f"⚠️ Error generating chart: {e}")
else:
st.markdown(f"🤖 **AI Response:** {response}")
except Exception as e:
st.error(f"⚠️ Error: {e}")
st.markdown("---")
st.markdown("🚀 **Powered by WaysAhead Global AI Engine**")