Update src/streamlit_app.py

src/streamlit_app.py

@@ -1,313 +1,266 @@
+# app.py
+"""
+Personal Finance Chatbot (Streamlit + IBM watsonx + Watson Assistant)
+- Single-file demo app
+- Requirements: streamlit, pandas, python-dotenv, ibm-watsonx-ai, ibm-watson
+- Set env vars before running (see README block below)
+"""
+
 import os
-import io
-import re
-from dataclasses import dataclass
-from typing import List, Dict, Optional
+import json
 import pandas as pd
 import streamlit as st
+from dotenv import load_dotenv
+from datetime import datetime
+from typing import Dict, Any, List
 
-# Hugging Face (optional)
+# IBM SDK imports
+try:
+    from ibm_watsonx_ai import APIClient, Credentials
+except Exception:
+    APIClient = None
 try:
-    from transformers import pipeline
-    HF_AVAILABLE = True
+    from ibm_watson import AssistantV2
+    from ibm_cloud_sdk_core.authenticators import IAMAuthenticator
 except Exception:
-    HF_AVAILABLE = False
+    AssistantV2 = None
 
-# -------------------------
-# Streamlit Config
-# -------------------------
-st.set_page_config(page_title="Personal Finance Chatbot", page_icon="💬", layout="wide")
+# Load env vars from .env if present
+load_dotenv()
 
 # -------------------------
-# Session State Init
+# Configuration / ENV VARS
 # -------------------------
-if "chat_history" not in st.session_state:
-    st.session_state["chat_history"] = []
-if "providers" not in st.session_state:
-    st.session_state["providers"] = {}
-if "provider_inited" not in st.session_state:
-    st.session_state["provider_inited"] = False
-if "provider_name" not in st.session_state:
-    st.session_state["provider_name"] = "huggingface"
+WATSONX_API_KEY = os.getenv("WATSONX_API_KEY")      # watsonx IAM apikey
+WATSONX_URL = os.getenv("WATSONX_URL")              # watsonx url/endpoint (e.g. https://us-south.ml.cloud.ibm.com)
+WATSONX_MODEL_ID = os.getenv("WATSONX_MODEL_ID", "ibm/granite-13b-instruct-v2")  # default example model
 
-# -------------------------
-# Categories & Intent Patterns
-# -------------------------
-CATEGORIES = {
-    "food": ["grocery", "restaurant", "dining"],
-    "transport": ["bus", "metro", "uber", "ola", "taxi"],
-    "entertainment": ["movie", "netflix", "spotify"],
-    "other": []
-}
+ASSISTANT_APIKEY = os.getenv("ASSISTANT_APIKEY")    # Watson Assistant api key (optional)
+ASSISTANT_URL = os.getenv("ASSISTANT_URL")          # Watson Assistant url (optional)
+ASSISTANT_ID = os.getenv("ASSISTANT_ID")            # Assistant ID (if using dialog)
 
-INTENT_PATTERNS = {
-    "savings": r"\bsav(e|ings|ing)\b",
-    "tax": r"\btax(es)?\b",
-    "invest": r"\binvest(ment|ing)?\b",
-    "budget": r"\bbudget(ing)?\b"
-}
+# Minimal checks
+if APIClient is None:
+    st.warning("ibm-watsonx-ai not installed. Install with: pip install ibm-watsonx-ai")
+if AssistantV2 is None:
+    st.info("ibm-watson (Assistant) client not installed or optional. Install: pip install ibm-watson")
 
 # -------------------------
-# AI Providers
+# Helper: WatsonX client
 # -------------------------
