csv Data_4

app.py

@@ -75,6 +75,310 @@ 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()
@@ -1185,9 +1489,7 @@ 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."""
@@ -1344,70 +1646,12 @@ async def csv_prescription_activity(request: Request):
  
 @app.post("/csv/cashflow/data")
 async def csv_cashflow_data(request: Request):
-    """Cash flow bubble data from CSV transactions."""
+    """Cash flow data from user-uploaded CSV transactions."""
     body = await request.json()
     transactions = body.get("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),
-    }
- 
+    return _build_cashflow_data(transactions)
+
+
  
 @app.post("/csv/cashflow/opinion")
 async def csv_cashflow_opinion(request: Request):
@@ -1448,56 +1692,73 @@ 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 CSV transactions."""
+    """Year Sphere data from user-uploaded CSV transactions."""
+    from plaid_client import get_recurring_from_csv
+    
     body = await request.json()
     transactions = body.get("transactions", [])
     
-    from collections import defaultdict
-    from datetime import date
+    if not transactions:
+        return {"months": [], "current_month_index": 0}
     
-    # Group by month
-    monthly = defaultdict(lambda: {'income': 0, 'expenses': 0, 'transactions': []})
+    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": []}
     
-    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,
-        })
+    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
     
-    # 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],
-        })
+    body = await request.json()
+    transactions = body.get("transactions", [])
+    
+    if not transactions:
+        return {"transactions": []}
     
-    return {'months': months}
+    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),
+    }

fiscal.py

@@ -383,7 +383,7 @@ def categorize_transactions_with_ai(descriptions: list) -> list:
         print(f"Categorization error: {e}", flush=True)
         return ['OTHER'] * len(descriptions)
 
-
+ 
 def categorize_transactions_batched(descriptions: list, batch_size: int = 40) -> list:
     """
     Categorize a large list of descriptions by batching.

plaid_client.py

@@ -62,6 +62,7 @@ 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):
     """
@@ -99,8 +100,6 @@ 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:
@@ -621,7 +620,6 @@ def get_recurring_from_fixtures() -> dict:
         }
 
 
-
 def _build_synthetic_balances_from_transactions(transactions: list) -> dict:
     """
     CSV mode: infer account balances from transaction history.
@@ -716,6 +714,4 @@ def get_recurring_from_csv(transactions: list) -> dict:
             'recurring_income': [],
             'projected_events': [],
             'analysis_period': None,
-        }
- 
- 
+        }