Phase 6: Daily Intelligence

backend/agents/analyst_agent.py

@@ -75,9 +75,7 @@ def _build_analyst_brief(
                 "title": i.get("summary"),
                 "created_at": i.get("created_at"),
                 "similarity": (
-                    round(i.get("similarity", 0), 3)
-                    if i.get("similarity") is not None
-                    else None
+                    round(i.get("similarity", 0), 3) if i.get("similarity") is not None else None
                 ),
             }
             for i in past_insights
@@ -90,28 +88,53 @@ def _build_analyst_brief(
         f"RELEVANT PAST DECISIONS"
         f" (semantic memory — ranked by relevance to current situation):"
         f"\n{recent_memory}\n\n"
-        f"CURRENT ACCOUNTS:\n{json.dumps([
-            {'type': a.get('account_type'), 'balance': a.get('balance'),
-             'credit_limit': a.get('credit_limit')}
-            for a in accounts
-        ], indent=2)}\n\n"
-        f"UPCOMING BILLS:\n{json.dumps([
-            {'merchant': b.get('merchant'), 'amount': b.get('avg_amount'),
-             'due': b.get('next_due_date'), 'enrichment': b.get('ai_enrichment')}
-            for b in bills
-        ], indent=2)}\n\n"
-        f"ACTIVE SUBSCRIPTIONS:\n{json.dumps([
-            {'merchant': s.get('merchant'), 'amount': s.get('avg_amount'),
-             'price_change_pct': s.get('price_change_pct'),
-             'enrichment': s.get('ai_enrichment')}
-            for s in subscriptions
-        ], indent=2)}\n\n"
+        f"CURRENT ACCOUNTS:\n{
+            json.dumps(
+                [
+                    {
+                        'type': a.get('account_type'),
+                        'balance': a.get('balance'),
+                        'credit_limit': a.get('credit_limit'),
+                    }
+                    for a in accounts
+                ],
+                indent=2,
+            )
+        }\n\n"
+        f"UPCOMING BILLS:\n{
+            json.dumps(
+                [
+                    {
+                        'merchant': b.get('merchant'),
+                        'amount': b.get('avg_amount'),
+                        'due': b.get('next_due_date'),
+                        'enrichment': b.get('ai_enrichment'),
+                    }
+                    for b in bills
+                ],
+                indent=2,
+            )
+        }\n\n"
+        f"ACTIVE SUBSCRIPTIONS:\n{
+            json.dumps(
+                [
+                    {
+                        'merchant': s.get('merchant'),
+                        'amount': s.get('avg_amount'),
+                        'price_change_pct': s.get('price_change_pct'),
+                        'enrichment': s.get('ai_enrichment'),
+                    }
+                    for s in subscriptions
+                ],
+                indent=2,
+            )
+        }\n\n"
         f"SPENDING SUMMARY (last 90 days by category):\n{json.dumps(tx_summary, indent=2)}\n\n"
         f"Generate 3-5 analyst decisions AND an updated user profile.\n\n"
         f"Return this exact JSON shape:\n"
-        f'{{\n'
+        f"{{\n"
         f'  "decisions": [\n'
-        f'    {{\n'
+        f"    {{\n"
         f'      "signal_type": "<behavioral|risk|opportunity|anomaly|subscription>",\n'
         f'      "severity": "<info|warning|critical>",\n'
         f'      "title": "<specific, concrete title>",\n'
@@ -120,17 +143,17 @@ def _build_analyst_brief(
         f'      "simulation": "<forward projection>",\n'
         f'      "confidence": <float 0-1>,\n'
         f'      "sources": ["<data sources>"]\n'
-        f'    }}\n'
-        f'  ],\n'
+        f"    }}\n"
+        f"  ],\n"
         f'  "updated_profile": {{\n'
         f'    "income_pattern": {{}},\n'
         f'    "spending_baselines": {{}},\n'
         f'    "behavioral_patterns": [],\n'
         f'    "known_risks": [],\n'
         f'    "analyst_notes": "<running notes about this user>",\n'
         f'    "resolved_patterns": []\n'
-        f'  }}\n'
-        f'}}'
+        f"  }}\n"
+        f"}}"
     )
 
 
@@ -166,11 +189,13 @@ def analyst_node(state: IntelligenceState) -> dict:
         message = client.messages.create(
             model="claude-sonnet-4-6",
             max_tokens=8096,
-            system=[{
-                "type": "text",
-                "text": _ANALYST_SYSTEM_PROMPT,
-                "cache_control": {"type": "ephemeral"},
-            }],
+            system=[
+                {
+                    "type": "text",
+                    "text": _ANALYST_SYSTEM_PROMPT,
+                    "cache_control": {"type": "ephemeral"},
+                }
+            ],
             messages=[{"role": "user", "content": brief}],
         )
         response_text = message.content[0].text

backend/agents/chat.py

@@ -93,31 +93,32 @@ def _extract_prediction_block(response_text: str) -> dict | None:
 
 def _log_prediction(user_id: str, prediction: dict) -> None:
     supabase = get_supabase()