-class AIProvider:
-    def __init__(self):
-        self.name = "base"
-
-    def generate(self, prompt: str, max_tokens: int = 512) -> str:
-        raise NotImplementedError
-
-
-class HuggingFaceProvider(AIProvider):
-    def __init__(self):
-        super().__init__()
-        self.name = "huggingface"
-        self.gen = None
-        if HF_AVAILABLE:
-            try:
-                self.gen = pipeline("text2text-generation", model="google/flan-t5-base")
-            except Exception as e:
-                st.warning(f"HuggingFace pipeline failed to load: {e}")
+def make_watsonx_client() -> Any:
+    """Create and return a watsonx APIClient or None if not configured"""
+    if not (WATSONX_API_KEY and WATSONX_URL):
+        return None
+    credentials = Credentials(url=WATSONX_URL, api_key=WATSONX_API_KEY)
+    client = APIClient(credentials=credentials)
+    return client
+
+def watsonx_generate(client: Any, prompt: str, model_id: str = None, max_tokens: int = 512) -> str:
+    """
+    Generate text using watsonx foundation model via ibm_watsonx_ai client.
+    This function uses ModelInference utilities available on the client.
+    """
+    if client is None:
+        return "Watsonx client is not configured. Please set WATSONX_API_KEY and WATSONX_URL."
+
+    model_id = model_id or WATSONX_MODEL_ID
+    # Use the client's models or model inference helper:
+    # many SDK variations exist; below uses a common pattern (generate_text).
+    try:
+        model_inference = client.model_inference(model_id=model_id)
+        # simple call; more params (temperature, top_k, max_output_tokens) can be passed via params
+        generated = model_inference.generate_text(prompt=prompt, max_output_tokens=max_tokens)
+        # returned type can be dict/json or string depending on sdk version
+        if isinstance(generated, (dict, list)):
+            # attempt to extract textual content
+            text = json.dumps(generated)  # fallback representation
         else:
-            st.info("Transformers not installed; responses will be rule-based only.")
-
-    def generate(self, prompt: str, max_tokens: int = 512) -> str:
-        if self.gen is None:
-            return (
-                "[Rule-based fallback]\n"
-                + prompt[:1000]
-                + "\n\n(Summarized suggestion) Consider tracking expenses, setting goals, "
-                  "building an emergency fund, and using diversified, low-cost index funds aligned "
-                  "with your risk tolerance.)"
-            )
-        out = self.gen(prompt, max_length=min(1024, max_tokens), do_sample=False)
-        return out[0]["generated_text"].strip()
-
-
-class IBMGraniteWatsonProvider(AIProvider):
-    def __init__(self, watson_api_key: Optional[str], watson_url: Optional[str], granite_key: Optional[str]):
-        super().__init__()
-        self.name = "ibm_granite_watson"
-        self.ok = bool(watson_api_key and watson_url) or bool(granite_key)
-        self.watson_api_key = watson_api_key
-        self.watson_url = watson_url
-        self.granite_key = granite_key
-
-    def generate(self, prompt: str, max_tokens: int = 512) -> str:
-        if not self.ok:
-            return "[IBM placeholder] Missing credentials — falling back text.\n" + prompt
-        return (
-            "[IBM Granite/Watson simulated response]\n"
-            "(Replace this with real SDK call)\n\n"
-            + prompt
-        )
+            text = str(generated)
+        return text
+    except Exception as e:
+        return f"Error calling watsonx generate: {e}"
 
 # -------------------------
-# Helpers
+# Optional: Watson Assistant helper (for intents / dialog)
 # -------------------------
-def categorize(desc: str) -> str:
-    desc_l = (desc or "").lower()
-    for cat, keys in CATEGORIES.items():
-        if any(k in desc_l for k in keys):
-            return cat
-    return "other"
-
-
-@dataclass
-class UserProfile:
-    name: str
-    user_type: str
-    age: int
-    country: str
-    monthly_income: float
-    risk_tolerance: str
-    goals: str
-
-    def style_prompt(self) -> str:
-        if self.user_type.lower().startswith("stud"):
-            return (
-                "Explain like a friendly mentor to a student. Keep it clear and concise, "
-                "use practical examples and low-jargon."
-            )
-        return (
-            "Explain like a professional financial coach. Be precise, structured, and include "
-            "brief rationale with trade-offs."
-        )
-
+def make_assistant_client():
+    if not (ASSISTANT_APIKEY and ASSISTANT_URL):
+        return None
+    auth = IAMAuthenticator(ASSISTANT_APIKEY)
+    assistant = AssistantV2(version="2024-01-01", authenticator=auth)
+    assistant.set_service_url(ASSISTANT_URL)
+    return assistant
 
