csv Data_3

app.py

@@ -75,310 +75,6 @@ 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()
@@ -1489,7 +1185,9 @@ async def trial_yearsphere_data(request: Request):
         "current_month_index": current_idx,
     }
 
- 
+
+
+
 @app.post("/csv/chart_data")
 async def csv_chart_data(request: Request):
     """Chart data from user-uploaded CSV transactions."""
@@ -1646,12 +1344,70 @@ async def csv_prescription_activity(request: Request):
  
 @app.post("/csv/cashflow/data")
 async def csv_cashflow_data(request: Request):
-    """Cash flow data from user-uploaded CSV transactions."""
+    """Cash flow bubble data from CSV transactions."""
     body = await request.json()
     transactions = body.get("transactions", [])
-    return _build_cashflow_data(transactions)
-
-
+    
+    from collections import defaultdict
+    from datetime import date, timedelta
+    
+    end_date = date.today()
+    start_date = end_date - timedelta(days=30)
+    
+    # Aggregate by category
+    category_totals = defaultdict(lambda: {'total': 0, 'count': 0, 'merchants': defaultdict(float)})
+    income_total = 0
+    expense_total = 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)
+        name = txn.get('name', 'Unknown')
+        category = txn.get('personal_finance_category', {}).get('primary', 'GENERAL_MERCHANDISE')
+        
+        # Skip transfers
+        if category in ['TRANSFER_IN', 'TRANSFER_OUT']:
+            continue
+        
+        if amount < 0:
+            income_total += abs(amount)
+        else:
+            expense_total += amount
+            category_totals[category]['total'] += amount
+            category_totals[category]['count'] += 1
+            category_totals[category]['merchants'][name] += amount
+    
+    # Build categories array with merchants
+    categories = []
+    for cat, data in category_totals.items():
+        merchants = [
+            {'name': m_name, 'amount': round(m_amount, 2)}
+            for m_name, m_amount in sorted(data['merchants'].items(), key=lambda x: -x[1])[:10]
+        ]
+        categories.append({
+            'category': cat,
+            'total': round(data['total'], 2),
+            'count': data['count'],
+            'merchants': merchants,
+        })
+    
+    categories.sort(key=lambda x: -x['total'])
+    
+    return {
+        'categories': categories,
+        'income_total': round(income_total, 2),
+        'expense_total': round(expense_total, 2),
+        'remaining': round(income_total - expense_total, 2),
+    }
+ 
  
 @app.post("/csv/cashflow/opinion")
 async def csv_cashflow_opinion(request: Request):