-    supabase.table("ai_predictions").insert({
-        "id": str(uuid.uuid4()),
-        "user_id": user_id,
-        "prediction_type": prediction["prediction_type"],
-        "prediction_payload": prediction["payload"],
-        "predicted_at": datetime.now(UTC).isoformat(),
-        "resolves_at": prediction["resolves_at"],
-    }).execute()
+    supabase.table("ai_predictions").insert(
+        {
+            "id": str(uuid.uuid4()),
+            "user_id": user_id,
+            "prediction_type": prediction["prediction_type"],
+            "prediction_payload": prediction["payload"],
+            "predicted_at": datetime.now(UTC).isoformat(),
+            "resolves_at": prediction["resolves_at"],
+        }
+    ).execute()
 
 
 def _retrieve_chat_context(user_id: str, query: str) -> dict:
-    supervisor_result = supervisor_graph.invoke({
-        "user_id": user_id,
-        "query": query,
-        "selected_tools": [],
-        "tool_results": {},
-    })
+    supervisor_result = supervisor_graph.invoke(
+        {
+            "user_id": user_id,
+            "query": query,
+            "selected_tools": [],
+            "tool_results": {},
+        }
+    )
 
     supabase = get_supabase()
 
     profile_result = (
-        supabase.table("user_financial_profiles")
-        .select("profile")
-        .eq("user_id", user_id)
-        .execute()
+        supabase.table("user_financial_profiles").select("profile").eq("user_id", user_id).execute()
     )
     profile = profile_result.data[0]["profile"] if profile_result.data else {}
 

backend/agents/enrichment_agent.py

@@ -47,20 +47,20 @@ def _build_enrichment_prompt(bills: list[dict], subscriptions: list[dict]) -> st
         f"Bills:\n{bills_section}\n\n"
         f"Subscriptions:\n{subs_section}\n\n"
         f"Return a JSON object with this exact shape:\n"
-        f'{{\n'
+        f"{{\n"
         f'  "bills": [\n'
-        f'    {{\n'
+        f"    {{\n"
         f'      "id": "<bill id>",\n'
         f'      "enrichment": {{\n'
         f'        "ai_confidence": <float 0-1>,\n'
         f'        "merchant_context": "<what this merchant sells>",\n'
         f'        "classification_note": "<why classified as recurring>",\n'
         f'        "is_subscription_candidate": <true|false>\n'
-        f'      }}\n'
-        f'    }}\n'
-        f'  ],\n'
+        f"      }}\n"
+        f"    }}\n"
+        f"  ],\n"
         f'  "subscriptions": [\n'
-        f'    {{\n'
+        f"    {{\n"
         f'      "id": "<subscription id>",\n'
         f'      "enrichment": {{\n'
         f'        "service_category": "<streaming|software|fitness|news|utilities|other>",\n'
@@ -69,10 +69,10 @@ def _build_enrichment_prompt(bills: list[dict], subscriptions: list[dict]) -> st
         f'        "price_trend_interpretation": "<plain English interpretation>",\n'
         f'        "cancel_recommendation": <true|false>,\n'
         f'        "cancel_reasoning": "<explanation or null>"\n'
-        f'      }}\n'
-        f'    }}\n'
-        f'  ]\n'
-        f'}}'
+        f"      }}\n"
+        f"    }}\n"
+        f"  ]\n"
+        f"}}"
     )
 
 
@@ -107,11 +107,13 @@ def enrichment_node(state: IntelligenceState) -> dict:
         message = client.messages.create(
             model="claude-sonnet-4-6",
             max_tokens=4096,
-            system=[{
-                "type": "text",
-                "text": _SYSTEM_PROMPT,
-                "cache_control": {"type": "ephemeral"},
-            }],
+            system=[
+                {
+                    "type": "text",
+                    "text": _SYSTEM_PROMPT,
+                    "cache_control": {"type": "ephemeral"},
+                }
+            ],
             messages=[{"role": "user", "content": prompt}],
         )
         response_text = message.content[0].text
@@ -152,9 +154,7 @@ def enrichment_node(state: IntelligenceState) -> dict:
         if "id" in e and "enrichment" in e
     }
 
-    enriched_bills = [
-        {**b, "ai_enrichment": enrichment_by_bill_id.get(b["id"])} for b in bills
-    ]
+    enriched_bills = [{**b, "ai_enrichment": enrichment_by_bill_id.get(b["id"])} for b in bills]
     enriched_subs = [
         {**s, "ai_enrichment": enrichment_by_sub_id.get(s["id"])} for s in subscriptions
     ]

backend/agents/tools.py

@@ -71,9 +71,7 @@ def _get_subscriptions_tool(user_id: str) -> dict:
 )
 def _get_spending_by_category_tool(user_id: str) -> dict:
     supabase = get_supabase()
-    accounts = (
-        supabase.table("accounts").select("id").eq("user_id", user_id).execute()
-    ).data or []
+    accounts = (supabase.table("accounts").select("id").eq("user_id", user_id).execute()).data or []
     account_ids = [a["id"] for a in accounts]
     if not account_ids:
         return {"transactions": []}
