from dotenv import load_dotenv load_dotenv() from fastapi.responses import StreamingResponse, HTMLResponse import json import requests from fastapi import FastAPI, HTTPException, Depends, Request from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from fiscal import get_answer, clear_memory, stream_answer from auth import verify_token from plaid.model.country_code import CountryCode from plaid.exceptions import ApiException from plaid_client import get_financial_snapshot, create_link_token, exchange_public_token import os import time _financial_cache: dict = {} app = FastAPI(title="FISCAL API") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) security = HTTPBearer() SANDBOX_ACCESS_TOKEN = os.environ["PLAID_ACCESS_TOKEN"] SUPABASE_URL = os.environ.get("SUPABASE_URL", "") SERVICE_KEY = os.environ.get("SUPABASE_SERVICE_KEY", "") SUPABASE_HEADERS = { "apikey": SERVICE_KEY, "Authorization": f"Bearer {SERVICE_KEY}", "Content-Type": "application/json" } class ChatRequest(BaseModel): message: str access_token: str = "" class ChatResponse(BaseModel): answer: str class LinkTokenResponse(BaseModel): link_token: str class ExchangeTokenRequest(BaseModel): public_token: str @app.get("/") def root(): return {"status": "ok", "service": "FISCAL"} def get_user_access_token(user_id: str) -> str: """Fetch user's real Plaid access token from Supabase, fallback to sandbox.""" try: resp = requests.get( f"{SUPABASE_URL}/rest/v1/bank_connections?user_id=eq.{user_id}&limit=1", headers=SUPABASE_HEADERS ) connections = resp.json() if connections: return connections[0]["plaid_access_token"] except Exception as e: print(f"Could not fetch user token: {e}") return SANDBOX_ACCESS_TOKEN @app.get("/health") def health(): return {"status": "ok", "service": "FISCAL"} def _build_cashflow_data(all_transactions: list) -> dict: """ Shared cash flow data builder. Returns income + expense breakdown grouped by category and merchant. Used by both /trial/cashflow/data and /csv/cashflow/data. """ from datetime import date, timedelta from collections import defaultdict import re end_date = date.today() start_date = end_date - timedelta(days=30) income_total = 0.0 category_data = defaultdict(lambda: { "total": 0.0, "merchants": defaultdict(lambda: {"total": 0.0, "count": 0, "transactions": []}), "count": 0, }) for txn in all_transactions: txn_date_val = txn.get('date') if isinstance(txn_date_val, str): try: txn_date = date.fromisoformat(txn_date_val) except ValueError: continue else: txn_date = txn_date_val if txn_date < start_date or txn_date > end_date: continue amount = txn['amount'] name = txn.get('name', 'Unknown') category = txn.get('personal_finance_category', {}).get('primary', 'OTHER') if category in ['TRANSFER_IN', 'TRANSFER_OUT']: continue name_lower = name.lower() is_income = amount < 0 or ( category == 'LOAN_PAYMENTS' and any(w in name_lower for w in ['payroll', 'salary', 'direct dep', 'employer', 'wages']) ) if is_income: income_total += abs(amount) continue if category == 'LOAN_PAYMENTS': continue normalized_merchant = re.sub(r'[0-9#]+', '', name).strip() normalized_merchant = re.sub(r'\s+', ' ', normalized_merchant).upper() or name category_data[category]["total"] += amount category_data[category]["count"] += 1 category_data[category]["merchants"][normalized_merchant]["total"] += amount category_data[category]["merchants"][normalized_merchant]["count"] += 1 category_data[category]["merchants"][normalized_merchant]["transactions"].append({ "date": str(txn_date), "amount": round(amount, 2), "name": name, }) categories = [] for cat_name, cat_info in category_data.items(): merchants = [] for merchant_name, merchant_info in cat_info["merchants"].items(): merchants.append({ "name": merchant_name, "display_name": merchant_info["transactions"][0]["name"] if merchant_info["transactions"] else merchant_name, "total": round(merchant_info["total"], 2), "count": merchant_info["count"], "transactions": sorted( merchant_info["transactions"], key=lambda x: x["date"], reverse=True )[:10], }) merchants.sort(key=lambda x: -x["total"]) readable_name = cat_name.replace("_", " ").title() categories.append({ "name": readable_name, "raw_name": cat_name, "total": round(cat_info["total"], 2), "count": cat_info["count"], "merchants": merchants, }) categories.sort(key=lambda x: -x["total"]) return { "income_total": round(income_total, 2), "categories": categories, "date_range": { "start": str(start_date), "end": str(end_date), } } def _build_yearsphere_data(all_transactions: list, recurring_data: dict) -> dict: """ Shared Year Sphere data builder. Takes raw transactions + recurring analysis, returns full sphere data shape. Used by both /trial/yearsphere/data and /csv/yearsphere/data. """ from datetime import date from collections import defaultdict from calendar import monthrange recurring_expenses = recurring_data.get("recurring_expenses", []) recurring_income = recurring_data.get("recurring_income", []) projected_events = recurring_data.get("projected_events", []) today = date.today() # Build the month window: 8 months back through 3 months forward window = [] for offset in range(-8, 4): target_month = today.month + offset target_year = today.year while target_month < 1: target_month += 12 target_year -= 1 while target_month > 12: target_month -= 12 target_year += 1 window.append((target_year, target_month)) # Build per-month recurring index recurring_by_month = defaultdict(list) seen_month_names = set() def add_recurring_occurrence(year, month, day, name, amount, is_income): dedupe_key = (year, month, name.upper().strip(), day) if dedupe_key in seen_month_names: return seen_month_names.add(dedupe_key) recurring_by_month[(year, month)].append({ "name": name[:30], "amount": round(abs(amount), 2), "date": f"{year:04d}-{month:02d}-{day:02d}", "is_income": is_income, }) for item in recurring_expenses: for hist in item.get("history", []): hist_date = date.fromisoformat(hist["date"]) add_recurring_occurrence( hist_date.year, hist_date.month, hist_date.day, item["merchant"], hist["amount"], is_income=False, ) for item in recurring_income: for hist in item.get("history", []): hist_date = date.fromisoformat(hist["date"]) add_recurring_occurrence( hist_date.year, hist_date.month, hist_date.day, item["merchant"], hist["amount"], is_income=True, ) for ev in projected_events: ev_date = date.fromisoformat(ev["date"]) if ev_date <= today: continue add_recurring_occurrence( ev_date.year, ev_date.month, ev_date.day, ev.get("merchant", "Unknown"), ev.get("amount", 0.0), is_income=ev.get("is_income", False), ) # Build per-month output months_out = [] for (year, month) in window: month_start = date(year, month, 1) _, last_day = monthrange(year, month) month_end = date(year, month, last_day) is_past = month_end < today is_current = month_start <= today <= month_end is_future = month_start > today total_spent = 0.0 total_income = 0.0 daily_spending = defaultdict(float) daily_transactions = defaultdict(list) if not is_future: for txn in all_transactions: txn_date_val = txn.get('date') if isinstance(txn_date_val, str): try: txn_date = date.fromisoformat(txn_date_val) except ValueError: continue else: txn_date = txn_date_val if txn_date < month_start or txn_date > month_end: continue amount = txn['amount'] name = txn.get('name', 'Unknown') category = txn.get('personal_finance_category', {}).get('primary', 'OTHER') name_lower = name.lower() if category in ['TRANSFER_IN', 'TRANSFER_OUT']: continue is_income = amount < 0 or ( category == 'LOAN_PAYMENTS' and any(w in name_lower for w in ['payroll', 'salary', 'direct dep', 'employer', 'wages']) ) daily_transactions[str(txn_date)].append({ "name": name[:30], "amount": round(abs(amount), 2), "is_income": is_income, "category": category, }) if is_income: total_income += abs(amount) else: total_spent += amount daily_spending[str(txn_date)] += amount for rec in recurring_by_month.get((year, month), []): rec_date_str = rec["date"] already_in_day = any( t["name"].upper().strip() == rec["name"].upper().strip() for t in daily_transactions.get(rec_date_str, []) ) if not already_in_day: daily_transactions[rec_date_str].append({ "name": rec["name"], "amount": rec["amount"], "is_income": rec["is_income"], "category": "RECURRING", }) if is_future: if rec["is_income"]: total_income += rec["amount"] else: total_spent += rec["amount"] daily_spending[rec_date_str] += rec["amount"] month_recurring = recurring_by_month.get((year, month), []) recurring_income_items = [ {"name": r["name"], "amount": r["amount"], "date": r["date"]} for r in month_recurring if r["is_income"] ] recurring_expense_items = [ {"name": r["name"], "amount": r["amount"], "date": r["date"]} for r in month_recurring if not r["is_income"] ] if is_future: total_spent = 0.0 days_in_month = last_day max_day_spending = max(daily_spending.values()) if daily_spending else 1 daily_data = [] for day_num in range(1, days_in_month + 1): day_str = str(date(year, month, day_num)) spent = daily_spending.get(day_str, 0.0) intensity = spent / max_day_spending if max_day_spending > 0 else 0 daily_data.append({ "day": day_num, "amount": round(spent, 2), "intensity": round(intensity, 2), "transactions": daily_transactions.get(day_str, []), }) first_weekday = month_start.weekday() first_weekday = (first_weekday + 1) % 7 months_out.append({ "year": year, "month": month, "month_name": month_start.strftime("%b").upper(), "month_name_full": month_start.strftime("%B"), "total_spent": round(total_spent, 2), "total_income": round(total_income, 2), "is_past": is_past, "is_current": is_current, "is_future": is_future, "daily_data": daily_data, "first_weekday": first_weekday, "days_in_month": days_in_month, "recurring_income": sorted(recurring_income_items, key=lambda x: -x["amount"])[:5], "recurring_expenses": sorted(recurring_expense_items, key=lambda x: -x["amount"])[:10], }) current_idx = next((i for i, m in enumerate(months_out) if m["is_current"]), 8) return { "months": months_out, "current_month_index": current_idx, } def get_cached_financial_context(access_token: str, user_id: str) -> str: now = time.time() if user_id in _financial_cache: cached_time, cached_context = _financial_cache[user_id] if now - cached_time < 1800: # 30 minutes print("Using cached financial context", flush=True) return cached_context context = get_financial_snapshot(access_token) # Only cache successful responses if context not in ["NO_BANK_DATA", "BANK_REAUTH_REQUIRED"]: _financial_cache[user_id] = (now, context) return context class LoadHistoryRequest(BaseModel): messages: list @app.post("/load_history") def load_history( request: LoadHistoryRequest, credentials: HTTPAuthorizationCredentials = Depends(security), ): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Unauthorized") from fiscal import _get_memory memory = _get_memory(user["sub"]) memory.clear() # Rebuild memory from saved messages messages = request.messages for i in range(0, len(messages) - 1, 2): if i + 1 < len(messages): human_msg = messages[i].get("text", "") ai_msg = messages[i + 1].get("text", "") if human_msg and ai_msg: memory.chat_memory.add_user_message(human_msg) memory.chat_memory.add_ai_message(ai_msg) return {"status": "history loaded"} @app.post("/chat", response_model=ChatResponse) def chat( request: ChatRequest, credentials: HTTPAuthorizationCredentials = Depends(security), ): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Invalid or expired token") # Use token from request if provided, fallback to sandbox access_token = request.access_token if request.access_token else SANDBOX_ACCESS_TOKEN financial_context = get_cached_financial_context(access_token, user["sub"]) print(f"Financial context preview: {financial_context[:100]}", flush=True) answer = get_answer( message=request.message, user_id=user["sub"], financial_context=financial_context, ) return ChatResponse(answer=answer) @app.post("/reset") def reset(credentials: HTTPAuthorizationCredentials = Depends(security)): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Invalid or expired token") clear_memory(user["sub"]) return {"status": "conversation cleared"} @app.post("/chat/stream") def chat_stream( request: ChatRequest, credentials: HTTPAuthorizationCredentials = Depends(security), ): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Invalid or expired token") financial_context = "" bank_reauth = False try: access_token = request.access_token if request.access_token else SANDBOX_ACCESS_TOKEN print(f"Using access token: {access_token[:20]}...", flush=True) financial_context = get_cached_financial_context(access_token, user["sub"]) print(f"Financial context preview: {financial_context[:150]}", flush=True) if financial_context == "BANK_REAUTH_REQUIRED": bank_reauth = True financial_context = "" financial_context = financial_context.replace("{", "(").replace("}", ")") except Exception as e: print(f"Plaid fetch error: {e}", flush=True) financial_context = "Bank data temporarily unavailable." def generate(): if bank_reauth: yield f"data: {json.dumps({'chunk': '', 'reauth': True})}\n\n" yield "data: [DONE]\n\n" return for chunk in stream_answer( message=request.message, user_id=user["sub"], financial_context=financial_context, ): yield f"data: {json.dumps({'chunk': chunk})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream") @app.post("/plaid/create_link_token", response_model=LinkTokenResponse) def plaid_create_link_token( credentials: HTTPAuthorizationCredentials = Depends(security), ): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Unauthorized") try: token = create_link_token(user["sub"]) return LinkTokenResponse(link_token=token) except Exception as e: print(f"Link token error: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/plaid/exchange_token") def plaid_exchange_token( request: ExchangeTokenRequest, credentials: HTTPAuthorizationCredentials = Depends(security), ): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Unauthorized") try: access_token = exchange_public_token(request.public_token) # Return access_token to app — app will save it directly return {"status": "connected", "access_token": access_token} except Exception as e: print(f"Exchange token error: {e}") raise HTTPException(status_code=500, detail=str(e)) class UpdateLinkTokenRequest(BaseModel): access_token: str @app.post("/plaid/update_link_token") def plaid_update_link_token( request: UpdateLinkTokenRequest, credentials: HTTPAuthorizationCredentials = Depends(security), ): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Unauthorized") try: from plaid.model.link_token_create_request import LinkTokenCreateRequest from plaid.model.link_token_create_request_user import LinkTokenCreateRequestUser from plaid.model.country_code import CountryCode from plaid_client import plaid_client req = LinkTokenCreateRequest( client_name="Cheque Up", country_codes=[CountryCode('CA')], language='en', user=LinkTokenCreateRequestUser(client_user_id=user["sub"]), access_token=request.access_token, redirect_uri="https://chequeup.ca", # ← add this ) response = plaid_client.link_token_create(req) return {"link_token": response['link_token']} except Exception as e: print(f"Update link token error: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/plaid/chart_data") def plaid_chart_data( request: ChatRequest, credentials: HTTPAuthorizationCredentials = Depends(security), ): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Unauthorized") access_token = request.access_token if request.access_token else SANDBOX_ACCESS_TOKEN try: from plaid_client import get_chart_data data = get_chart_data(access_token) return data except Exception as e: print(f"Chart data error: {e}", flush=True) raise HTTPException(status_code=500, detail=str(e)) @app.delete("/account") def delete_account( credentials: HTTPAuthorizationCredentials = Depends(security), ): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Unauthorized") user_id = user["sub"] try: requests.delete( f"{SUPABASE_URL}/rest/v1/bank_connections?user_id=eq.{user_id}", headers=SUPABASE_HEADERS ) requests.delete( f"{SUPABASE_URL}/rest/v1/profiles?id=eq.{user_id}", headers=SUPABASE_HEADERS ) response = requests.delete( f"{SUPABASE_URL}/auth/v1/admin/users/{user_id}", headers=SUPABASE_HEADERS ) if response.status_code not in [200, 204]: raise Exception(f"Auth delete failed: {response.text}") return {"status": "account deleted"} except Exception as e: import traceback print(f"DELETE ACCOUNT ERROR: {type(e).__name__}: {str(e)}") print(traceback.format_exc()) raise HTTPException(status_code=500, detail=str(e)) @app.get("/delete", response_class=HTMLResponse) def delete_info(): return """