-def load_transactions(uploaded_file: Optional[io.BytesIO]) -> pd.DataFrame:
-    if uploaded_file is None:
-        data = {
-            "date": ["2025-08-01", "2025-08-02", "2025-08-03"],
-            "description": ["grocery store", "uber ride", "netflix subscription"],
-            "amount": [-1500, -300, -5000]
-        }
-        df = pd.DataFrame(data)
-    else:
-        try:
-            df = pd.read_csv(uploaded_file)
-        except Exception as e:
-            st.error(f"Could not read CSV: {e}. Showing demo data.")
-            data = {
-                "date": ["2025-08-01", "2025-08-02", "2025-08-03"],
-                "description": ["grocery store", "uber ride", "netflix subscription"],
-                "amount": [-1500, -300, -5000]
-            }
-            df = pd.DataFrame(data)
-    df["category"] = df["description"].apply(categorize)
+# -------------------------
+# Budget parsing & analytics
+# -------------------------
+def parse_transactions_csv(uploaded_file) -> pd.DataFrame:
+    """
+    Expect CSV with columns: date, description, amount, category (category optional).
+    Returns dataframe with normalized date and numeric amount.
+    """
+    df = pd.read_csv(uploaded_file)
+    # basic normalization
+    if "date" in df.columns:
+        df["date"] = pd.to_datetime(df["date"], errors="coerce")
+    if "amount" in df.columns:
+        df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
+    # create category if missing (simple rule-based)
+    if "category" not in df.columns:
+        df["category"] = df["description"].fillna("").str.lower().apply(guess_category_from_desc)
+    df = df.dropna(subset=["amount"])
     return df
 
+def guess_category_from_desc(desc: str) -> str:
+    desc = (desc or "").lower()
+    if any(k in desc for k in ["uber", "ola", "cab", "taxi"]):
+        return "transport"
+    if any(k in desc for k in ["grocery", "walmart", "bigbasket", "grocer"]):
+        return "groceries"
+    if any(k in desc for k in ["rent", "apartment", "house"]):
+        return "rent"
+    if any(k in desc for k in ["netflix", "spotify", "prime", "hulu"]):
+        return "entertainment"
+    if any(k in desc for k in ["salary", "pay", "deposit"]):
+        return "income"
+    return "other"
 
-def budget_summary(df: pd.DataFrame, monthly_income_hint: Optional[float] = None) -> Dict[str, float]:
-    df["month"] = pd.to_datetime(df["date"]).dt.to_period("M").astype(str)
-    income = df.loc[df["amount"] > 0, "amount"].sum()
-    expenses = -df.loc[df["amount"] < 0, "amount"].sum()
-    net = income - expenses
-    if monthly_income_hint and monthly_income_hint > 0:
-        income = max(income, monthly_income_hint)
-        net = income - expenses
-    savings_rate = (net / income) * 100 if income > 0 else 0.0
+def budget_summary(df: pd.DataFrame) -> Dict[str, Any]:
+    """
+    Generate a simple summary: total income, total expenses, top categories.
+    """
+    income = df[df["amount"] > 0]["amount"].sum()
+    expenses = -df[df["amount"] < 0]["amount"].sum()  # amounts might be negative for expenses
     by_cat = df.groupby("category")["amount"].sum().sort_values()
-    top_spend = (-by_cat[by_cat < 0]).nlargest(5)
+    top_exp = by_cat[by_cat < 0].sort_values().head(5) * -1
     return {
-        "income_total": float(round(income, 2)),
-        "expense_total": float(round(expenses, 2)),
-        "net_savings": float(round(net, 2)),
-        "savings_rate_pct": float(round(savings_rate, 2)),
-        "top_spend_json": top_spend.to_json(),
+        "total_income": float(income),
+        "total_expenses": float(expenses),
+        "net_savings": float(income - expenses),
+        "top_expense_categories": top_exp.to_dict()
     }
 