@@ -87,3 +85,47 @@ def _get_spending_by_category_tool(user_id: str) -> dict:
         .execute()
     )
     return {"transactions": result.data or []}
+
+
+@register_tool(
+    "get_safe_to_spend",
+    "Returns how much money is safe to spend today without risking upcoming bills",
+    keywords=["safe to spend", "safe amount", "how much can i spend", "available money"],
+)
+def _get_safe_to_spend_tool(user_id: str) -> dict:
+    supabase = get_supabase()
+    cached = (
+        supabase.table("safe_to_spend_cache")
+        .select("safe_amount, breakdown, computed_at")
+        .eq("user_id", user_id)
+        .execute()
+    ).data or []
+    if cached:
+        return cached[0]
+    return {"safe_amount": None, "breakdown": {}, "computed_at": None}
+
+
+@register_tool(
+    "get_pay_timing",
+    "Returns when and how much to pay on each credit card to stay under 8% utilization,"
+    " plus dangerous bill stacking windows",
+    keywords=["pay timing", "when to pay", "credit utilization", "bill stacking", "pay my card"],
+)
+def _get_pay_timing_tool(user_id: str) -> dict:
+    from engines.pay_timing import compute_pay_timing
+
+    supabase = get_supabase()
+    accounts = (
+        supabase.table("accounts")
+        .select("id, account_type, balance, credit_limit, closing_date")
+        .eq("user_id", user_id)
+        .execute()
+    ).data or []
+    bills = (
+        supabase.table("bills")
+        .select("avg_amount, next_due_date, merchant")
+        .eq("user_id", user_id)
+        .execute()
+    ).data or []
+    balance = sum(a.get("balance") or 0 for a in accounts if a.get("account_type") != "credit")
+    return compute_pay_timing(accounts=accounts, bills=bills, balance=balance)

backend/celery_app.py

@@ -16,6 +16,7 @@
         "tasks.detect_subscriptions",
         "tasks.categorize_transactions",
         "tasks.run_intelligence_pipeline",
+        "tasks.recompute_safe_to_spend",
     ],
 )
 
@@ -26,3 +27,13 @@
     timezone="UTC",
     enable_utc=True,
 )
+
+from celery.schedules import crontab  # noqa: E402
+
+celery.conf.beat_schedule = {
+    "nightly-safe-to-spend-recompute": {
+        "task": "tasks.recompute_safe_to_spend.recompute_safe_to_spend_for_user",
+        "schedule": crontab(hour=2, minute=0),
+        "args": [],
+    },
+}

backend/engines/pay_timing.py

@@ -0,0 +1,67 @@
+from collections import Counter
+from datetime import date, timedelta
+
+_TARGET_UTILIZATION = 0.08
+_STACK_WINDOW_DAYS = 3
+
+
+def infer_closing_date(transactions: list[dict]) -> int:
+    days = [int(t["timestamp"][8:10]) for t in transactions if t.get("timestamp")]
+    if not days:
+        return 1
+    return Counter(days).most_common(1)[0][0]
+
+
+def bills_in_window(bills: list[dict], window_days: int) -> list[dict]:
+    cutoff = date.today() + timedelta(days=window_days)
+    return [
+        b
+        for b in bills
+        if b.get("next_due_date") and date.fromisoformat(b["next_due_date"]) <= cutoff
+    ]
+
+
+def compute_pay_timing(
+    accounts: list[dict],
+    bills: list[dict],
+    balance: float,
+) -> dict:
+    card_recommendations = []
+    for acct in accounts:
+        if acct.get("account_type") != "credit":
+            continue
+        current_balance = acct.get("balance") or 0.0
+        limit = acct.get("credit_limit") or 0.0
+        if limit == 0:
+            continue
+        target_balance = round(limit * _TARGET_UTILIZATION, 2)
+        pay_amount = max(0.0, round(current_balance - target_balance, 2))
+
+        card_recommendations.append(
+            {
+                "account_id": acct["id"],
+                "pay_amount": pay_amount,
+                "target_utilization": _TARGET_UTILIZATION,
+                "closing_day": acct.get("closing_date"),
+            }
+        )
+
+    stacked_windows = []
+    today = date.today()
+    window_bills = bills_in_window(bills, _STACK_WINDOW_DAYS)
+    total_due = sum(b.get("avg_amount") or 0 for b in window_bills)
+    if window_bills and total_due > balance:
+        sorted_bills = sorted(window_bills, key=lambda b: b.get("avg_amount") or 0, reverse=True)
+        stacked_windows.append(
+            {
+                "window_start": today.isoformat(),
+                "window_end": (today + timedelta(days=_STACK_WINDOW_DAYS)).isoformat(),
+                "total_due": round(total_due, 2),
+                "priority_order": sorted_bills,
+            }
+        )
+
+    return {
+        "card_recommendations": card_recommendations,
+        "stacked_windows": stacked_windows,
+    }