import streamlit as st
import pandas as pd
import numpy as np
import plotly.express as px
import time
from utils import (
validate_email,
validate_password_strength,
format_currency,
get_timestamp,
save_chat_session,
load_chat_session,
delete_chat_session,
clear_all_chat_history,
persist_user,
get_persisted_users,
save_active_session,
get_active_session,
clear_active_session,
get_ai_response,
stream_ai_response,
check_ollama_connection,
get_active_backend,
get_all_chat_sessions,
get_faq_response,
is_banking_query
)
st.set_page_config(
page_title="Central Bank AI",
page_icon="🏦",
layout="wide",
initial_sidebar_state="expanded"
)
def apply_custom_style(theme="dark"):
# Define color palette based on theme
if theme == "dark":
colors = {
"bg": "#0B1220",
"card_bg": "#111827",
"text": "#f1f5f9",
"text_secondary": "#94a3b8",
"primary": "#2563EB",
"secondary": "#0ea5e9",
"border": "#1F2937",
"input_bg": "rgba(30, 41, 59, 0.8)",
"shadow": "rgba(0, 0, 0, 0.4)",
"success": "#10B981",
"warning": "#f59e0b",
"danger": "#EF4444",
"sidebar_bg": "#0F172A",
"hover": "rgba(255, 255, 255, 0.05)"
}
else:
colors = {
"bg": "#F8FAFC",
"card_bg": "#FFFFFF",
"text": "#0F172A",
"text_secondary": "#64748B",
"primary": "#1E40AF",
"secondary": "#2563EB",
"border": "#E2E8F0",
"input_bg": "#F8FAFC",
"shadow": "rgba(0, 0, 0, 0.04)",
"success": "#10B981",
"warning": "#d97706",
"danger": "#EF4444",
"sidebar_bg": "#F1F5F9",
"hover": "#EFF6FF"
}
st.session_state.colors = colors
st.markdown(f"""
""", unsafe_allow_html=True)
def init_session_state():
if "users" not in st.session_state:
st.session_state.users = get_persisted_users()
if "logged_in" not in st.session_state:
last_user = get_active_session()
if last_user and last_user in st.session_state.users:
st.session_state.logged_in = True
st.session_state.username = last_user
st.session_state.email = st.session_state.users[last_user]["email"]
else:
st.session_state.logged_in = False
if "theme" not in st.session_state:
st.session_state.theme = "light"
apply_custom_style(st.session_state.theme)
if "username" not in st.session_state:
st.session_state.username = ""
if "email" not in st.session_state:
st.session_state.email = ""
if "current_page" not in st.session_state:
st.session_state.current_page = "login"
if "chat_sessions" not in st.session_state:
if st.session_state.logged_in and st.session_state.username:
st.session_state.chat_sessions = get_all_chat_sessions(st.session_state.username)
else:
st.session_state.chat_sessions = []
if "current_chat_id" not in st.session_state:
st.session_state.current_chat_id = None
if "messages" not in st.session_state:
st.session_state.messages = []
if "balance" not in st.session_state:
st.session_state.balance = 850000.00
if "interest_rate" not in st.session_state:
st.session_state.interest_rate = 6.5
if "accrued_interest" not in st.session_state:
st.session_state.accrued_interest = 55000.00
if "active_loans" not in st.session_state:
st.session_state.active_loans = 2
if "total_loan_amount" not in st.session_state:
st.session_state.total_loan_amount = 3500000.00
init_session_state()
def login(username, password):
if username in st.session_state.users:
if st.session_state.users[username]["password"] == password:
st.session_state.logged_in = True
st.session_state.username = username
st.session_state.email = st.session_state.users[username]["email"]
st.session_state.current_page = "dashboard"
st.session_state.chat_sessions = get_all_chat_sessions(username)
save_active_session(username)
return True
return False
def signup(username, email, password):
if username in st.session_state.users:
return False, "Username already exists"
st.session_state.users[username] = {
"email": email,
"password": password
}
persist_user(username, email, password)
return True, "Account created successfully!"
def logout():
st.session_state.logged_in = False
st.session_state.username = ""
st.session_state.email = ""
st.session_state.current_page = "login"
st.session_state.messages = []
st.session_state.current_chat_id = None
clear_active_session()
def get_mock_transactions():
dates = pd.date_range(end=pd.Timestamp.now(), periods=30, freq='D')
types = []
amounts = []
cats = []
for _ in range(30):
if np.random.random() > 0.8:
types.append("Income")
cats.append(np.random.choice(["Salary", "Investment", "Refund"]))
amounts.append(round(float(np.random.uniform(5000, 25000)), 2))
else:
types.append("Expense")
cats.append(np.random.choice(["Food", "Rent", "Shopping", "Transport", "Entertainment", "Utilities"]))
amounts.append(round(float(np.random.uniform(100, 5000)), 2))
data = {"Date": dates, "Category": cats, "Type": types, "Amount": amounts}
return pd.DataFrame(data)
def show_login_page():
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
st.title("🏦 Central Bank AI")
st.subheader("Login")
st.divider()
with st.form("login_form"):
username = st.text_input("Username", placeholder="Enter your username")
password = st.text_input("Password", type="password", placeholder="Enter your password")
submit = st.form_submit_button("Login", use_container_width=True, type="primary")
if submit:
if login(username, password):
st.success("Login successful!")
st.rerun()
else:
st.error("Invalid username or password")
st.divider()
if st.button("Don't have an account? Sign Up", use_container_width=True):
st.session_state.current_page = "signup"
st.rerun()
def show_signup_page():
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
st.title("🏦 Central Bank AI")
st.subheader("Create Account")
st.divider()
with st.form("signup_form"):
username = st.text_input("Username", placeholder="Choose a username")
email = st.text_input("Email", placeholder="Enter your email")
password = st.text_input("Password", type="password", placeholder="Create a password")
confirm_password = st.text_input("Confirm Password", type="password", placeholder="Re-enter your password")
submit = st.form_submit_button("Create Account", use_container_width=True, type="primary")
if submit:
if not username or not email or not password or not confirm_password:
st.error("All fields are required")
elif password != confirm_password:
st.error("Passwords do not match")
else:
success, msg = signup(username, email, password)
if success:
st.success(msg)
st.info("Please login with your credentials")
st.session_state.current_page = "login"
time.sleep(1)
st.rerun()
else:
st.error(msg)
st.divider()
if st.button("Already have an account? Login", use_container_width=True):
st.session_state.current_page = "login"
st.rerun()
def show_dashboard():
with st.sidebar:
st.markdown(f"""
""", unsafe_allow_html=True)
# User Info Section (New CSS Class)
st.markdown(f"""
{st.session_state.username[0].upper() if st.session_state.username else 'U'}
{st.session_state.username}
{st.session_state.email}
""", unsafe_allow_html=True)
if "current_tab" not in st.session_state:
st.session_state.current_tab = "Dashboard"
st.markdown("Navigation
", unsafe_allow_html=True)
nav_btn_style1 = "primary" if st.session_state.current_tab == "Dashboard" else "secondary"
nav_btn_style2 = "primary" if st.session_state.current_tab == "Banking Assistant" else "secondary"
if st.button("📊 Dashboard", use_container_width=True, type=nav_btn_style1):
st.session_state.current_tab = "Dashboard"
st.rerun()
if st.button(" Banking Assistant", use_container_width=True, type=nav_btn_style2):
st.session_state.current_tab = "Banking Assistant"
st.rerun()
page = st.session_state.current_tab
st.markdown("", unsafe_allow_html=True)
st.markdown("", unsafe_allow_html=True)
if st.button("Logout", use_container_width=True):
logout()
st.rerun()
st.markdown("
", unsafe_allow_html=True)
# Push Chat History to the bottom
st.markdown("", unsafe_allow_html=True)
st.markdown("Recent Chats
", unsafe_allow_html=True)
new_col, clear_col = st.columns([1, 1])
with new_col:
if st.button("➕ New Chat", use_container_width=True):
st.session_state.messages = []
st.session_state.current_chat_id = None
st.rerun()
with clear_col:
if st.session_state.chat_sessions and st.button("🗑️ Clear All", use_container_width=True):
clear_all_chat_history(st.session_state.username, st.session_state)
st.session_state.messages = []
st.session_state.current_chat_id = None
st.rerun()
# Chat Sessions
st.markdown("", unsafe_allow_html=True)
if st.session_state.chat_sessions:
# Display only top 5 initially, or all if "show_all_chats" is True
if "show_all_chats" not in st.session_state:
st.session_state.show_all_chats = False
display_chats = st.session_state.chat_sessions if st.session_state.show_all_chats else st.session_state.chat_sessions[:5]
for chat in display_chats:
preview = chat.get('preview', 'No messages')
chat_id = chat['session_id']
chat_col1, chat_col2 = st.columns([4, 1])
with chat_col1:
if st.button(f"📄 {preview}", key=f"chat_{chat_id}", use_container_width=True):
st.session_state.messages = chat['messages']
st.session_state.current_chat_id = chat_id
st.rerun()
with chat_col2:
if st.button("❌", key=f"del_{chat_id}", use_container_width=True):
delete_chat_session(st.session_state.username, st.session_state, chat_id)
if st.session_state.current_chat_id == chat_id:
st.session_state.messages = []
st.session_state.current_chat_id = None
st.rerun()
# Show "See all" button if there are more than 5 chats
if len(st.session_state.chat_sessions) > 5:
if st.session_state.show_all_chats:
if st.button("See Less", use_container_width=True):
st.session_state.show_all_chats = False
st.rerun()
else:
if st.button(f"See All ({len(st.session_state.chat_sessions)})", use_container_width=True):
st.session_state.show_all_chats = True
st.rerun()
else:
st.caption("No recent chats")
st.markdown("
", unsafe_allow_html=True)
st.title("Dashboard" if page == "Dashboard" else "Banking Assistant")
if page == "Dashboard":
st.markdown("## 📊 Dashboard Overview")
# Custom Metric Cards
st.markdown(f"""
Account Balance
{format_currency(st.session_state.balance)}
Interest Rate
{st.session_state.interest_rate}%
Active Loans
{st.session_state.active_loans}
""", unsafe_allow_html=True)
st.markdown("", unsafe_allow_html=True)
# 2 & 3. Insights & Health Score
col_health, col_insights = st.columns(2)
with col_health:
st.markdown(f"""
🟢 Financial Health Score
78 / 100
✓ Good savings ratio
✓ Low EMI burden
✓ Stable spending
""", unsafe_allow_html=True)
with col_insights:
st.markdown(f"""
💡 Smart Insights
📈 This month your spending increased by 12% compared to last month.
🛍️ Most spending category: Shopping.
⚠️ EMI due in 5 days.
""", unsafe_allow_html=True)
st.markdown("", unsafe_allow_html=True)
# 4 & 5. Net Worth & Upcoming Dues
col_nw, col_dues = st.columns(2)
with col_nw:
st.markdown(f"""
💎 Net Worth
Assets (Savings + FD + Investments)
{format_currency(st.session_state.balance + 3500000)}
Liabilities (Loans + Credit Dues)
{format_currency(st.session_state.total_loan_amount)}
Total Net Worth
{format_currency(st.session_state.balance + 3500000 - st.session_state.total_loan_amount)}
""", unsafe_allow_html=True)
with col_dues:
st.markdown(f"""
📅 Upcoming Payments
Home Loan EMI
₹12,000 due 5 Mar
Credit Card Bill
₹8,400 due 9 Mar
Electricity Bill
₹1,500 due 2 Mar
""", unsafe_allow_html=True)
st.divider()
# Visualizations
col_left, col_right = st.columns([2, 1])
df = get_mock_transactions()
with col_left:
st.write("### 📉 Income vs Expenses")
daily_data = df.groupby(['Date', 'Type'])['Amount'].sum().reset_index()
fig_bar = px.bar(
daily_data,
x='Date',
y='Amount',
color='Type',
barmode='group',
color_discrete_map={"Income": st.session_state.colors['success'], "Expense": st.session_state.colors['danger']}
)
fig_bar.update_layout(margin=dict(t=0, b=0, l=0, r=0), height=300, paper_bgcolor="rgba(0,0,0,0)", plot_bgcolor="rgba(0,0,0,0)", showlegend=True, legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1))
if st.session_state.theme == "dark":
fig_bar.update_layout(font_color="white")
else:
fig_bar.update_layout(font_color="black")
st.plotly_chart(fig_bar, use_container_width=True)
with col_right:
st.write("### 🍰 Expenses Breakdown")
expense_df = df[df['Type'] == 'Expense']
if expense_df.empty:
st.info("No expenses recorded yet.")
else:
category_data = expense_df.groupby('Category')['Amount'].sum().reset_index()
fig = px.pie(
category_data,
values='Amount',
names='Category',
hole=0.4,
color_discrete_sequence=[st.session_state.colors['primary'], st.session_state.colors['secondary'], '#38bdf8', '#818cf8', '#a78bfa', '#f472b6']
)
fig.update_layout(
margin=dict(t=0, b=0, l=0, r=0),
height=300,
showlegend=False
)
# Ensure transparent background for the chart
fig.update_layout(paper_bgcolor="rgba(0,0,0,0)", plot_bgcolor="rgba(0,0,0,0)")
if st.session_state.theme == "dark":
fig.update_layout(font_color="white")
else:
fig.update_layout(font_color="black")
st.plotly_chart(fig, use_container_width=True)
st.divider()
# Consolidated Transactions
st.markdown("### 📝 Recent Transaction History")
st.dataframe(
df.sort_values(by="Date", ascending=False),
use_container_width=True,
hide_index=True
)
else:
is_connected = check_ollama_connection()
col_h1, col_h2 = st.columns([4, 1])
with col_h2:
backend = get_active_backend()
if is_connected:
label = "☁️ Groq AI" if backend == "groq" else "🟢 Ollama"
st.markdown(f'● {label}
', unsafe_allow_html=True)
else:
st.markdown('○ Offline
', unsafe_allow_html=True)
st.markdown("", unsafe_allow_html=True)
# FAQ Suggestions
st.markdown("Popular Questions:
", unsafe_allow_html=True)
faq_row1_col1, faq_row1_col2, faq_row1_col3 = st.columns(3)
with faq_row1_col1:
if st.button("💰 Balance?", use_container_width=True):
st.session_state.faq_trigger = "What is my balance?"
st.rerun()
with faq_row1_col2:
if st.button("📈 Interest?", use_container_width=True):
st.session_state.faq_trigger = "What are the current interest rates?"
st.rerun()
with faq_row1_col3:
if st.button("📞 Support", use_container_width=True):
st.session_state.faq_trigger = "How do I contact customer care?"
st.rerun()
faq_row2_col1, faq_row2_col2, faq_row2_col3 = st.columns(3)
with faq_row2_col1:
if st.button("🕒 Hours", use_container_width=True):
st.session_state.faq_trigger = "What are the working hours?"
st.rerun()
with faq_row2_col2:
if st.button("🏦 Min Bal", use_container_width=True):
st.session_state.faq_trigger = "What is the minimum balance?"
st.rerun()
with faq_row2_col3:
if st.button("📋 FD Rates", use_container_width=True):
st.session_state.faq_trigger = "What are the FD rates?"
st.rerun()
chat_container = st.container(height=400, border=False)
with chat_container:
for message in st.session_state.messages:
role = message["role"]
if role == "user":
st.markdown(f'{message["content"]}
', unsafe_allow_html=True)
else:
st.markdown(f'{message["content"]}
', unsafe_allow_html=True)
prompt = st.chat_input("Ask about your finances or banking services...")
if getattr(st.session_state, 'faq_trigger', None):
prompt = st.session_state.faq_trigger
st.session_state.faq_trigger = None
if prompt:
st.session_state.messages.append({"role": "user", "content": prompt})
with chat_container:
st.markdown(f'{prompt}
', unsafe_allow_html=True)
faq_response = get_faq_response(prompt)
res_box = st.empty()
full_response = ""
if faq_response:
full_response = faq_response
res_box.markdown(f'{full_response}
', unsafe_allow_html=True)
elif is_banking_query(prompt):
if check_ollama_connection():
last_update_time = time.time()
for chunk in stream_ai_response(prompt, history=st.session_state.messages[:-1]):
if chunk:
full_response += chunk
current_time = time.time()
if current_time - last_update_time > 0.05:
res_box.markdown(f'{full_response}▌
', unsafe_allow_html=True)
last_update_time = current_time
res_box.markdown(f'{full_response}
', unsafe_allow_html=True)
else:
full_response = "I'm having trouble reaching the AI engine right now. However, I can still help with basic queries like your balance or interest rates. How can I assist you?"
res_box.markdown(f'{full_response}
', unsafe_allow_html=True)
else:
# Non-banking refusal
full_response = "I am a banking assistant and can only answer banking-related queries. Please feel free to ask about accounts, loans, cards, or other financial services."
res_box.markdown(f'{full_response}
', unsafe_allow_html=True)
if not full_response:
full_response = "I'm having trouble reaching the main AI engine right now. However, I can still help with basic queries like your balance or interest rates. How can I assist you?"
st.session_state.messages.append({"role": "assistant", "content": full_response})
# Save using the persistent utility
new_id = save_chat_session(st.session_state.username, st.session_state, st.session_state.messages, st.session_state.current_chat_id)
if not st.session_state.current_chat_id:
st.session_state.current_chat_id = new_id
st.rerun()
if not st.session_state.logged_in:
if st.session_state.current_page == "login":
show_login_page()
elif st.session_state.current_page == "signup":
show_signup_page()
else:
show_dashboard()