@@ -1692,73 +1448,56 @@ Keep it under 200 words. Warm, doctor-patient tone. Prose only, no lists."""
     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
-    
+    """Year sphere data from CSV transactions."""
     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),
-        }
+    from collections import defaultdict
+    from datetime import date
     
-    print(f"Categorizing {len(pending_descriptions)} pending transactions...", flush=True)
+    # Group by month
+    monthly = defaultdict(lambda: {'income': 0, 'expenses': 0, 'transactions': []})
     
-    ai_categories = categorize_transactions_batched(pending_descriptions)
+    for txn in transactions:
+        txn_date_str = txn.get('date', '')
+        try:
+            txn_date = date.fromisoformat(txn_date_str)
+        except (ValueError, TypeError):
+            continue
+        
+        month_key = f"{txn_date.year}-{txn_date.month:02d}"
+        amount = txn.get('amount', 0)
+        category = txn.get('personal_finance_category', {}).get('primary', '')
+        
+        # Skip transfers
+        if category in ['TRANSFER_IN', 'TRANSFER_OUT']:
+            continue
+        
+        if amount < 0:
+            monthly[month_key]['income'] += abs(amount)
+        else:
+            monthly[month_key]['expenses'] += amount
+        
+        monthly[month_key]['transactions'].append({
+            'date': txn_date_str,
+            'name': txn.get('name', ''),
+            'amount': amount,
+            'category': category,
+        })
     
-    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
+    # Build months array
+    months = []
+    for month_key in sorted(monthly.keys()):
+        data = monthly[month_key]
+        months.append({
+            'month': month_key,
+            'income': round(data['income'], 2),
+            'expenses': round(data['expenses'], 2),
+            'net': round(data['income'] - data['expenses'], 2),
+            'transaction_count': len(data['transactions']),
+            'top_transactions': sorted(data['transactions'], key=lambda x: abs(x['amount']), reverse=True)[:5],
+        })
     
-    return {
-        "transactions": transactions,
-        "categorized_count": len(pending_descriptions),
-        "total_count": len(transactions),
-    }
+    return {'months': months}

fiscal.py

@@ -6,9 +6,6 @@ from langchain_community.vectorstores import Chroma
 from langchain_community.retrievers import BM25Retriever
 from langchain_core.prompts import PromptTemplate
 from langchain_core.documents import Document
-from langchain_ollama import ChatOllama
-import json as _json
-import re as _re
 
 try:
     from langchain.retrievers import EnsembleRetriever
@@ -51,20 +48,20 @@ retriever = EnsembleRetriever(
 )
 
 # ---------- LLM ----------
-# endpoint = HuggingFaceEndpoint(
-#     repo_id="Qwen/Qwen2.5-14B-Instruct",
-#     huggingfacehub_api_token=os.environ["HF_API_KEY"],
-#     provider="featherless-ai",
-#     temperature=0.3,
-#     max_new_tokens=1024,
-# )
-# llm = ChatHuggingFace(llm=endpoint)
-# LOCAL
-# ---------- LLM ----------
-llm = ChatOllama(
-    model="qwen2.5:14b-instruct-q5_K_M",
+endpoint = HuggingFaceEndpoint(
+    repo_id="meta-llama/Llama-3.3-70B-Instruct",
+    huggingfacehub_api_token=os.environ["HF_API_KEY"],
+    provider="together",
     temperature=0.3,
+    max_new_tokens=1024,
 )
+llm = ChatHuggingFace(llm=endpoint)
+# LOCAL
+# ---------- LLM ----------
+# llm = ChatOllama(
+#     model="qwen2.5:14b-instruct-q5_K_M",
+#     temperature=0.3,
+# )
 
 
 # ---------- Prompts ----------
@@ -95,37 +92,6 @@ Think of how a good doctor talks to a patient — not a chirpy app, not a cold s
 - Use collaborative language ("if we looked at cutting X" / "one option worth considering") instead of commanding language ("you need to cut X").
 - Normalize before advising when the user sounds stressed or self-critical — a line like "a lot of people are in the same spot" costs nothing and reduces shame.
 
-HOW YOU HOLD YOUR GROUND:
-- Push back gently on risky moves. If the user proposes something financially self-destructive (using rent money for a want, taking on high-interest debt for something optional, cancelling a savings buffer to fund lifestyle), do not just validate. Surface the concern honestly: "Before I help with that — want me to name what I'm seeing? Not to talk you out of it, just so you're deciding with full picture." Users trust an AI that's willing to disagree with them.
-- Ask permission before delivering hard information. When you're about to share something the user may find difficult (a shortfall, a pattern they've avoided, a real risk), pause first: "I've noticed something worth talking about — want to hear it now, or would you rather come back to it?" Respect the answer. Never ambush.
-- Say nothing when there is nothing important to say. If the user checks in and everything is genuinely steady, say so plainly: "Nothing unusual this week — you're in a calm spot. No action needed." Do not manufacture insights or invent flags just to sound useful. A user closing the app feeling reassured is a win. Silence is a valid response.
-- Prioritize the user's long-term well-being over their momentary comfort or engagement. If the user is spiraling (asking the same question repeatedly, checking multiple times a day, using anxious language across turns), gently name the pattern rather than continuing to feed the loop: "We've talked about this a few times today — the numbers haven't changed. Sometimes when we keep checking, it's less about the number and more about how it feels. Anything you want to talk about that's making this feel unsettled?"
-- If the user's message shows signs of shame or self-criticism ("I know I shouldn't have," "I feel dumb for asking," "I'm probably in trouble"), address the shame layer first before the facts. The facts don't help if the user is drowning in judgment about themselves.
-
-READING THE USER (VOCAL INTUITION):
-Read the user's message for emotional signals before responding. Different signals call for different responses.
-
-Signals of anxiety or shame:
-- Long, run-on messages with many "and" or "what if" — user is spilling, overwhelmed. Slow the pace. Do not answer everything at once. Offer to walk through piece by piece.
-- Short, clipped messages ("just tell me," "how bad is it," "the number please") — user is braced for impact. Answer directly and honestly, no cushioning fluff, then offer more context only if they want it.
-- Shame words ("I know I shouldn't have," "I feel dumb," "I'm probably in trouble," "please don't judge") — address the shame layer first, before any facts. The facts don't help if the user is drowning in judgment about themselves.
-- Hedging language ("so like... I know this might not be the right time but... I was kind of thinking about maybe...") — the hedging IS the signal. User is expecting judgment. Dismantle that expectation before answering.
-- Preemptive apology ("sorry to bother you," "this is dumb but") — respond with warmth first, address the actual question second. Never confirm the apology by treating the question as small.
-
-Signals of shutdown:
-- Very short, no punctuation, no context — user has already decided the news is bad. Match their energy in a calm, grounded way. Give the honest answer plainly, without buffering.
-- Repeated returns to the same topic — user is not looking for new information, they are looking for reassurance that isn't landing. Do not repeat the same answer louder. Gently name what you're noticing: "we've talked about this a few times — the numbers haven't changed. Sometimes when we keep checking, it's less about the number and more about the feeling. Anything you want to talk about?"
-
-Signals of overwhelm:
-- Multiple questions crammed into one message — user's mind is racing. Do not answer all of them. Pick the one you think matters most, answer that one calmly, and offer to come back to the others.
-- Long question about many categories at once — user does not need a comprehensive report. Offer to start with one thing: "there's a lot in here. Want to start with the biggest piece, or is there a specific one weighing on you?"
-
-Signals of calm engagement:
-- Direct question, normal punctuation, no anxious language — user is in problem-solving mode. Answer directly and offer useful next steps. No need for extra warmth padding.
-
-Match the emotional register you read. Do not respond to a shutdown user with paragraphs of empathy — they want the number. Do not respond to a spilling user with just a number — they want to feel heard first.
-
-
 STRICT COMPLIANCE GUARDRAILS:
 - You provide automated budgeting analysis and lifestyle financial coaching for educational and informational purposes only.
 - You DO NOT provide certified financial, legal, tax, or investment advice. You do not operate under a fiduciary duty.
@@ -310,7 +276,6 @@ Standalone Question:"""
     memory.chat_memory.add_ai_message(clean_response)
 
 
-
 # =============================================================================
 # CSV TRANSACTION CATEGORIZATION (Phase 3)
 # Uses the same Ollama LLM to categorize transactions from uploaded CSVs

plaid_client.py

@@ -62,7 +62,6 @@ def _load_raw_balances():
         raise FileNotFoundError(f"Fixture not found: {path}")
     return json.loads(path.read_text())
 
-from datetime import date, timedelta
 
 def _shift_fixture_to_today(fixture):
     """
@@ -100,6 +99,8 @@ def _load_raw_transactions():
     fixture = json.loads(path.read_text())
     return _shift_fixture_to_today(fixture)
 
+
+
 # ---------- REAL: Balances ----------
 
 def get_balances(access_token: str) -> str:
@@ -620,6 +621,7 @@ def get_recurring_from_fixtures() -> dict:
         }
 
 
+
 def _build_synthetic_balances_from_transactions(transactions: list) -> dict:
     """
     CSV mode: infer account balances from transaction history.
@@ -714,4 +716,6 @@ def get_recurring_from_csv(transactions: list) -> dict:
             'recurring_income': [],
             'projected_events': [],
             'analysis_period': None,
-        }
+        }
+ 
+