Delete Your FISCAL Account

You can delete your account directly from the FISCAL app:

  1. Open the FISCAL app
  2. Tap the settings icon (wrench) in the top right
  3. Tap Delete Account
  4. Confirm deletion

This will permanently delete your account and all associated data.

For assistance, contact us at support@mjproductions.com

""" @app.get("/privacy", response_class=HTMLResponse) def privacy_policy(): return """ FISCAL Privacy Policy

FISCAL Privacy Policy

Last updated: May 2026

Information We Collect

We collect your email address and a hashed password to create and manage your account.

Financial Data

With your permission, FISCAL accesses your bank account data via Plaid to provide personalized financial advice. This data is processed ephemerally — it is never stored in our databases and is only used in the moment to generate your AI response.

How We Use Your Data

Your financial data is shared with our AI provider (Featherless AI) solely to generate your personalized financial responses. It is not stored, sold, or used for advertising.

Data Storage

We store only your email address and account credentials securely via Supabase. Chat messages and financial data are never stored.

Data Deletion

You can delete your account and all associated data at any time from within the FISCAL app under Options → Delete Account.

Contact

For privacy questions, contact us at mjproductions594@gmail.com

""" @app.post("/plaid/debug_data") def plaid_debug_data( request: ChatRequest, credentials: HTTPAuthorizationCredentials = Depends(security), ): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Unauthorized") access_token = request.access_token if request.access_token else SANDBOX_ACCESS_TOKEN from plaid_client import plaid_client from plaid.model.accounts_balance_get_request import AccountsBalanceGetRequest from plaid.model.transactions_get_request import TransactionsGetRequest from plaid.model.transactions_get_request_options import TransactionsGetRequestOptions from datetime import date, timedelta # Raw balances bal_req = AccountsBalanceGetRequest(access_token=access_token) bal_resp = plaid_client.accounts_balance_get(bal_req) accounts_raw = [] for acc in bal_resp['accounts']: accounts_raw.append({ "name": acc['name'], "type": str(acc['type']), "subtype": str(acc['subtype']), "current": acc['balances']['current'], "available": acc['balances']['available'], "limit": acc['balances']['limit'], }) # Raw transactions (last 10) end_date = date.today() start_date = end_date - timedelta(days=30) txn_req = TransactionsGetRequest( access_token=access_token, start_date=start_date, end_date=end_date, options=TransactionsGetRequestOptions(count=20, include_personal_finance_category=True) ) txn_resp = plaid_client.transactions_get(txn_req) txns_raw = [] for txn in txn_resp['transactions'][:20]: txns_raw.append({ "date": str(txn['date']), "name": txn['name'], "amount": txn['amount'], "category": txn.get('personal_finance_category', {}).get('primary', 'UNKNOWN'), "account_id": txn['account_id'], }) return { "accounts": accounts_raw, "transactions": txns_raw, "total_transactions": txn_resp['total_transactions'], } @app.get("/plaid/canadian_banks") def get_canadian_banks( credentials: HTTPAuthorizationCredentials = Depends(security), ): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Unauthorized") from plaid.model.institutions_search_request import InstitutionsSearchRequest from plaid.model.institutions_search_request_options import InstitutionsSearchRequestOptions from plaid.model.products import Products from plaid_client import plaid_client banks = [] bank_names = ["Royal Bank", "TD Canada", "Bank of Montreal", "Scotiabank", "CIBC"] for name in bank_names: try: req = InstitutionsSearchRequest( query=name, country_codes=[CountryCode('CA')], products=[Products('transactions')], options=InstitutionsSearchRequestOptions( include_optional_metadata=True ) ) resp = plaid_client.institutions_search(req) if resp['institutions']: inst = resp['institutions'][0] banks.append({ "institution_id": inst['institution_id'], "name": inst['name'], "logo": inst.get('logo'), "color": inst.get('primary_color', '#38bdf8'), }) except Exception as e: print(f"Bank search error for {name}: {e}", flush=True) return {"banks": banks} @app.post("/plaid/recurring") def plaid_recurring( request: ChatRequest, credentials: HTTPAuthorizationCredentials = Depends(security), ): user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Unauthorized") access_token = request.access_token if request.access_token else SANDBOX_ACCESS_TOKEN from plaid_client import get_recurring_transactions return get_recurring_transactions(access_token) @app.post("/plaid/calendar_export") def calendar_export( request: ChatRequest, credentials: HTTPAuthorizationCredentials = Depends(security), ): """Generate an .ics file for recurring transactions.""" user = verify_token(credentials.credentials) if not user: raise HTTPException(status_code=401, detail="Unauthorized") access_token = request.access_token if request.access_token else SANDBOX_ACCESS_TOKEN from plaid_client import get_recurring_transactions data = get_recurring_transactions(access_token) # Build ICS file lines = [ "BEGIN:VCALENDAR", "VERSION:2.0", "PRODID:-//FISCAL//Calendar//EN", "CALSCALE:GREGORIAN", ] for event in data['projected_events']: if event['status'] != 'confirmed': continue d = event['date'].replace('-', '') prefix = "💰" if event['is_income'] else "💸" summary = f"{prefix} {event['merchant']} - ${event['amount']:.2f}" lines.extend([ "BEGIN:VEVENT", f"DTSTART;VALUE=DATE:{d}", f"DTEND;VALUE=DATE:{d}", f"SUMMARY:{summary}", f"DESCRIPTION:{event['frequency'].title()} - {event['category']}", "END:VEVENT", ]) lines.append("END:VCALENDAR") from fastapi.responses import Response return Response( content="\r\n".join(lines), media_type="text/calendar", headers={"Content-Disposition": "attachment; filename=fiscal-calendar.ics"} ) @app.post("/trial/chat/stream") async def trial_chat_stream(request: Request): body = await request.json() message = body.get("message", "") if not message: return {"error": "No message provided"} trial_user_id = f"trial-{request.client.host}" # Trial mode: load fixture data directly from plaid_client import get_snapshot_from_fixtures financial_context = get_snapshot_from_fixtures() def generate(): for token in stream_answer(message, trial_user_id, financial_context=financial_context): yield f"data: {json.dumps({'chunk': token})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream") @app.post("/trial/reset") async def trial_reset(request: Request): trial_user_id = f"trial-{request.client.host}" clear_memory(trial_user_id) return {"status": "ok"} @app.post("/trial/plaid/create_link_token") async def trial_create_link_token(request: Request): # Beta mode: no real Plaid link needed return {"link_token": "trial-mock-link-token"} @app.post("/trial/plaid/exchange_token") async def trial_exchange_token(request: Request): # Beta mode: return mock token immediately return {"access_token": "trial-mock-access-token"} @app.post("/trial/plaid/chart_data") async def trial_chart_data(request: Request): from plaid_client import get_chart_data_from_fixtures return get_chart_data_from_fixtures() @app.post("/trial/plaid/recurring") async def trial_recurring(request: Request): from plaid_client import get_recurring_from_fixtures return get_recurring_from_fixtures() @app.post("/trial/prescription/diagnose") async def trial_prescription_diagnose(request: Request): """ Streaming diagnosis for a prescription (account, income, or expense). Takes prescription context + optional follow-up question. """ body = await request.json() prescription_type = body.get("type", "account") # account | income | expense | loan prescription_name = body.get("name", "") prescription_amount = body.get("amount", 0) prescription_details = body.get("details", "") # frequency, months, subtype, etc. follow_up_question = body.get("question", "") # optional — for follow-ups trial_user_id = f"trial-{request.client.host}" # Load fixture context so the LLM sees the full financial picture from plaid_client import get_snapshot_from_fixtures financial_context = get_snapshot_from_fixtures() # Build the diagnosis prompt if follow_up_question: # Follow-up: user is asking about the already-diagnosed prescription prompt = f"""The user previously asked you to diagnose their {prescription_type}: "{prescription_name}" ({prescription_details}, currently ${prescription_amount:.2f}). They're now asking a follow-up question: "{follow_up_question}" Answer their question directly, keeping the same calm, doctor-patient tone. Reference the specific prescription details where relevant.""" else: prompt = f"""The user just handed you their {prescription_type} for a check-up: "{prescription_name}" ({prescription_details}, currently ${prescription_amount:.2f}). This is not generic financial advice. Your value depends on making the user feel: "It noticed something specific about ME." Ground every observation in their actual transaction data. REQUIREMENTS FOR SPECIFICITY (this is what separates you from a search engine): - Reference at least 2 specific numbers, dates, or patterns from their financial context - Use exact amounts and time ranges: "$214 over 60 days," "3 deposits of $500 in July," "withdrew $180 before payday each time" - Never make statements that could apply to anyone. Avoid: "this is a solid foundation," "consider automatic transfers," "keep up the good work" - If you can't ground a statement in their specific data, don't say it REQUIREMENTS FOR CAUTIOUS LANGUAGE: - Do not overclaim. The absence of large withdrawals is not proof of progress. State what you observe, not what you infer. - Use hedged phrasing: "based on the last 60 days," "from what I can see," "your balance has stayed around $X" (not "you're doing great with savings") - If you only see limited data, say so: "I only see about 60 days of activity here, so I can't tell yet if this is a longer trend" - Prefer observation over judgment. "Your deposits and withdrawals have roughly balanced" is better than "you're managing well." STRUCTURE (adapt as needed — don't force every section): **What I see** — Specific numbers and patterns from THEIR data. What has actually happened in this {prescription_type} recently? **What it might mean** — Cautiously interpret. Use hedged language when inferring. **What you could consider** — 1-2 specific next steps grounded in their actual pattern (not generic advice). If nothing specific to suggest, say so honestly. **What's going well** — Name something specific and observable that's working, even if small. Then, at the end, present clear next actions using this EXACT format (this becomes tap-to-ask buttons in the UI): [ACTIONS] - Set a savings goal - Ask about a specific transaction - Find money I could safely save - Set up a reminder [/ACTIONS] Include only 2-4 actions that make sense for THIS {prescription_type}. Skip actions that don't fit — for example, "Set a savings goal" doesn't apply to a chequing account. If there's no obvious next step, use just "Ask about this account". Under 200 words for the prose. Use markdown headers (**bold**) for each section. Keep sections short — 1-2 sentences each. CRITICAL: Respond with prose only for the diagnosis body. Do NOT include [CHART_DATA] blocks, JSON, or code fences. The [ACTIONS] block at the end is required.""" def generate(): for token in stream_answer(prompt, trial_user_id, financial_context=financial_context): yield f"data: {json.dumps({'chunk': token})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream") @app.post("/trial/prescription/activity") async def trial_prescription_activity(request: Request): """ Returns per-prescription heartbeat activity data (last 90 days of transactions related to this prescription). """ from plaid_client import _load_raw_transactions from datetime import date, timedelta import re body = await request.json() prescription_type = body.get("type", "account") prescription_name = body.get("name", "") try: fixture = _load_raw_transactions() all_transactions = fixture['transactions'] except Exception as e: print(f"Activity fixture error: {e}", flush=True) return {"points": [], "total_events": 0} end_date = date.today() start_date = end_date - timedelta(days=90) # Filter transactions relevant to this prescription relevant = [] name_lower = prescription_name.lower() for txn in all_transactions: txn_date_val = txn.get('date') if isinstance(txn_date_val, str): txn_date = date.fromisoformat(txn_date_val) else: txn_date = txn_date_val if txn_date < start_date or txn_date > end_date: continue merchant = txn.get('name', '').lower() category = txn.get('personal_finance_category', {}).get('primary', 'OTHER') amount = txn['amount'] # Matching logic per prescription type is_match = False if prescription_type in ('income', 'expense'): # Match merchant name loosely (normalized like recurring detection) normalized_prescription = re.sub(r'[0-9#]+', '', prescription_name).strip().upper() normalized_merchant = re.sub(r'[0-9#]+', '', txn.get('name', '')).strip().upper() if normalized_prescription in normalized_merchant or normalized_merchant in normalized_prescription: is_match = True elif prescription_type == 'account': # For accounts: show all spending activity as the pulse if category not in ['TRANSFER_IN', 'TRANSFER_OUT'] and amount > 0: is_match = True elif prescription_type == 'loan': # For loans: show loan payments if category == 'LOAN_PAYMENTS': is_match = True if is_match: relevant.append({ 'date': str(txn_date), 'amount': round(amount, 2), 'name': txn.get('name', 'Unknown'), 'is_income': amount < 0, }) # Group by date for the pulse chart daily = {} for r in relevant: d = r['date'] if d not in daily: daily[d] = {'date': d, 'total': 0, 'count': 0, 'is_income': r['is_income']} daily[d]['total'] += abs(r['amount']) daily[d]['count'] += 1 # Build the pulse array: one point per day over 90 days points = [] current = start_date while current <= end_date: day_str = str(current) if day_str in daily: points.append({ 'date': day_str, 'value': round(daily[day_str]['total'], 2), 'count': daily[day_str]['count'], 'is_income': daily[day_str]['is_income'], }) else: points.append({ 'date': day_str, 'value': 0, 'count': 0, 'is_income': False, }) current += timedelta(days=1) return { 'points': points, 'total_events': len(relevant), 'date_range': {'start': str(start_date), 'end': str(end_date)}, } @app.post("/trial/cashflow/data") async def trial_cashflow_data(request: Request): """ Returns income + expense breakdown grouped by category and merchant for the cash flow bubble visualization. """ from plaid_client import _load_raw_transactions from datetime import date, timedelta from collections import defaultdict import re try: fixture = _load_raw_transactions() all_transactions = fixture['transactions'] except Exception as e: print(f"Cashflow fixture error: {e}", flush=True) return {"income_total": 0, "categories": []} # Filter to last 30 days end_date = date.today() start_date = end_date - timedelta(days=30) income_total = 0.0 category_data = defaultdict(lambda: { "total": 0.0, "merchants": defaultdict(lambda: {"total": 0.0, "count": 0, "transactions": []}), "count": 0, }) for txn in all_transactions: txn_date_val = txn.get('date') if isinstance(txn_date_val, str): txn_date = date.fromisoformat(txn_date_val) else: txn_date = txn_date_val if txn_date < start_date or txn_date > end_date: continue amount = txn['amount'] name = txn.get('name', 'Unknown') category = txn.get('personal_finance_category', {}).get('primary', 'OTHER') # Skip transfers if category in ['TRANSFER_IN', 'TRANSFER_OUT']: continue # Identify income (negative amounts or payroll patterns) name_lower = name.lower() is_income = amount < 0 or ( category == 'LOAN_PAYMENTS' and any(w in name_lower for w in ['payroll', 'salary', 'direct dep', 'employer', 'wages']) ) if is_income: income_total += abs(amount) continue # Skip other loan payments (mortgage will get its own treatment via recurring) if category == 'LOAN_PAYMENTS': continue # Group into category → merchant normalized_merchant = re.sub(r'[0-9#]+', '', name).strip() normalized_merchant = re.sub(r'\s+', ' ', normalized_merchant).upper() or name category_data[category]["total"] += amount category_data[category]["count"] += 1 category_data[category]["merchants"][normalized_merchant]["total"] += amount category_data[category]["merchants"][normalized_merchant]["count"] += 1 category_data[category]["merchants"][normalized_merchant]["transactions"].append({ "date": str(txn_date), "amount": round(amount, 2), "name": name, }) # Convert to sorted list categories = [] for cat_name, cat_info in category_data.items(): # Build merchant list merchants = [] for merchant_name, merchant_info in cat_info["merchants"].items(): merchants.append({ "name": merchant_name, "display_name": merchant_info["transactions"][0]["name"] if merchant_info["transactions"] else merchant_name, "total": round(merchant_info["total"], 2), "count": merchant_info["count"], "transactions": sorted( merchant_info["transactions"], key=lambda x: x["date"], reverse=True )[:10], # Cap at 10 most recent }) merchants.sort(key=lambda x: -x["total"]) # Human-readable category name readable_name = cat_name.replace("_", " ").title() categories.append({ "name": readable_name, "raw_name": cat_name, "total": round(cat_info["total"], 2), "count": cat_info["count"], "merchants": merchants, }) categories.sort(key=lambda x: -x["total"]) return { "income_total": round(income_total, 2), "categories": categories, "date_range": { "start": str(start_date), "end": str(end_date), } } @app.post("/trial/cashflow/opinion") async def trial_cashflow_opinion(request: Request): """ Streaming Cheque Up opinion on cash flow at three levels: - overview: general assessment of the whole month - category: opinion on a specific spending category - merchant: opinion on a specific merchant within a category """ body = await request.json() level = body.get("level", "overview") # overview | category | merchant trial_user_id = f"trial-{request.client.host}" # Load fixture context from plaid_client import get_snapshot_from_fixtures financial_context = get_snapshot_from_fixtures() if level == "overview": income_total = body.get("income_total", 0) top_categories = body.get("top_categories", []) # [{"name": "Food", "total": 340, "count": 12}, ...] total_spent = sum(c.get("total", 0) for c in top_categories) remaining = income_total - total_spent cats_str = "\n".join([f"- {c['name']}: ${c['total']:.0f} ({c['count']} transactions)" for c in top_categories[:6]]) prompt = f"""The user is looking at their cash flow for the last 30 days. Income this month: ${income_total:.2f} Top spending categories: {cats_str} Total spent: ${total_spent:.2f} Remaining: ${remaining:.2f} Give a warm, doctor-patient style overview of the whole month. Cover: 1. What the balance looks like (income vs spending) 2. What's healthy or notable in the split 3. Which category stands out (biggest, or most movable) — just observation, not judgment Keep it under 90 words. Warm, calm, non-judgmental. No advice yet — just the "here's what I see" moment. Speak like a doctor reviewing a chart, not a financial advisor giving tips. CRITICAL: Respond with prose only. No headers, no bullets, no chart syntax, no lists.""" elif level == "category": cat_name = body.get("category_name", "") cat_total = body.get("category_total", 0) cat_count = body.get("category_count", 0) income_total = body.get("income_total", 0) top_merchants = body.get("top_merchants", []) merchants_str = "\n".join([f"- {m['name']}: ${m['total']:.0f} ({m['count']} times)" for m in top_merchants[:5]]) pct_of_income = (cat_total / income_total * 100) if income_total > 0 else 0 prompt = f"""The user just zoomed into their "{cat_name}" spending. Category: {cat_name} Total spent: ${cat_total:.2f} ({cat_count} transactions) Percentage of monthly income: {pct_of_income:.0f}% Top merchants in this category: {merchants_str} Give a warm, doctor-patient style diagnosis of this category. Cover: 1. Whether the amount is normal, high, or low for this category 2. What the merchant breakdown shows (concentration vs spread) 3. One small observation about the pattern — only if genuinely useful Keep under 90 words. Warm, calm, non-judgmental. Speak like a doctor examining a specific symptom. CRITICAL: Respond with prose only. No headers, no bullets, no chart syntax, no lists.""" elif level == "merchant": merchant_name = body.get("merchant_name", "") merchant_total = body.get("merchant_total", 0) merchant_count = body.get("merchant_count", 0) category_name = body.get("category_name", "") transactions = body.get("transactions", []) avg = merchant_total / max(merchant_count, 1) recent_str = "\n".join([f"- {t['date']}: ${t['amount']:.2f}" for t in transactions[:5]]) prompt = f"""The user just zoomed into a specific merchant: {merchant_name} Merchant: {merchant_name} Category: {category_name} Total spent here: ${merchant_total:.2f} Number of transactions: {merchant_count} Average per transaction: ${avg:.2f} Recent transactions: {recent_str} Give a warm, doctor-patient style opinion on this specific merchant. Cover: 1. What the pattern looks like (frequency, size) 2. Whether it's a habit worth noticing (recurring visits, high per-transaction cost) 3. Reassurance if it's fine, or a gentle prompt if there's something worth looking at Keep under 80 words. Warm, calm, non-judgmental, specific to this merchant. CRITICAL: Respond with prose only. No headers, no bullets, no chart syntax, no lists.""" else: return {"error": "Invalid level"} def generate(): for token in stream_answer(prompt, trial_user_id, financial_context=financial_context, use_memory=False): yield f"data: {json.dumps({'chunk': token})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream") @app.post("/trial/yearsphere/data") async def trial_yearsphere_data(request: Request): """ Returns 12 months of data for the Year Sphere. Consumes get_recurring_from_fixtures() — same source as the working calendar. Recurring items are placed on the days they historically hit (past) or are projected to hit (future). """ from plaid_client import _load_raw_transactions, get_recurring_from_fixtures from datetime import date from collections import defaultdict from calendar import monthrange try: fixture = _load_raw_transactions() all_transactions = fixture['transactions'] except Exception as e: print(f"YearSphere fixture error: {e}", flush=True) return {"months": [], "current_month_index": 0} # Pull the same recurring data the working calendar uses try: recurring_data = get_recurring_from_fixtures() except Exception as e: print(f"YearSphere recurring fetch error: {e}", flush=True) recurring_data = {"recurring_expenses": [], "recurring_income": [], "projected_events": []} recurring_expenses = recurring_data.get("recurring_expenses", []) recurring_income = recurring_data.get("recurring_income", []) projected_events = recurring_data.get("projected_events", []) today = date.today() # Build the month window: 8 months back through 3 months forward window = [] for offset in range(-8, 4): target_month = today.month + offset target_year = today.year while target_month < 1: target_month += 12 target_year -= 1 while target_month > 12: target_month -= 12 target_year += 1 window.append((target_year, target_month)) # --- Build a per-month index of recurring occurrences (both historical + projected) --- # Key: (year, month) -> list of {name, amount, date, is_income} recurring_by_month: dict = defaultdict(list) # Track which (year, month, normalized_name) we've seen so we don't double-list seen_month_names: set = set() def add_recurring_occurrence(year: int, month: int, day: int, name: str, amount: float, is_income: bool): # Guard against duplicate entries for the same recurring item in the same month # (e.g. biweekly items hit twice — those are legit, don't dedupe those; dedupe monthly ones) dedupe_key = (year, month, name.upper().strip(), day) if dedupe_key in seen_month_names: return seen_month_names.add(dedupe_key) recurring_by_month[(year, month)].append({ "name": name[:30], "amount": round(abs(amount), 2), "date": f"{year:04d}-{month:02d}-{day:02d}", "is_income": is_income, }) # Add historical occurrences from the recurring_expenses / recurring_income "history" arrays for item in recurring_expenses: for hist in item.get("history", []): hist_date = date.fromisoformat(hist["date"]) add_recurring_occurrence( hist_date.year, hist_date.month, hist_date.day, item["merchant"], hist["amount"], is_income=False, ) for item in recurring_income: for hist in item.get("history", []): hist_date = date.fromisoformat(hist["date"]) add_recurring_occurrence( hist_date.year, hist_date.month, hist_date.day, item["merchant"], hist["amount"], is_income=True, ) # Add projected future events for ev in projected_events: ev_date = date.fromisoformat(ev["date"]) if ev_date <= today: continue # Only add projections for future dates add_recurring_occurrence( ev_date.year, ev_date.month, ev_date.day, ev.get("merchant", "Unknown"), ev.get("amount", 0.0), is_income=ev.get("is_income", False), ) # --- Build per-month output --- months_out = [] for (year, month) in window: month_start = date(year, month, 1) _, last_day = monthrange(year, month) month_end = date(year, month, last_day) is_past = month_end < today is_current = month_start <= today <= month_end is_future = month_start > today total_spent = 0.0 total_income = 0.0 daily_spending = defaultdict(float) daily_transactions = defaultdict(list) # Real transactions from fixture (only for past + current) if not is_future: for txn in all_transactions: txn_date_val = txn.get('date') if isinstance(txn_date_val, str): txn_date = date.fromisoformat(txn_date_val) else: txn_date = txn_date_val if txn_date < month_start or txn_date > month_end: continue amount = txn['amount'] name = txn.get('name', 'Unknown') category = txn.get('personal_finance_category', {}).get('primary', 'OTHER') name_lower = name.lower() if category in ['TRANSFER_IN', 'TRANSFER_OUT']: continue is_income = amount < 0 or ( category == 'LOAN_PAYMENTS' and any(w in name_lower for w in ['payroll', 'salary', 'direct dep', 'employer', 'wages']) ) daily_transactions[str(txn_date)].append({ "name": name[:30], "amount": round(abs(amount), 2), "is_income": is_income, "category": category, }) if is_income: total_income += abs(amount) else: total_spent += amount daily_spending[str(txn_date)] += amount # Overlay recurring occurrences (historical for past/current, projected for future) # These are added on top of raw transactions so future months get populated, # and past months get any recurring occurrences that the raw fixture might miss. for rec in recurring_by_month.get((year, month), []): rec_date_str = rec["date"] already_in_day = any( t["name"].upper().strip() == rec["name"].upper().strip() for t in daily_transactions.get(rec_date_str, []) ) if not already_in_day: daily_transactions[rec_date_str].append({ "name": rec["name"], "amount": rec["amount"], "is_income": rec["is_income"], "category": "RECURRING", }) if is_future: if rec["is_income"]: total_income += rec["amount"] else: total_spent += rec["amount"] daily_spending[rec_date_str] += rec["amount"] # For the recurring lists at the bottom of the zoomed view, use the recurring_by_month entries month_recurring = recurring_by_month.get((year, month), []) recurring_income_items = [ {"name": r["name"], "amount": r["amount"], "date": r["date"]} for r in month_recurring if r["is_income"] ] recurring_expense_items = [ {"name": r["name"], "amount": r["amount"], "date": r["date"]} for r in month_recurring if not r["is_income"] ] # Future months: total_spent stays 0 so the sphere stays flat # (recurring projections were added above but we reset total_spent to 0 for the elevation math) if is_future: total_spent = 0.0 # Build daily_data days_in_month = last_day max_day_spending = max(daily_spending.values()) if daily_spending else 1 daily_data = [] for day_num in range(1, days_in_month + 1): day_str = str(date(year, month, day_num)) spent = daily_spending.get(day_str, 0.0) intensity = spent / max_day_spending if max_day_spending > 0 else 0 daily_data.append({ "day": day_num, "amount": round(spent, 2), "intensity": round(intensity, 2), "transactions": daily_transactions.get(day_str, []), }) first_weekday = month_start.weekday() first_weekday = (first_weekday + 1) % 7 # Sun=0 months_out.append({ "year": year, "month": month, "month_name": month_start.strftime("%b").upper(), "month_name_full": month_start.strftime("%B"), "total_spent": round(total_spent, 2), "total_income": round(total_income, 2), "is_past": is_past, "is_current": is_current, "is_future": is_future, "daily_data": daily_data, "first_weekday": first_weekday, "days_in_month": days_in_month, "recurring_income": sorted(recurring_income_items, key=lambda x: -x["amount"])[:5], "recurring_expenses": sorted(recurring_expense_items, key=lambda x: -x["amount"])[:10], }) current_idx = next((i for i, m in enumerate(months_out) if m["is_current"]), 8) return { "months": months_out, "current_month_index": current_idx, } @app.post("/csv/chart_data") async def csv_chart_data(request: Request): """Chart data from user-uploaded CSV transactions.""" from plaid_client import get_chart_data_from_csv body = await request.json() transactions = body.get("transactions", []) return get_chart_data_from_csv(transactions) @app.post("/csv/recurring") async def csv_recurring(request: Request): """Recurring pattern detection from user-uploaded CSV transactions.""" from plaid_client import get_recurring_from_csv body = await request.json() transactions = body.get("transactions", []) return get_recurring_from_csv(transactions) @app.post("/csv/chat/stream") async def csv_chat_stream(request: Request): """Chat with AI using CSV-uploaded transactions as context.""" from plaid_client import get_snapshot_from_csv body = await request.json() message = body.get("message", "") transactions = body.get("transactions", []) csv_user_id = f"csv-{request.client.host}" financial_context = get_snapshot_from_csv(transactions) def generate(): for token in stream_answer(message, csv_user_id, financial_context=financial_context): yield f"data: {json.dumps({'chunk': token})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream") @app.post("/csv/reset") async def csv_reset(request: Request): """Reset chat history for CSV mode.""" csv_user_id = f"csv-{request.client.host}" reset_history(csv_user_id) return {"status": "reset"} @app.post("/csv/prescription/diagnose") async def csv_prescription_diagnose(request: Request): """Streaming diagnosis using CSV transaction context.""" from plaid_client import get_snapshot_from_csv body = await request.json() prescription_type = body.get("type", "account") prescription_name = body.get("name", "") prescription_amount = body.get("amount", 0) prescription_details = body.get("details", "") follow_up_question = body.get("question", "") transactions = body.get("transactions", []) csv_user_id = f"csv-{request.client.host}" financial_context = get_snapshot_from_csv(transactions) if follow_up_question: prompt = f"""The user previously asked you to diagnose their {prescription_type}: "{prescription_name}" ({prescription_details}, currently ${prescription_amount:.2f}). They're now asking a follow-up question: "{follow_up_question}" Answer their question directly, keeping the same calm, doctor-patient tone. Reference the specific prescription details where relevant.""" else: prompt = f"""The user just handed you their {prescription_type} for a check-up: "{prescription_name}" ({prescription_details}, currently ${prescription_amount:.2f}). Give a warm, doctor-patient style diagnosis that covers: 1. **What it is** — a plain-language description of this {prescription_type} 2. **The amount** — acknowledge the current amount without judgment 3. **How it looks** — is this amount healthy, concerning, or somewhere in between? Be honest but calm. 4. **Recent history** — mention what you can observe from their recent activity 5. **What could go better** — 1-2 specific, actionable suggestions (only if genuinely useful) 6. **Reassurance** — even if there's room to improve, name what's going right. If the account is in good shape, celebrate it warmly. Keep the tone calm, non-judgmental, and specific to this {prescription_type}. Don't lecture. Don't use words like "leak" or "spike." Speak like a doctor who cares about the patient, not a financial advisor trying to upsell them. Under 200 words. Use markdown headers (**bold**) for each section. Keep sections short — 1-2 sentences each. CRITICAL: Respond with prose only. Do NOT include [CHART_DATA]...[/CHART_DATA] blocks, JSON, code fences, or any structured data.""" def generate(): for token in stream_answer(prompt, csv_user_id, financial_context=financial_context): yield f"data: {json.dumps({'chunk': token})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream") @app.post("/csv/prescription/activity") async def csv_prescription_activity(request: Request): """Activity data for the pulse chart from CSV transactions.""" body = await request.json() transactions = body.get("transactions", []) # Reuse the same activity aggregation logic as trial mode # Group transactions by day, count events from collections import defaultdict from datetime import date, timedelta end_date = date.today() start_date = end_date - timedelta(days=90) daily_activity = defaultdict(lambda: {'count': 0, 'total_expense': 0, 'total_income': 0}) for txn in transactions: txn_date_str = txn.get('date', '') try: txn_date = date.fromisoformat(txn_date_str) except (ValueError, TypeError): continue if txn_date < start_date or txn_date > end_date: continue amount = txn.get('amount', 0) category = txn.get('personal_finance_category', {}).get('primary', '') # Skip transfers to avoid double-counting if category in ['TRANSFER_IN', 'TRANSFER_OUT']: continue key = txn_date.isoformat() daily_activity[key]['count'] += 1 if amount > 0: daily_activity[key]['total_expense'] += amount else: daily_activity[key]['total_income'] += abs(amount) # Build activity array activity = [] current = start_date while current <= end_date: key = current.isoformat() data = daily_activity.get(key, {'count': 0, 'total_expense': 0, 'total_income': 0}) activity.append({ 'date': key, 'value': round(data['total_expense'] + data['total_income'], 2), 'count': data['count'], 'is_income': data['total_income'] > data['total_expense'], }) current += timedelta(days=1) return { 'activity': activity, 'total_events': sum(a['count'] for a in activity), } @app.post("/csv/cashflow/data") async def csv_cashflow_data(request: Request): """Cash flow data from user-uploaded CSV transactions.""" body = await request.json() transactions = body.get("transactions", []) return _build_cashflow_data(transactions) @app.post("/csv/cashflow/opinion") async def csv_cashflow_opinion(request: Request): """AI opinion on cash flow, using CSV transaction context.""" from plaid_client import get_snapshot_from_csv body = await request.json() transactions = body.get("transactions", []) category_focus = body.get("category", None) csv_user_id = f"csv-{request.client.host}" financial_context = get_snapshot_from_csv(transactions) if category_focus: prompt = f"""The user is looking at their spending in the "{category_focus}" category. Give them a brief, warm perspective on it. Focus on: - What patterns you see in this category - Whether the amount feels reasonable given their overall picture - One specific observation (not a lecture) Keep it under 150 words. Calm, non-judgmental tone. No lists — just conversational prose.""" else: prompt = """Give the user a brief overview of their cash flow this month. Cover: - Overall picture (income vs. expenses) - The category that stands out most - One thing worth noticing (not a lecture, not advice unless it's genuinely useful) Keep it under 200 words. Warm, doctor-patient tone. Prose only, no lists.""" def generate(): for token in stream_answer(prompt, csv_user_id, financial_context=financial_context): yield f"data: {json.dumps({'chunk': token})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream") # ============================================================================= # NOW REPLACE the /csv/yearsphere/data endpoint with this: # ============================================================================= @app.post("/csv/yearsphere/data") async def csv_yearsphere_data(request: Request): """Year Sphere data from user-uploaded CSV transactions.""" from plaid_client import get_recurring_from_csv body = await request.json() transactions = body.get("transactions", []) if not transactions: return {"months": [], "current_month_index": 0} try: recurring_data = get_recurring_from_csv(transactions) except Exception as e: print(f"CSV yearsphere recurring error: {e}", flush=True) recurring_data = {"recurring_expenses": [], "recurring_income": [], "projected_events": []} return _build_yearsphere_data(transactions, recurring_data) @app.post("/csv/categorize") async def csv_categorize(request: Request): """Categorize PENDING transactions using the AI.""" from fiscal import categorize_transactions_batched body = await request.json() transactions = body.get("transactions", []) if not transactions: return {"transactions": []} pending_indices = [] pending_descriptions = [] for i, txn in enumerate(transactions): current_cat = txn.get('personal_finance_category', {}).get('primary', '') if current_cat == 'PENDING_CATEGORIZATION': pending_indices.append(i) desc = txn.get('raw_description') or txn.get('name', 'Unknown') pending_descriptions.append(desc) if not pending_descriptions: return { "transactions": transactions, "categorized_count": 0, "total_count": len(transactions), } print(f"Categorizing {len(pending_descriptions)} pending transactions...", flush=True) ai_categories = categorize_transactions_batched(pending_descriptions) for idx, category in zip(pending_indices, ai_categories): if 'personal_finance_category' not in transactions[idx]: transactions[idx]['personal_finance_category'] = {} transactions[idx]['personal_finance_category']['primary'] = category return { "transactions": transactions, "categorized_count": len(pending_descriptions), "total_count": len(transactions), }