Upload your UPI/Bank statements (Paytm, GPay, PhonePe, etc.) â Parse â Analyze â Get insights
', unsafe_allow_html=True) # --------- Sidebar --------- st.sidebar.title("âšī¸ How to Use") st.sidebar.write("1) Upload one or more statement PDFs.") st.sidebar.write("2) Review parsed transactions.") st.sidebar.write("3) (Optional) Enter monthly income for savings%.") st.sidebar.write("4) Generate LLM insights or use offline analytics.") with st.sidebar.expander("âī¸ Settings", expanded=False): user_monthly_income = st.number_input("Monthly Income (âš) â optional", min_value=0, step=1000, value=0) enable_ocr = st.checkbox("Enable OCR fallback (slower)", value=False) chunk_limit = st.number_input("Max rows sent to LLM", min_value=100, max_value=5000, value=800, step=100) show_offline_insights = st.checkbox("Show offline insights (no LLM)", value=True) if GEMINI_API_KEY: st.sidebar.success("Gemini API key detected from environment.") else: st.sidebar.warning("No GEMINI_API_KEY found. LLM features will be disabled. Offline insights still work.") # --------- Helpers (parsing, normalization, categorization) --------- DATE_PATTERNS = [ "%d-%m-%Y", "%d/%m/%Y", "%Y-%m-%d", "%d-%b-%Y", "%d %b %Y", "%d %B %Y" ] DATE_RE = re.compile(r"\b(\d{2}[/-]\d{2}[/-]\d{4}|\d{4}-\d{2}-\d{2}|\d{2}\s?[A-Za-z]{3}\s?\d{4})\b") AMOUNT_RE = re.compile( r"(?:âš|INR)?\s*([+-]?\d{1,3}(?:,\d{3})*(?:\.\d{1,2})|[+-]?\d+(?:\.\d{1,2}))" ) DEBIT_KEYS = { "debit", "debited", "withdrawal", "withdraw", "purchase", "payment", "paid", "upi to", "money sent", "sent", "transfer to", "atm", "pos", "bill", "emi", "petrol", "fuel", "dining", "zomato", "swiggy", "insurance premium", "online purchase" } CREDIT_KEYS = { "credit", "credited", "cr", "received", "refund", "reversal", "cashback", "salary", "deposit", "transfer from", "interest", "redeem", "redemption" } BALANCE_KEYS = {"opening balance", "closing balance", "balance only", "closingbal", "openingbal"} CATEGORY_RULES = { "food": ["swiggy", "zomato", "restaurant", "hotel", "food"], "grocery": ["bigbasket", "dmart", "grocery", "mart", "supermarket"], "utilities": ["electricity", "water", "gas", "biller", "broadband", "internet", "dth"], "transport": ["uber", "ola", "metro", "fuel", "petrol", "diesel", "fastag"], "shopping": ["amazon", "flipkart", "myntra", "ajio", "store", "shop"], "entertainment": ["netflix", "prime", "hotstar", "movie", "bookmyshow"], "financial": ["bank", "emi", "loan", "insurance", "sip", "mutual fund"], "transfer": ["upi", "imps", "neft", "rtgs", "to", "from"], } UPI_ID_RE = re.compile(r"[A-Za-z0-9._-]+@[A-Za-z]+") MERCHANT_CANON = { "swiggy": "Swiggy", "zomato": "Zomato", "amazon": "Amazon", "flipkart": "Flipkart", "uber": "Uber", "ola": "Ola" } def _normalize_date(s: str) -> str | None: s = s.strip() for fmt in DATE_PATTERNS: try: return datetime.strptime(s, fmt).strftime("%Y-%m-%d") except Exception: continue return None def _extract_texts_from_pdf(file_bytes: bytes, use_ocr: bool = False) -> str: text_chunks: List[str] = [] with pdfplumber.open(io.BytesIO(file_bytes)) as pdf: for page in pdf.pages: t = page.extract_text() or "" # If page has no extractable text and OCR fallback is enabled if (not t) and use_ocr and OCR_AVAILABLE: try: images = convert_from_bytes(file_bytes, first_page=page.page_number, last_page=page.page_number) if images: t = pytesseract.image_to_string(images[0]) or "" except Exception: # If OCR fails keep text as empty string t = "" text_chunks.append(t) return "\n".join(text_chunks).strip() def _canonical_merchant(name: str) -> str: low = (name or "").lower() # direct substring match for key, canon in MERCHANT_CANON.items(): if key in low: return canon # fuzzy match to known merchants try: best = process.extractOne(low, list(MERCHANT_CANON.keys()), scorer=fuzz.WRatio) if best and isinstance(best, tuple) and len(best) >= 2 and best[1] > 85: return MERCHANT_CANON.get(best[0], name) except Exception: pass return (name or "").strip()[:80] def _categorize(desc: str) -> str: d = (desc or "").lower() for cat, kws in CATEGORY_RULES.items(): if any(k in d for k in kws): return cat return "other" # ----------------------- # Amount/type extraction helpers # ----------------------- def _find_amounts(txt: str) -> list[float]: vals: list[float] = [] for m in AMOUNT_RE.findall(txt or ""): s = m if isinstance(m, str) else (m[0] if isinstance(m, (list, tuple)) else str(m)) try: vals.append(float(s.replace(",", ""))) except Exception: pass return vals def _guess_type_from_text(txt: str) -> str: low = (txt or "").lower() if any(k in low for k in BALANCE_KEYS): return "balance" if re.search(r"\bcr\b", low): return "credit" if re.search(r"\bdr\b", low): return "debit" if any(k in low for k in CREDIT_KEYS): return "credit" if any(k in low for k in DEBIT_KEYS): return "debit" if "transfer from" in low or ("from" in low and "transfer" in low): return "credit" if "transfer to" in low or ("to" in low and "transfer" in low): return "debit" return "unknown" def _extract_amount_and_type(txt: str) -> tuple[float | None, str, float | None]: amounts = _find_amounts(txt) tx_type = _guess_type_from_text(txt) if tx_type == "balance" and len(amounts) == 1: return (None, "balance_only", amounts[0]) balance_amt = None tx_amt = None if len(amounts) >= 2: # common pattern: transaction amount then balance balance_amt = amounts[-1] tx_amt = amounts[0] elif len(amounts) == 1: tx_amt = amounts[0] return (tx_amt, tx_type, balance_amt) # ----------------------- # Parsing pipeline (extract_and_parse) # ----------------------- @st.cache_data(show_spinner=False) def extract_and_parse(files: List[Tuple[str, bytes]], use_ocr: bool = False) -> pd.DataFrame: all_rows: List[Dict[str, Any]] = [] for filename, file_bytes in files: raw_text = _extract_texts_from_pdf(file_bytes, use_ocr=use_ocr) source = filename lines = [ln.strip() for ln in raw_text.splitlines() if ln.strip()] block: List[str] = [] for ln in lines: if DATE_RE.search(ln) and block: all_rows.extend(_parse_block(" ".join(block), source)) block = [ln] else: block.append(ln) if block: all_rows.extend(_parse_block(" ".join(block), source)) df = pd.DataFrame(all_rows).drop_duplicates() if not df.empty: df["date"] = df["date"].apply(lambda x: _normalize_date(str(x)) or "") df["amount"] = pd.to_numeric(df["amount"], errors="coerce") df = df.dropna(subset=["amount"]).reset_index(drop=True) with pd.option_context('mode.chained_assignment', None): df["month"] = df["date"].apply(lambda s: s[:7] if isinstance(s, str) and len(s) >= 7 else "") df["merchant"] = df["merchant"].fillna("").apply(_canonical_merchant) df["category"] = df["description"].fillna("").apply(_categorize) df["type"] = df["type"].fillna("") df.loc[~df["type"].isin(["debit", "credit"]), "type"] = "unknown" if "currency" not in df.columns: df["currency"] = "INR" if "transaction_id" not in df.columns: df["transaction_id"] = [str(uuid.uuid4()) for _ in range(len(df))] df["signed_amount"] = np.where( df["type"] == "credit", df["amount"], np.where(df["type"] == "debit", -df["amount"], np.nan) ) return df # ----------------------- # Improved block parser # ----------------------- def _parse_block(txt: str, source_pdf: str) -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] date_match = DATE_RE.search(txt or "") date_str = date_match.group(0) if date_match else "" tx_amt, tx_type, balance_amt = _extract_amount_and_type(txt or "") if tx_type in {"balance_only", "balance"} and tx_amt is None: return rows upi_match = UPI_ID_RE.search(txt or "") merchant = upi_match.group(0) if upi_match else "" if not merchant: tokens = [t for t in re.split(r"[^A-Za-z0-9@._-]+", txt or "") if t and not t.replace('.', '', 1).isdigit()] merchant = max(tokens, key=len)[:80] if tokens else "" if tx_type == "unknown": low = (txt or "").lower() if any(k in low for k in ["salary", "deposit", "refund", "received", "cashback"]): tx_type = "credit" elif any(k in low for k in ["withdraw", "payment", "purchase", "upi to", "sent", " to "]): tx_type = "debit" if tx_amt is not None: rows.append({ "date": date_str, "time": "", "amount": float(tx_amt), "type": tx_type if tx_type in {"debit", "credit"} else "unknown", "merchant": merchant, "description": (txt or "").strip()[:300], "method": _infer_method_from_source(source_pdf), "source_pdf": source_pdf, "running_balance": balance_amt if balance_amt is not None else np.nan, }) return rows def _infer_method_from_source(name: str) -> str: low = (name or "").lower() if "paytm" in low: return "Paytm" if "phonepe" in low: return "PhonePe" if "gpay" in low or "google" in low: return "GPay" return "UPI" # ----------------------- # Offline analytics (unchanged) # ----------------------- def compute_offline_insights(df: pd.DataFrame, income: int = 0) -> Dict[str, Any]: out: Dict[str, Any] = {"months": {}, "top_categories": [], "top_merchants": [], "notes": []} if df.empty: return out month_grp = df[df["type"].isin(["debit", "credit"])].groupby(["month", "type"]).agg(total=("amount", "sum")).reset_index() months: Dict[str, Dict[str, float]] = {} for _, r in month_grp.iterrows(): m = r["month"] or "unknown" if m not in months: months[m] = {"income": 0.0, "expenses": 0.0} if r["type"] == "credit": months[m]["income"] += float(r["total"]) elif r["type"] == "debit": months[m]["expenses"] += float(r["total"]) out["months"] = months cat = df[df["type"] == "debit"].groupby("category")["amount"].sum().sort_values(ascending=False) out["top_categories"] = [(k, float(v)) for k, v in cat.head(10).items()] mer = df[df["type"] == "debit"].groupby("merchant")["amount"].sum().sort_values(ascending=False) out["top_merchants"] = [(k, float(v)) for k, v in mer.head(10).items()] notes: List[str] = [] for m, vals in months.items(): inc = income if income > 0 else vals.get("income", 0.0) exp = vals.get("expenses", 0.0) if inc > 0: savings_pct = max(0.0, (inc - exp) / inc * 100.0) notes.append(f"{m}: Savings {savings_pct:.1f}% (Income âš{inc:,.0f}, Expenses âš{exp:,.0f})") else: notes.append(f"{m}: Expenses âš{exp:,.0f} (No income provided/detected)") out["notes"] = notes return out # ----------------------- # LLM Recommendation Generator (Gemini) # ----------------------- def build_llm_context(df: pd.DataFrame, max_rows: int = 800) -> str: cols = ["date", "amount", "type", "merchant", "category", "method"] slim = df[cols].copy() if not df.empty else pd.DataFrame(columns=cols) if len(slim) > max_rows: slim = slim.sort_values("date").tail(max_rows) return slim.to_json(orient="records", date_format="iso") @st.cache_data(show_spinner=False) def generate_recommendations_gemini(df: pd.DataFrame, income: int = 0, max_rows: int = 800) -> str: """ Ask Gemini for: - Monthly budget planning - Suggestions to reduce unnecessary spending - Personalized financial advice Output expected as Markdown with clearly labeled sections. """ if not GENAI_READY: return "â ī¸ LLM not configured. Set GEMINI_API_KEY to enable AI insights." try: context_json = build_llm_context(df, max_rows=max_rows) system_msg = ( "You are a practical, precise personal finance advisor. Use the provided JSON transactions to compute figures." ) user_instructions = { "goals": [ "1) Provide a monthly budget plan (monthly target per category and total monthly budget).", "2) Suggest concrete ways to reduce unnecessary or recurring spending with estimated âš savings.", "3) Provide 3 personalized financial recommendations (e.g., emergency fund, SIPs, debt paydown), with estimated impact in âš or % where possible." ], "notes": "Use transaction data strictly from DATA(JSON). If income is provided, use it for savings calculations; otherwise infer income from credits.", "output_format": { "sections": ["Monthly Budget Plan", "Reduce Spending Suggestions", "Personalized Advice"], "format": "Markdown with bullet lists and numeric estimates. Keep it concise." }, "user_income": income } prompt = ( f"SYSTEM:\n{system_msg}\n\n" f"DATA(JSON):\n{context_json}\n\n" f"INSTRUCTIONS:\n{json.dumps(user_instructions, ensure_ascii=False)}\n\n" "Produce the response in Markdown with the three sections exactly as in output_format." ) # Use the configured genai client model = genai.GenerativeModel(GEMINI_MODEL_NAME) resp = model.generate_content(prompt) return (getattr(resp, 'text', None) or resp or "").strip() if resp else "â ī¸ Empty response from LLM." except Exception as e: err_str = str(e) if "429" in err_str or "quota" in err_str.lower(): return "â ī¸ LLM Quota Exceeded. You've made too many requests. Please try again later." # Special-case for blocked prompts if SDK provides such exception try: BlockedExc = getattr(genai.types, 'generation_types', None) # if a specific exception type is available we could catch it above except Exception: BlockedExc = None return f"â ī¸ LLM error: {err_str}" # ----------------------- # UI: File upload, parse, analyze, LLM button # ----------------------- uploaded_files = st.file_uploader( "đ Upload one or more PDF statements", type=["pdf"], accept_multiple_files=True ) if uploaded_files: files_payload = [(f.name, f.read()) for f in uploaded_files] st.success(f"â {len(files_payload)} file(s) uploaded.") if enable_ocr and not OCR_AVAILABLE: st.warning("OCR requested but OCR libraries (pytesseract/pdf2image) are not available. Parsing will proceed without OCR.") with st.spinner("đ Extracting & parsing transactions..."): df = extract_and_parse(files_payload, use_ocr=(enable_ocr and OCR_AVAILABLE)) if df.empty: st.error("No transactions parsed. Try enabling OCR or verify the PDF format.") else: st.subheader("đ Parsed Transactions (preview)") st.dataframe(df.head(500), use_container_width=True) # Basic KPIs (ignore unknown) df_known = df[df["type"].isin(["debit", "credit"])].copy() total_exp = float(df_known.loc[df_known["type"] == "debit", "amount"].sum()) total_inc = float(df_known.loc[df_known["type"] == "credit", "amount"].sum()) kpi1, kpi2, kpi3 = st.columns(3) kpi1.metric("Total Expenses (âš)", f"{total_exp:,.0f}") kpi2.metric("Total Income (âš)", f"{total_inc:,.0f}") net = total_inc - total_exp kpi3.metric("Net (âš)", f"{net:,.0f}", delta=None) # Charts & downloads tab1, tab2, tab3 = st.tabs(["đ Monthly", "đī¸ Categories", "đˇī¸ Merchants"]) with tab1: if "month" in df_known.columns and df_known["month"].nunique() > 0: msum = df_known.groupby(["month", "type"]).amount.sum().unstack(fill_value=0) st.bar_chart(msum) with tab2: cat_sum = df_known[df_known["type"] == "debit"].groupby("category").amount.sum().sort_values(ascending=False) st.bar_chart(cat_sum) with tab3: mer_sum = df_known[df_known["type"] == "debit"].groupby("merchant").amount.sum().sort_values(ascending=False).head(20) st.bar_chart(mer_sum) st.download_button( label="âŦī¸ Download Parsed CSV", data=df.to_csv(index=False).encode("utf-8"), file_name="transactions_parsed.csv", mime="text/csv", ) # Offline insights if show_offline_insights: st.subheader("đ Offline Insights (no LLM)") off = compute_offline_insights(df, income=user_monthly_income) st.json(off) # LLM recommendations st.subheader("đ§ LLM Recommendations") if GENAI_READY: if st.button("Generate Recommendations (Gemini)"): with st.spinner("Generating recommendations via Gemini..."): md = generate_recommendations_gemini(df, income=user_monthly_income, max_rows=int(chunk_limit)) st.markdown(md or "â ī¸ No insights returned.") st.download_button( label="âŦī¸ Download Recommendations (Markdown)", data=(md or "").encode("utf-8"), file_name="financial_recommendations.md", mime="text/markdown", ) st.markdown('', unsafe_allow_html=True) try: st.balloons() except Exception: pass else: st.info("Provide GEMINI_API_KEY in environment to enable AI-powered recommendations.") else: st.info("Upload at least one PDF to begin.")