-
-def spending_suggestions(df: pd.DataFrame, profile: UserProfile) -> List[str]:
-    tips = []
-    summary = budget_summary(df, monthly_income_hint=profile.monthly_income)
-    if summary["savings_rate_pct"] < 10:
-        tips.append("Your savings rate is low. Try setting aside at least 20% of your income.")
-    if "entertainment" in df["category"].values:
-        tips.append("Entertainment spending is high. Consider limiting subscriptions or outings.")
-    if "food" in df["category"].values:
-        tips.append("Track food expenses. Cooking at home can save money.")
-    if not tips:
-        tips.append("Good job! Your spending looks balanced.")
-    return tips
-
-
-def detect_intent(text: str) -> str:
-    t = text.lower()
-    for k, pat in INTENT_PATTERNS.items():
-        if re.search(pat, t):
-            return k
-    return "general"
-
-
-def build_system_prompt(profile: UserProfile) -> str:
-    return (
-        f"You are a financial assistant. User profile: {profile}. "
-        f"Respond in style: {profile.style_prompt()}"
-    )
-
-
-def craft_user_prompt(query: str, intent: str, summary: Dict[str, float]) -> str:
-    return f"User asked about {intent}: {query}\nBudget summary: {summary}"
-
-# -------------------------
-# Sidebar Inputs
-# -------------------------
-with st.sidebar:
-    provider_choice = st.radio(
-        "Select AI Provider",
-        ["HuggingFace", "IBM Granite Watson", "Auto (Best Available)"]
-    )
-    uploaded = st.file_uploader("Upload your transactions CSV", type=["csv"])
-    st.markdown("### Profile")
-    profile = UserProfile(
-        name=st.text_input("Name", "Rahul"),
-        user_type=st.selectbox("User Type", ["Student", "Professional"]),
-        age=st.number_input("Age", 18, 100, 25),
-        country=st.text_input("Country", "India"),
-        monthly_income=st.number_input("Monthly Income", 0.0, 1e7, 50000.0),
-        risk_tolerance=st.selectbox("Risk Tolerance", ["Low", "Medium", "High"]),
-        goals=st.text_area("Financial Goals", "Save for emergency fund and invest in mutual funds")
-    )
-
 # -------------------------
-# Provider Initialization
+# Prompt engineering helpers
 # -------------------------
-if not st.session_state["provider_inited"]:
-    st.session_state["providers"]["ibm"] = IBMGraniteWatsonProvider(
-        watson_api_key=os.getenv("IBM_WATSON_API_KEY"),
-        watson_url=os.getenv("IBM_WATSON_URL"),
-        granite_key=os.getenv("IBM_GRANITE_API_KEY"),
+def build_prompt(user_question: str, demographic: str, budget_summary_text: str = "") -> str:
+    """
+    Build a watsonx prompt which adjusts tone and complexity based on demographic.
+    demographic: "student" or "professional"
+    """
+    tone = "friendly, simple, and educational" if demographic.lower() == "student" else "concise, professional, actionable"
+    complexity = "use short, clear sentences and examples" if demographic.lower() == "student" else "use precise financial language; include bullet recommendations"
+    prompt = (
+        f"You are a helpful personal finance assistant. Adopt a {tone} tone and {complexity}.\n\n"
+        f"Context (if any):\n{budget_summary_text}\n\n"
+        f"User question: {user_question}\n\n"
+        f"Provide:\n1) Short answer to the user's question.\n2) A 3-point actionable plan or suggestion.\n3) If the question is budget-related, give 1 quick saving tip.\n\nAnswer:"
     )
-    st.session_state["providers"]["hf"] = HuggingFaceProvider()
-
-    chosen = "huggingface"
-    ibm_ok = st.session_state["providers"]["ibm"].ok
-    if provider_choice.startswith("IBM") and ibm_ok:
-        chosen = "ibm_granite_watson"
-    elif provider_choice.startswith("Auto"):
-        chosen = "ibm_granite_watson" if ibm_ok else "huggingface"
-    st.session_state["provider_name"] = chosen
-    st.session_state["provider_inited"] = True
-
-provider_name = st.session_state["provider_name"]
-if provider_choice.startswith("IBM"):
-    provider_name = "ibm_granite_watson" if st.session_state["providers"]["ibm"].ok else "huggingface"
-elif provider_choice.startswith("Auto"):
-    provider_name = "ibm_granite_watson" if st.session_state["providers"]["ibm"].ok else "huggingface"
-else:
-    provider_name = "huggingface"
-st.session_state["provider_name"] = provider_name
-
-provider = (
-    st.session_state["providers"]["ibm"] if provider_name == "ibm_granite_watson"
-    else st.session_state["providers"]["hf"]
-)
+    return prompt
 
 # -------------------------
-# Layout
+# Streamlit UI
 # -------------------------
-col_chat, col_right = st.columns([0.62, 0.38])
-
-with col_right:
-    st.subheader("📊 Budget Summary")
-    df = load_transactions(uploaded)
-    st.dataframe(df, use_container_width=True, height=250)
-    summary = budget_summary(df, monthly_income_hint=profile.monthly_income)
-    m1, m2, m3, m4 = st.columns(4)
-    m1.metric("Income (₹)", f"{summary['income_total']:.0f}")
-    m2.metric("Expenses (₹)", f"{summary['expense_total']:.0f}")
-    m3.metric("Net (₹)", f"{summary['net_savings']:.0f}")
-    m4.metric("Savings Rate", f"{summary['savings_rate_pct']:.1f}%")
-
-    st.markdown("### 🧠 AI Spending Suggestions")
-    for tip in spending_suggestions(df, profile):
-        st.write("• ", tip)
-
-with col_chat:
-    st.subheader("🗣️ Ask about savings, taxes, investments, or budgeting")
-    for turn in st.session_state["chat_history"]:
-        with st.chat_message(turn["role"]):
-            st.markdown(turn["content"])
-    user_msg = st.chat_input("Type your question… (e.g., How much should I invest monthly for a ₹10L goal?)")
-    if user_msg:
-        st.session_state["chat_history"].append({"role": "user", "content": user_msg})
-        intent = detect_intent(user_msg)
-        sys_prompt = build_system_prompt(profile)
-        usr_prompt = craft_user_prompt(user_msg, intent, summary)
-        final_prompt = sys_prompt + "\n\n" + usr_prompt
-        with st.chat_message("assistant"):
-            with st.spinner(f"Thinking with {provider.name}…"):
-                try:
-                    ai = provider.generate(final_prompt, max_tokens=768)
-                except Exception as e:
-                    ai = f"Provider error: {e}\nFalling back to heuristic guidance.\n" + usr_prompt
-            st.markdown(ai)
-        st.session_state["chat_history"].append({"role": "assistant", "content": ai})
-
-st.markdown("""
----
-**Disclaimers**  
-This chatbot provides educational information only and is **not** financial, tax, or legal advice.  
-Tax rules change frequently; consult a qualified professional for personalized advice.
-""")
+st.set_page_config(page_title="Personal Finance Chatbot", layout="wide")
+st.title("💬 Personal Finance Chatbot — Savings, Taxes, Investments (IBM watsonx)")
+
+# Sidebar: upload transactions and settings
+st.sidebar.header("Data & Settings")
+uploaded = st.sidebar.file_uploader("Upload transactions CSV (date,description,amount,category optional)", type=["csv"])
+demographic = st.sidebar.selectbox("User type (affects tone & complexity)", ["student", "professional"])
+model_choice = st.sidebar.text_input("watsonx model id", value=WATSONX_MODEL_ID)
+
+st.sidebar.markdown("**API status**")
+watsonx_client = make_watsonx_client()
+assistant_client = make_assistant_client()
+st.sidebar.write("watsonx configured:", bool(watsonx_client))
+st.sidebar.write("Assistant configured:", bool(assistant_client))
+
+# Load transactions if provided
+tx_df = None
+budget_text = ""
+if uploaded:
+    try:
+        tx_df = parse_transactions_csv(uploaded)
+        st.sidebar.success(f"Loaded {len(tx_df)} transactions")
+        summary = budget_summary(tx_df)
+        # create a short textual budget summary for context to the model
+        budget_text = (
+            f"Total income: {summary['total_income']:.2f}. "
+            f"Total expenses: {summary['total_expenses']:.2f}. "
+            f"Net savings: {summary['net_savings']:.2f}. "
+            f"Top expense categories: {', '.join([f'{k}: {v:.2f}' for k,v in summary['top_expense_categories'].items()])}."
+        )
+    except Exception as e:
+        st.sidebar.error(f"Failed to parse CSV: {e}")
+
+# Main: chat area
+if "chat_messages" not in st.session_state:
+    st.session_state.chat_messages = []
+
+chat_col, info_col = st.columns([3,1])
+with chat_col:
+    st.subheader("Chat")
+    # display history
+    for msg in st.session_state.chat_messages:
+        if msg["role"] == "user":
+            st.chat_message("user").write(msg["content"])
+        else:
+            st.chat_message("assistant").write(msg["content"])
+
+    user_input = st.chat_input("Ask about savings, taxes, budgets, or investments...")
+    if user_input:
+        # show the user's message immediately
+        st.session_state.chat_messages.append({"role":"user", "content": user_input})
+        # Build prompt for watsonx
+        prompt = build_prompt(user_input, demographic, budget_summary_text=budget_text)
+        with st.spinner("Generating response from watsonx..."):
+            if watsonx_client:
+                response_text = watsonx_generate(watsonx_client, prompt=prompt, model_id=model_choice, max_tokens=512)
+            else:
+                # fallback heuristic answer if watsonx not configured
+                response_text = (
+                    "Watsonx not configured. (Set WATSONX_API_KEY and WATSONX_URL.)\n\n"
+                    "Quick tips: 1) Track monthly expenses; 2) Build 3-6 months emergency fund; 3) Automate savings."
+                )
+        # Append assistant response
+        st.session_state.chat_messages.append({"role":"assistant", "content": response_text})
+        st.experimental_rerun()
+
+with info_col:
+    st.subheader("Quick Tools")
+    if tx_df is not None:
+        st.markdown("**Budget snapshot**")
+        st.write(budget_text)
+        if st.button("Show top expenses table"):
+            top = tx_df.groupby("category")["amount"].sum().abs().sort_values(ascending=False).reset_index()
+            top.columns = ["category","total_spent"]
+            st.table(top.head(10))
+    else:
+        st.info("Upload transactions to enable budget summaries & insights")
+
+# Footer: example prompts & environment instructions
+st.markdown("---")
+st.markdown("**Example prompts**")
+st.markdown("- `How much should I save monthly given my income is ₹50,000?`")
+st.markdown("- `Suggest tax-saving instruments suitable for a young professional in India.`")
+st.markdown("- `I spent too much on food. How can I cut dining expenses by 20%?`")
+
+st.markdown("---")
+st.markdown("**Setup / Requirements**")
+st.code("""
+pip install streamlit pandas python-dotenv ibm-watsonx-ai ibm-watson
+# Environment variables (example)
+export WATSONX_API_KEY='your_watsonx_api_key'
+export WATSONX_URL='https://<region>.ml.cloud.ibm.com'
+export WATSONX_MODEL_ID='ibm/granite-13b-instruct-v2'
+# Optional (Assistant)
+export ASSISTANT_APIKEY='your_assistant_apikey'
+export ASSISTANT_URL='https://api.<region>.assistant.watson.cloud.ibm.com'
+export ASSISTANT_ID='your_assistant_id'
+# Run
+streamlit run app.py
+""", language="bash")
+
+st.caption("This is a prototype demo — for production, add secure secret handling, input validation, rate-limiting, logging, and robust error handling.")