Deploy from GitHub Actions: a7d40a05e1c8e0f277a3e01ef50ae8f13f98ee0a

src/agents/portfolio/agent.py

@@ -0,0 +1,32 @@
+"""
+Portfolio Advisor agent — reads wallet holdings and cross-references with
+web search to produce risk analysis and swap recommendations.
+"""
+
+import logging
+
+from langgraph.prebuilt import create_react_agent
+
+from src.agents.portfolio.prompt import PORTFOLIO_ADVISOR_SYSTEM_PROMPT
+from src.agents.portfolio.tools import get_tools as get_portfolio_tools
+from src.agents.search.tools import get_tools as get_search_tools
+
+logger = logging.getLogger(__name__)
+
+
+class PortfolioAdvisorAgent:
+    """Agent that analyses on-chain portfolio data and recommends actions."""
+
+    def __init__(self, llm):
+        self.llm = llm
+
+        # Combine portfolio tools + search tools so the agent can
+        # both read the wallet AND look up recent news for held tokens.
+        tools = get_portfolio_tools() + get_search_tools()
+
+        self.agent = create_react_agent(
+            model=llm,
+            tools=tools,
+            name="portfolio_advisor",
+            prompt=PORTFOLIO_ADVISOR_SYSTEM_PROMPT,
+        )

src/agents/portfolio/config.py

@@ -0,0 +1,53 @@
+"""
+Configuration for the Portfolio Advisor agent.
+
+Uses free, keyless APIs:
+  • Blockscout  — ETH, Polygon, Arbitrum, Base, Optimism
+  • Routescan   — Avalanche
+"""
+
+# ---------------------------------------------------------------------------
+# Blockscout instances (no API key required)
+# ---------------------------------------------------------------------------
+BLOCKSCOUT_CHAINS: dict[str, str] = {
+    "ethereum": "https://eth.blockscout.com",
+    "polygon": "https://polygon.blockscout.com",
+    "arbitrum": "https://arbitrum.blockscout.com",
+    "base": "https://base.blockscout.com",
+    "optimism": "https://optimism.blockscout.com",
+}
+
+# ---------------------------------------------------------------------------
+# Routescan (no API key required) — chain name → EVM chain ID
+# ---------------------------------------------------------------------------
+ROUTESCAN_CHAINS: dict[str, int] = {
+    "avalanche": 43114,
+}
+
+# ---------------------------------------------------------------------------
+# Native token metadata per chain
+# ---------------------------------------------------------------------------
+NATIVE_SYMBOL: dict[str, str] = {
+    "ethereum": "ETH",
+    "polygon": "POL",
+    "arbitrum": "ETH",
+    "base": "ETH",
+    "optimism": "ETH",
+    "avalanche": "AVAX",
+}
+
+NATIVE_DECIMALS: int = 18
+
+# ---------------------------------------------------------------------------
+# Classification sets
+# ---------------------------------------------------------------------------
+STABLECOIN_SYMBOLS: set[str] = {
+    "USDC", "USDT", "DAI", "TUSD", "BUSD", "FRAX", "MIM", "USDe",
+}
+
+BLUE_CHIP_SYMBOLS: set[str] = {
+    "ETH", "WETH", "BTC", "WBTC", "BNB", "AVAX", "WAVAX", "MATIC", "POL",
+}
+
+# Minimum USD value for a position to be included in the report.
+MIN_VALUE_USD: float = 0.01

src/agents/portfolio/prompt.py

@@ -0,0 +1,98 @@
+"""
+System prompt for the Portfolio Advisor agent.
+
+Combines detailed per-token analysis, risk assessment, market context,
+and proactive Swap upselling.
+"""
+
+from src.agents.markdown_instructions import MARKDOWN_INSTRUCTIONS
+
+PORTFOLIO_ADVISOR_SYSTEM_PROMPT = f"""You are Zico's **Senior Portfolio & Risk Analyst**.
+Your mission is to help users understand their on-chain portfolio in full detail — every single token, its value, its risk, and what action to take.
+
+Always respond in English, regardless of the user's language.
+
+## Workflow
+
+1. **Fetch the portfolio** — Always call `get_user_portfolio` first to get real, live data. Never guess or fabricate holdings.
+2. **Present EVERY token** — List ALL tokens from the tool result in a detailed table. Never omit or summarize tokens.
+3. **Analyze each position** — For every token, assess the risk level based on its category and market position.
+4. **Research** — For the top 3 holdings or any position that raises concern, use the search tool to look up recent news, hacks, governance issues, or market sentiment.
+5. **Advise** — Provide concrete recommendations with specific swap actions.
+
+## CRITICAL: Always Show the Complete Token Table
+
+When presenting the portfolio, you MUST include a full table with ALL tokens. Use this exact format:
+
+| Token | Chain | Balance | Value (USD) | % of Portfolio | Risk Level |
+|-------|-------|---------|-------------|----------------|------------|
+| ETH | ethereum | 0.0026 | $5.04 | 58.1% | 🟢 Low |
+| STETH | ethereum | 0.001 | $2.03 | 23.4% | 🟢 Low |
+| ... | ... | ... | ... | ... | ... |
+
+List EVERY token returned by the tool. Do NOT skip any token, even those with tiny values. The user wants to see their complete inventory.
+
+## Risk Level Classification
+
+Assign risk levels to each token based on these rules:
+
+- **🟢 Low Risk**: BTC, ETH, WETH, WBTC — established, battle-tested protocols with massive liquidity. Users can comfortably hold large positions.
+- **🟡 Medium Risk**: stETH, MATIC/POL, BNB, AVAX, WAVAX — top-tier L1/L2 tokens or major liquid staking derivatives. Solid but more volatile than BTC/ETH.
+- **🟠 High Risk**: Smaller-cap altcoins, meme tokens, newer DeFi tokens — these can lose 50-90% of value quickly. Position sizes should be small.
+- **🔴 Critical Risk**: Unknown tokens, tokens with very low liquidity, tokens that appeared via airdrop without the user buying them — may be scam tokens.
+- **⚪ Stable**: USDC, USDT, DAI — stablecoins. Low risk but capital is not growing.
+
+## Market Perspective (Use This for Analysis)
+
+When advising, provide market-informed perspective:
+- **BTC/ETH** are considered the "blue chips" of crypto — higher exposure is generally acceptable for long-term holders
+- **Layer 2 tokens** (ARB, OP, BASE) are higher risk than ETH but still quality projects
+- **Stablecoins** are safe but idle capital — suggest yield opportunities (lending, staking) when appropriate
+- **Altcoins/Meme coins** — warn about concentration risk, suggest keeping these under 10-20% of total portfolio
+- Always consider: is the user over-concentrated in one asset? Under-diversified? Holding idle capital?
+
+## Risk Identification Rules
+
+- **Over-concentration**: Flag any single non-stablecoin asset that represents ≥ 30% of total portfolio value.
+- **Low diversification**: Flag if the portfolio has < 3 distinct assets.
+- **High altcoin exposure**: Flag if altcoins represent > 50% of total value.
+- **Stablecoin dominance**: Note if stablecoins are > 70% (capital may not be working for the user).
+- **News-driven risk**: If search reveals negative news (hacks, exploits, depegs, regulatory action) for a held asset, escalate the alert.
+
+## Swap Upsell Guidelines
+
+Whenever you identify a risk or improvement opportunity, suggest a **specific Swap action** the user can perform through our platform:
+
+- Be specific: mention exact tokens, approximate amounts, and the reasoning.
+- Frame swaps as protective or opportunistic moves, not sales pitches.
+- Use the pattern: "I can help you **swap X% of [Token] to [Token]** right now to [reason]."
+- Suggest diversifying into ETH, stablecoins (USDC), or other blue chips when reducing risk.
+- If the portfolio looks healthy, acknowledge it and suggest yield opportunities (staking, DCA).
+
+## Response Structure
+
+Always structure your response with these sections:
+
+### 1. Portfolio Overview
+- Total value, total asset count, chains with holdings
+
+### 2. Complete Holdings Table
+- Full table with ALL tokens (see format above) — NEVER skip tokens
+
+### 3. Allocation Breakdown
+- Blue chips %, Stablecoins %, Altcoins %
+- Visual summary of where the money is
+
+### 4. Risk Assessment
+- Per-token risk analysis with severity
+- Overall portfolio health score
+
+### 5. Market Context & News
+- Relevant findings from web search for key holdings
+- Current market sentiment for major positions
+
+### 6. Recommendations
+- Specific swap or rebalancing actions with exact amounts
+- Priority-ordered: most important action first
+
+{MARKDOWN_INSTRUCTIONS}"""

src/agents/portfolio/tools.py

@@ -0,0 +1,412 @@
+"""
+Portfolio tools — fetches multi-chain wallet balances via free, keyless APIs.
+
+Data sources:
+  • Blockscout API v2 — ETH, Polygon, Arbitrum, Base, Optimism
+  • Routescan API v2  — Avalanche
+
+Both APIs return token balances with USD pricing included, so no separate
+price-feed is needed.
+
+Session management follows the same ContextVar pattern used by the swap,
+lending, and staking agents so the wallet_address is injected by the graph
+node rather than asked from the user.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from contextlib import contextmanager
+from contextvars import ContextVar
+from threading import Lock
+from typing import Any, Dict, List, Optional
+
+import requests
+from langchain_core.tools import tool
+from pydantic import BaseModel
+
+from src.agents.portfolio.config import (
+    BLOCKSCOUT_CHAINS,
+    BLUE_CHIP_SYMBOLS,
+    MIN_VALUE_USD,
+    NATIVE_DECIMALS,
+    NATIVE_SYMBOL,
+    ROUTESCAN_CHAINS,
+    STABLECOIN_SYMBOLS,
+)
+
+logger = logging.getLogger(__name__)
+
+_HTTP_TIMEOUT = 15  # seconds per request
+
+# ---------------------------------------------------------------------------
+# Session context (user_id, conversation_id, wallet_address)
+# ---------------------------------------------------------------------------
+
+_CURRENT_SESSION: ContextVar[tuple[str, str, str]] = ContextVar(
+    "_current_portfolio_session",
+    default=("", "", ""),
+)
+
+
+def set_current_portfolio_session(
+    user_id: Optional[str],
+    conversation_id: Optional[str],
+    wallet_address: Optional[str],
+) -> None:
+    _CURRENT_SESSION.set((
+        (user_id or "").strip(),
+        (conversation_id or "").strip(),
+        (wallet_address or "").strip(),
+    ))
+
+
+def clear_current_portfolio_session() -> None:
+    _CURRENT_SESSION.set(("", "", ""))
+
+
+@contextmanager
+def portfolio_session(
+    user_id: Optional[str],
+    conversation_id: Optional[str],
+    wallet_address: Optional[str],
+):
+    """Context manager that guarantees session scoping for portfolio tool calls."""
+    set_current_portfolio_session(user_id, conversation_id, wallet_address)
+    try:
+        yield
+    finally:
+        clear_current_portfolio_session()
+
+
+# ---------------------------------------------------------------------------
+# Simple TTL cache (avoids hammering explorers on follow-up questions)
+# ---------------------------------------------------------------------------
+
+_PORTFOLIO_CACHE: Dict[str, tuple[Any, float]] = {}
+_CACHE_TTL = 60  # seconds
+_cache_lock = Lock()
+
+
+def _get_cached(key: str) -> Optional[Any]:
+    with _cache_lock:
+        entry = _PORTFOLIO_CACHE.get(key)
+        if entry and (time.time() - entry[1]) < _CACHE_TTL:
+            return entry[0]
+        return None
+
+
+def _set_cached(key: str, value: Any) -> None:
+    with _cache_lock:
+        _PORTFOLIO_CACHE[key] = (value, time.time())
+
+
+# ---------------------------------------------------------------------------
+# Blockscout helpers
+# ---------------------------------------------------------------------------
+
+def _blockscout_fetch_address(base_url: str, address: str) -> Dict[str, Any]:
+    """GET /api/v2/addresses/{addr} — native balance + coin exchange rate."""
+    try:
+        resp = requests.get(
+            f"{base_url}/api/v2/addresses/{address}",
+            timeout=_HTTP_TIMEOUT,
+        )
+        resp.raise_for_status()
+        return resp.json()
+    except Exception as exc:
+        logger.warning("Blockscout address fetch failed (%s): %s", base_url, exc)
+        return {}
+
+
+def _blockscout_fetch_tokens(base_url: str, address: str) -> List[Dict[str, Any]]:
+    """GET /api/v2/addresses/{addr}/token-balances — ERC-20 holdings with price."""
+    try:
+        resp = requests.get(
+            f"{base_url}/api/v2/addresses/{address}/token-balances",
+            timeout=_HTTP_TIMEOUT,
+        )
+        resp.raise_for_status()
+        return resp.json()  # list of token objects
+    except Exception as exc:
+        logger.warning("Blockscout token-balances fetch failed (%s): %s", base_url, exc)
+        return []
+
+
+def _process_blockscout_chain(
+    chain_name: str,
+    base_url: str,
+    address: str,
+) -> List[Dict[str, Any]]:
+    """Fetch and normalise all holdings on a single Blockscout-indexed chain."""
+    assets: List[Dict[str, Any]] = []
+
+    # ── Native token ──
+    addr_data = _blockscout_fetch_address(base_url, address)
+    if addr_data:
+        try:
+            raw_balance = int(addr_data.get("coin_balance") or "0")
+            balance = raw_balance / (10 ** NATIVE_DECIMALS)
+            exchange_rate = float(addr_data.get("exchange_rate") or 0)
+            value_usd = balance * exchange_rate
+            symbol = NATIVE_SYMBOL.get(chain_name, "ETH")
+            if value_usd >= MIN_VALUE_USD or balance >= 0.000001:
+                assets.append({
+                    "symbol": symbol,
+                    "name": symbol,
+                    "balance": round(balance, 6),
+                    "value_usd": round(value_usd, 2),
+                    "chain": chain_name,
+                    "contract_address": "native",
+                    "category": _classify(symbol),
+                })
+        except (ValueError, TypeError):
+            pass
+
+    # ── ERC-20 tokens ──
+    tokens = _blockscout_fetch_tokens(base_url, address)
+    for tok in tokens:
+        try:
+            token_info = tok.get("token", {})
+            decimals = int(token_info.get("decimals") or 18)
+            raw_value = int(tok.get("value") or "0")
+            balance = raw_value / (10 ** decimals)
+            if balance <= 0:
+                continue
+
+            symbol = (token_info.get("symbol") or "???").upper()
+            exchange_rate = float(token_info.get("exchange_rate") or 0)
+            value_usd = balance * exchange_rate
+
+            if value_usd < MIN_VALUE_USD and balance < 0.000001:
+                continue
+
+            assets.append({
+                "symbol": symbol,
+                "name": token_info.get("name", symbol),
+                "balance": round(balance, 6),
+                "value_usd": round(value_usd, 2),
+                "chain": chain_name,
+                "contract_address": token_info.get("address_hash", ""),
+                "category": _classify(symbol),
+            })
+        except (ValueError, TypeError):
+            continue
+
+    return assets
+
+
+# ---------------------------------------------------------------------------
+# Routescan helpers
+# ---------------------------------------------------------------------------
+
+_ROUTESCAN_BASE = "https://api.routescan.io/v2/network/mainnet/evm"
+
+
+def _routescan_fetch_tokens(chain_id: int, address: str) -> List[Dict[str, Any]]:
+    """GET /v2/network/mainnet/evm/{chainId}/address/{addr}/erc20-holdings."""
+    try:
+        resp = requests.get(
+            f"{_ROUTESCAN_BASE}/{chain_id}/address/{address}/erc20-holdings",
+            timeout=_HTTP_TIMEOUT,
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        return data.get("items", [])
+    except Exception as exc:
+        logger.warning("Routescan fetch failed (chain %s): %s", chain_id, exc)
+        return []
+
+
+def _routescan_fetch_native(chain_id: int, address: str) -> Dict[str, Any]:
+    """GET /v2/network/mainnet/evm/{chainId}/address/{addr}."""
+    try:
+        resp = requests.get(
+            f"{_ROUTESCAN_BASE}/{chain_id}/address/{address}",
+            timeout=_HTTP_TIMEOUT,
+        )
+        resp.raise_for_status()
+        return resp.json()
+    except Exception as exc:
+        logger.warning("Routescan native fetch failed (chain %s): %s", chain_id, exc)
+        return {}
+
+
+def _process_routescan_chain(
+    chain_name: str,
+    chain_id: int,
+    address: str,
+) -> List[Dict[str, Any]]:
+    """Fetch and normalise all holdings on a single Routescan-indexed chain."""
+    assets: List[Dict[str, Any]] = []
+
+    # ── Native token ──
+    native_data = _routescan_fetch_native(chain_id, address)
+    if native_data:
+        try:
+            raw_balance = int(native_data.get("balance") or "0")
+            balance = raw_balance / (10 ** NATIVE_DECIMALS)
+            # Routescan native endpoint may not include price;
+            # use CoinGecko-style fallback only for AVAX
+            native_price = float(native_data.get("tokenPriceUsd") or 0)
+            value_usd = balance * native_price
+            symbol = NATIVE_SYMBOL.get(chain_name, "AVAX")
+            if value_usd >= MIN_VALUE_USD or balance >= 0.000001:
+                assets.append({
+                    "symbol": symbol,
+                    "name": symbol,
+                    "balance": round(balance, 6),
+                    "value_usd": round(value_usd, 2),
+                    "chain": chain_name,
+                    "contract_address": "native",
+                    "category": _classify(symbol),
+                })
+        except (ValueError, TypeError):
+            pass
+
+    # ── ERC-20 tokens ──
+    tokens = _routescan_fetch_tokens(chain_id, address)
+    for tok in tokens:
+        try:
+            decimals = int(tok.get("tokenDecimals") or 18)
+            raw_qty = int(tok.get("tokenQuantity") or "0")
+            balance = raw_qty / (10 ** decimals)
+            if balance <= 0:
+                continue
+
+            symbol = (tok.get("tokenSymbol") or "???").upper()
+            value_usd = float(tok.get("tokenValueInUsd") or 0)
+
+            if value_usd < MIN_VALUE_USD and balance < 0.000001:
+                continue
+
+            assets.append({
+                "symbol": symbol,
+                "name": tok.get("tokenName", symbol),
+                "balance": round(balance, 6),
+                "value_usd": round(value_usd, 2),
+                "chain": chain_name,
+                "contract_address": tok.get("tokenAddress", ""),
+                "category": _classify(symbol),
+            })
+        except (ValueError, TypeError):
+            continue
+
+    return assets
+
+
+# ---------------------------------------------------------------------------
+# Classification
+# ---------------------------------------------------------------------------
+
+def _classify(symbol: str) -> str:
+    upper = symbol.upper()
+    if upper in STABLECOIN_SYMBOLS:
+        return "stablecoin"
+    if upper in BLUE_CHIP_SYMBOLS:
+        return "blue_chip"
+    return "altcoin"
+
+
+# ---------------------------------------------------------------------------
+# The Tool
+# ---------------------------------------------------------------------------
+
+class _GetPortfolioArgs(BaseModel):
+    """No user-facing args — wallet_address comes from the session context."""
+    pass
+
+
+@tool("get_user_portfolio", args_schema=_GetPortfolioArgs)
+def get_user_portfolio_tool() -> str:
+    """Fetch the connected user's multi-chain token holdings with USD values.
+
+    Returns a JSON object with total_value_usd, top_holdings, allocation
+    breakdown (stablecoins %, blue_chips %, altcoins %), and the full
+    asset list.  Use this data to analyze portfolio risk and concentration.
+    """
+    _, _, wallet_address = _CURRENT_SESSION.get()
+
+    if not wallet_address:
+        return json.dumps({"error": "No wallet address available. Ask the user to connect their wallet."})
+
+    # Check cache
+    cache_key = f"portfolio:{wallet_address.lower()}"
+    cached = _get_cached(cache_key)
+    if cached:
+        return cached
+
+    # ── Fetch all chains in parallel ──
+    all_assets: List[Dict[str, Any]] = []
+
+    with ThreadPoolExecutor(max_workers=6) as pool:
+        futures = {}
+
+        # Blockscout chains
+        for chain_name, base_url in BLOCKSCOUT_CHAINS.items():
+            fut = pool.submit(_process_blockscout_chain, chain_name, base_url, wallet_address)
+            futures[fut] = chain_name
+
+        # Routescan chains
+        for chain_name, chain_id in ROUTESCAN_CHAINS.items():
+            fut = pool.submit(_process_routescan_chain, chain_name, chain_id, wallet_address)
+            futures[fut] = chain_name
+
+        for fut in as_completed(futures):
+            chain = futures[fut]
+            try:
+                all_assets.extend(fut.result())
+            except Exception as exc:
+                logger.warning("Failed to fetch %s: %s", chain, exc)
+
+    if not all_assets:
+        return json.dumps({
+            "wallet_address": wallet_address,
+            "total_value_usd": 0,
+            "asset_count": 0,
+            "chains_checked": list(BLOCKSCOUT_CHAINS.keys()) + list(ROUTESCAN_CHAINS.keys()),
+            "top_holdings": [],
+            "allocation": {"stablecoins_pct": 0, "blue_chips_pct": 0, "altcoins_pct": 0},
+            "all_assets": [],
+            "note": "No token balances found. The wallet may be empty or the APIs may be temporarily unavailable.",
+        })
+
+    # ── Aggregate ──
+    total_value = sum(a["value_usd"] for a in all_assets)
+    for a in all_assets:
+        a["percentage"] = round((a["value_usd"] / total_value * 100) if total_value > 0 else 0, 2)
+
+    all_assets.sort(key=lambda a: a["value_usd"], reverse=True)
+
+    stablecoins_usd = sum(a["value_usd"] for a in all_assets if a["category"] == "stablecoin")
+    blue_chips_usd = sum(a["value_usd"] for a in all_assets if a["category"] == "blue_chip")
+    altcoins_usd = sum(a["value_usd"] for a in all_assets if a["category"] == "altcoin")
+
+    result = json.dumps({
+        "wallet_address": wallet_address,
+        "total_value_usd": round(total_value, 2),
+        "asset_count": len(all_assets),
+        "chains_checked": list(BLOCKSCOUT_CHAINS.keys()) + list(ROUTESCAN_CHAINS.keys()),
+        "top_holdings": all_assets[:5],
+        "allocation": {
+            "stablecoins_pct": round((stablecoins_usd / total_value * 100) if total_value > 0 else 0, 1),
+            "blue_chips_pct": round((blue_chips_usd / total_value * 100) if total_value > 0 else 0, 1),
+            "altcoins_pct": round((altcoins_usd / total_value * 100) if total_value > 0 else 0, 1),
+        },
+        "all_assets": all_assets,
+    })
+
+    _set_cached(cache_key, result)
+    return result
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def get_tools() -> list:
+    """Return the toolset for the portfolio advisor agent."""
+    return [get_user_portfolio_tool]

src/agents/routing/semantic_router.py

@@ -27,6 +27,7 @@ class IntentCategory(str, Enum):
     STAKING = "staking"
     DCA = "dca"
     MARKET_DATA = "market_data"
+    PORTFOLIO = "portfolio"
     EDUCATION = "education"
     SEARCH = "search"
     GENERAL = "general"
@@ -39,6 +40,7 @@ _INTENT_AGENT_MAP: Dict[IntentCategory, str] = {
     IntentCategory.STAKING: "staking_agent",
     IntentCategory.DCA: "dca_agent",
     IntentCategory.MARKET_DATA: "crypto_agent",
+    IntentCategory.PORTFOLIO: "portfolio_advisor",
     IntentCategory.EDUCATION: "default_agent",
     IntentCategory.SEARCH: "search_agent",
     IntentCategory.GENERAL: "default_agent",
@@ -114,6 +116,32 @@ INTENT_EXEMPLARS: Dict[IntentCategory, List[str]] = {
         "Bitcoin market cap",
         "What is the TVL of Avalanche?",
     ],
+    IntentCategory.PORTFOLIO: [
+        "Analyze my portfolio",
+        "What's my biggest risk?",
+        "How should I rebalance my wallet?",
+        "Show me my token holdings",
+        "What tokens do I have?",
+        "Am I too exposed to any single token?",
+        "Give me a risk assessment of my wallet",
+        "How diversified is my portfolio?",
+        "What should I do with my tokens?",
+        "What is the amount of each token?",
+        "How much of each token do I hold?",
+        "Show me the value of each position",
+        "List all my tokens with balances",
+        "Where am I most exposed?",
+        "Which tokens should I sell?",
+        "Where should I take more risk?",
+        "Where should I be more careful?",
+        "Is my wallet safe?",
+        "Analise minha carteira",
+        "Qual meu maior risco?",
+        "Como balancear minha carteira?",
+        "Quais tokens eu tenho?",
+        "Quanto tenho de cada token?",
+        "Onde posso arriscar mais?",
+    ],
     IntentCategory.EDUCATION: [
         "What is DeFi?",
         "How does liquid staking work?",

src/app.py

@@ -2,10 +2,12 @@ import asyncio
 import base64
 import json
 import os
+import time
 from typing import Any, Dict, List, Optional
 
 from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import StreamingResponse
 from langchain_core.messages import HumanMessage
 from pydantic import BaseModel
 
@@ -135,6 +137,7 @@ def _map_agent_type(agent_name: str) -> str:
         "swap_agent": "token swap",
         "lending_agent": "lending",
         "staking_agent": "staking",
+        "portfolio_advisor": "portfolio analysis",
         "supervisor": "supervisor",
     }
     return mapping.get(agent_name, "supervisor")
@@ -177,6 +180,7 @@ def _invoke_graph(
     user_id,
     conversation_id,
     *,
+    wallet_address: str | None = None,
     pre_classified: Dict[str, Any] | None = None,
 ):
     """Invoke the StateGraph and return the result state.
@@ -189,6 +193,7 @@ def _invoke_graph(
         "messages": conversation_messages,
         "user_id": user_id,
         "conversation_id": conversation_id,
+        "wallet_address": wallet_address,
     }
     if pre_classified:
         initial_state.update(pre_classified)
@@ -408,7 +413,7 @@ def chat(request: ChatRequest):
         cost_snapshot = cost_tracker.get_snapshot()
 
         # Invoke the StateGraph
-        result = _invoke_graph(conversation_messages, user_id, conversation_id)
+        result = _invoke_graph(conversation_messages, user_id, conversation_id, wallet_address=wallet)
 
         # Calculate and save cost delta
         cost_delta = cost_tracker.calculate_delta(cost_snapshot)
@@ -441,6 +446,318 @@ def chat(request: ChatRequest):
         raise HTTPException(status_code=500, detail=str(e))
 
 
+# ---------------------------------------------------------------------------
+# SSE Streaming endpoint
+# ---------------------------------------------------------------------------
+
+# Human-readable labels for each graph node
+_NODE_LABELS: Dict[str, str] = {
+    "entry_node": "Preparing context...",
+    "semantic_router_node": "Routing your request...",
+    "llm_router_node": "Analyzing intent...",
+    "swap_agent_node": "Consulting swap protocols...",
+    "lending_agent_node": "Checking lending markets...",
+    "staking_agent_node": "Reviewing staking options...",
+    "dca_agent_node": "Planning DCA strategy...",
+    "crypto_agent_node": "Fetching market data...",
+    "search_agent_node": "Searching the web...",
+    "default_agent_node": "Thinking...",
+    "database_agent_node": "Querying portfolio...",
+    "portfolio_advisor_node": "Analyzing your portfolio...",
+    "formatter_node": "Formatting response...",
+    "error_node": "Validating parameters...",
+}
+
+
+def _sse(event_type: str, data: dict) -> str:
+    """Format a Server-Sent Event string."""
+    return f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
+
+
+async def _persist_response_bg(
+    full_response: str,
+    response_agent: str,
+    response_metadata: dict,
+    user_id: str,
+    conversation_id: str,
+    cost_delta: dict,
+) -> None:
+    """Background task: persist assistant message and update costs."""
+    try:
+        agent_name = _map_agent_type(response_agent)
+        response_message = ChatMessage(
+            role="assistant",
+            content=full_response,
+            agent_name=agent_name,
+            agent_type=_map_agent_type(agent_name),
+            metadata=response_metadata,
+            conversation_id=conversation_id,
+            user_id=user_id,
+            requires_action=(
+                True if agent_name in ("token swap", "lending", "staking") else False
+            ),
+            action_type=(
+                "swap"
+                if agent_name == "token swap"
+                else "lending"
+                if agent_name == "lending"
+                else "staking"
+                if agent_name == "staking"
+                else None
+            ),
+        )
+        await asyncio.to_thread(
+            chat_manager_instance.add_message,
+            response_message.dict(),
+            conversation_id,
+            user_id,
+        )
+
+        if cost_delta.get("cost", 0) > 0 or cost_delta.get("calls", 0) > 0:
+            await asyncio.to_thread(
+                chat_manager_instance.update_conversation_costs,
+                cost_delta,
+                conversation_id,
+                user_id,
+            )
+
+        # Clear DeFi metadata when intent is ready
+        _clear_ready_metadata(agent_name, response_metadata, user_id, conversation_id)
+    except Exception:
+        logger.exception("Failed to persist streamed response")
+
+
+@app.post("/chat/stream")
+async def chat_stream(request: ChatRequest):
+    """SSE streaming endpoint — streams thought process + tokens in real-time.
+
+    Event types:
+      - ``status``  : node lifecycle (step label, routing info)
+      - ``token``   : incremental text chunks from the final LLM response
+      - ``tool_io`` : tool invocation results (truncated for the wire)
+      - ``done``    : final metadata envelope (agent, nodes, costs)
+      - ``error``   : unrecoverable error
+    """
+    uid: str | None = None
+    cid: str | None = None
+    try:
+        uid, cid = _resolve_identity(request)
+    except HTTPException as exc:
+        # Return error as a streaming event so the client can parse it
+        async def _err():
+            yield _sse("error", {"message": exc.detail})
+
+        return StreamingResponse(_err(), media_type="text/event-stream")
+
+    user_id, conversation_id = uid, cid
+
+    wallet = request.wallet_address.strip() if request.wallet_address else None
+    if wallet and wallet.lower() == "default":
+        wallet = None
+    display_name = None
+    if isinstance(request.message.metadata, dict):
+        display_name = request.message.metadata.get("display_name")
+
+    # Session setup, message persistence, and history fetch (non-blocking)
+    await asyncio.to_thread(
+        chat_manager_instance.ensure_session,
+        user_id,
+        conversation_id,
+        wallet_address=wallet,
+        display_name=display_name,
+    )
+
+    if request.message.role == "user":
+        clean_content = _sanitize_user_message_content(request.message.content)
+        if clean_content is not None:
+            request.message.content = clean_content
+
+    await asyncio.to_thread(
+        chat_manager_instance.add_message,
+        request.message.dict(),
+        conversation_id,
+        user_id,
+    )
+
+    conversation_messages = await asyncio.to_thread(
+        chat_manager_instance.get_messages,
+        conversation_id,
+        user_id,
+    )
+
+    initial_state: Dict[str, Any] = {
+        "messages": conversation_messages,
+        "user_id": user_id,
+        "conversation_id": conversation_id,
+        "wallet_address": wallet,
+    }
+
+    async def event_generator():
+        """Yields SSE events from LangGraph astream_events."""
+        cost_tracker = Config.get_cost_tracker()
+        cost_snapshot = cost_tracker.get_snapshot()
+
+        final_response_chunks: List[str] = []
+        response_agent = "supervisor"
+        response_metadata: Dict[str, Any] = {}
+        nodes_executed: List[str] = []
+        # Track which node is the "final agent" so we only stream its tokens
+        current_agent_node: str | None = None
+        streaming_tokens = False
+
+        try:
+            async for event in graph.astream_events(
+                initial_state, version="v2"
+            ):
+                kind = event["event"]
+                name = event.get("name", "")
+
+                # ── Node starts ──
+                if kind == "on_chain_start" and name in _NODE_LABELS:
+                    nodes_executed.append(name)
+                    # Track agent node for token attribution
+                    if name.endswith("_agent_node"):
+                        current_agent_node = name
+                    yield _sse("status", {
+                        "step": name,
+                        "label": _NODE_LABELS[name],
+                        "ts": time.time(),
+                    })
+
+                # ── Semantic router result ──
+                elif kind == "on_chain_end" and name == "semantic_router_node":
+                    output = event.get("data", {}).get("output", {})
+                    if isinstance(output, dict):
+                        yield _sse("status", {
+                            "step": "routed",
+                            "agent": output.get("route_agent", "unknown"),
+                            "confidence": output.get("route_confidence", 0),
+                            "ts": time.time(),
+                        })
+
+                # ── Tool invocations ──
+                elif kind == "on_tool_start":
+                    yield _sse("status", {
+                        "step": "tool",
+                        "tool": name,
+                        "label": f"Using {name}...",
+                        "ts": time.time(),
+                    })
+
+                elif kind == "on_tool_end":
+                    tool_output = event.get("data", {}).get("output", "")
+                    preview = str(tool_output)[:200]
+                    yield _sse("tool_io", {
+                        "tool": name,
+                        "output": preview,
+                        "ts": time.time(),
+                    })
+
+                # ── LLM token streaming ──
+                elif kind == "on_chat_model_stream":
+                    chunk = event.get("data", {}).get("chunk")
+                    if chunk and hasattr(chunk, "content") and chunk.content:
+                        text = chunk.content if isinstance(chunk.content, str) else ""
+                        if text:
+                            # Only stream tokens from agent nodes, not from
+                            # router/formatter internal calls
+                            parent_tags = event.get("tags", [])
+                            is_formatter = "formatter" in name.lower() or any(
+                                "formatter" in t for t in parent_tags
+                            )
+                            if not is_formatter and current_agent_node:
+                                if not streaming_tokens:
+                                    streaming_tokens = True
+                                    yield _sse("status", {
+                                        "step": "generating",
+                                        "label": "Generating response...",
+                                        "ts": time.time(),
+                                    })
+                                final_response_chunks.append(text)
+                                yield _sse("token", {"t": text})
+
+                # ── Node ends — capture graph output ──
+                elif kind == "on_chain_end" and name == "LangGraph":
+                    output = event.get("data", {}).get("output", {})
+                    if isinstance(output, dict):
+                        response_agent = output.get(
+                            "response_agent", response_agent
+                        )
+                        response_metadata = output.get(
+                            "response_metadata", response_metadata
+                        )
+                        # If we didn't stream tokens (e.g. formatter rewrote),
+                        # use the final_response from the graph
+                        if not final_response_chunks:
+                            graph_response = output.get("final_response", "")
+                            if graph_response:
+                                final_response_chunks.append(graph_response)
+
+        except Exception as exc:
+            logger.exception("Stream error for user=%s conversation=%s", user_id, conversation_id)
+            yield _sse("error", {"message": str(exc)})
+            return
+
+        # ── Build final metadata ──
+        full_response = "".join(final_response_chunks)
+        cost_delta = cost_tracker.calculate_delta(cost_snapshot)
+
+        agent_name = _map_agent_type(response_agent)
+
+        # Enrich metadata (same logic as _build_response_payload)
+        if not response_metadata:
+            if agent_name == "token swap":
+                swap_meta = metadata.get_swap_agent(
+                    user_id=user_id, conversation_id=conversation_id
+                )
+                if swap_meta:
+                    response_metadata = swap_meta
+            elif agent_name == "lending":
+                lending_meta = metadata.get_lending_agent(
+                    user_id=user_id, conversation_id=conversation_id
+                )
+                if lending_meta:
+                    response_metadata = lending_meta
+            elif agent_name == "staking":
+                staking_meta = metadata.get_staking_agent(
+                    user_id=user_id, conversation_id=conversation_id
+                )
+                if staking_meta:
+                    response_metadata = staking_meta
+
+        yield _sse("done", {
+            "agent": agent_name,
+            "nodes": nodes_executed,
+            "metadata": response_metadata,
+            "response": full_response,
+            "costs": {
+                "total_usd": cost_delta.get("cost", 0),
+            },
+        })
+
+        # ── Background: persist response + costs ──
+        asyncio.create_task(
+            _persist_response_bg(
+                full_response,
+                response_agent,
+                response_metadata,
+                user_id,
+                conversation_id,
+                cost_delta,
+            )
+        )
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
 # Supported audio MIME types
 AUDIO_MIME_TYPES = {
     ".mp3": "audio/mpeg",
@@ -481,6 +798,11 @@ Return ONLY a JSON object (no markdown fences) with these fields:
 {"transcription": "<exact transcription>", "intent": "<category>", "confidence": <0.0-1.0>}
 """
 
+_AUDIO_TRANSCRIBE_ONLY_PROMPT = """\
+You will receive an audio clip. Transcribe exactly what is being said.
+Return ONLY the transcription text, nothing else. No JSON, no markdown, no labels.
+"""
+
 # Maps audio classification intents to agent runtime names
 _AUDIO_INTENT_AGENT_MAP: Dict[str, str] = {
     "swap": "swap_agent",
@@ -488,6 +810,7 @@ _AUDIO_INTENT_AGENT_MAP: Dict[str, str] = {
     "staking": "staking_agent",
     "dca": "dca_agent",
     "market_data": "crypto_agent",
+    "portfolio": "portfolio_advisor",
     "search": "search_agent",
     "education": "default_agent",
     "general": "default_agent",
@@ -697,6 +1020,7 @@ async def chat_audio(
             conversation_messages,
             request_user_id,
             request_conversation_id,
+            wallet_address=wallet,
             pre_classified=pre_classified,
         )
 
@@ -742,5 +1066,65 @@ async def chat_audio(
         raise HTTPException(status_code=500, detail=str(e))
 
 
+@app.post("/transcribe")
+async def transcribe_audio(
+    audio: UploadFile = File(..., description="Audio file (mp3, wav, flac, ogg, webm, m4a)"),
+):
+    """Transcribe audio to text without invoking the agent pipeline.
+
+    Stateless endpoint — no user_id, conversation_id, or persistence.
+    Returns ``{"text": "<transcription>"}``.
+    """
+    try:
+        audio_content = await audio.read()
+        if len(audio_content) > MAX_AUDIO_SIZE:
+            raise HTTPException(
+                status_code=413,
+                detail=f"Audio file too large. Maximum size is {MAX_AUDIO_SIZE // (1024 * 1024)}MB.",
+            )
+        if len(audio_content) == 0:
+            raise HTTPException(status_code=400, detail="Audio file is empty.")
+
+        mime_type = _get_audio_mime_type(audio.filename or "", audio.content_type)
+        encoded_audio = base64.b64encode(audio_content).decode("utf-8")
+
+        from src.llm.tiers import ModelTier
+        llm = Config.get_llm(model=ModelTier.TRANSCRIPTION, with_cost_tracking=True)
+
+        message = HumanMessage(
+            content=[
+                {"type": "text", "text": _AUDIO_TRANSCRIBE_ONLY_PROMPT},
+                {"type": "media", "data": encoded_audio, "mime_type": mime_type},
+            ]
+        )
+
+        response = await asyncio.to_thread(llm.invoke, [message])
+
+        raw_content = response.content
+        if isinstance(raw_content, list):
+            text_parts = []
+            for part in raw_content:
+                if isinstance(part, dict) and part.get("text"):
+                    text_parts.append(part["text"])
+                elif isinstance(part, str):
+                    text_parts.append(part)
+            raw_content = " ".join(text_parts).strip()
+
+        text = (raw_content or "").strip()
+        if not text:
+            raise HTTPException(
+                status_code=400,
+                detail="Could not transcribe the audio. Please try again with a clearer recording.",
+            )
+
+        return {"text": text}
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.exception("Transcribe endpoint failed")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
 # Include chat manager router
 app.include_router(chat_manager_router)

src/graphs/edges.py

@@ -27,6 +27,7 @@ _INTENT_TO_NODE = {
     IntentCategory.STAKING.value: "staking_agent_node",
     IntentCategory.DCA.value: "dca_agent_node",
     IntentCategory.MARKET_DATA.value: "crypto_agent_node",
+    IntentCategory.PORTFOLIO.value: "portfolio_advisor_node",
     IntentCategory.SEARCH.value: "search_agent_node",
     IntentCategory.EDUCATION.value: "default_agent_node",
     IntentCategory.GENERAL.value: "default_agent_node",
@@ -41,6 +42,7 @@ _AGENT_NAME_TO_NODE = {
     "search_agent": "search_agent_node",
     "default_agent": "default_agent_node",
     "database_agent": "database_agent_node",
+    "portfolio_advisor": "portfolio_advisor_node",
 }
 
 # DeFi state key → node mapping

src/graphs/factory.py

@@ -24,6 +24,7 @@ from src.graphs.nodes import (
     search_agent_node,
     default_agent_node,
     database_agent_node,
+    portfolio_advisor_node,
 )
 from src.graphs.edges import decide_route, after_llm_router
 from src.agents.formatter.node import formatter_node
@@ -40,6 +41,7 @@ _AGENT_NODES = [
     "search_agent_node",
     "default_agent_node",
     "database_agent_node",
+    "portfolio_advisor_node",
 ]
 
 
@@ -70,6 +72,7 @@ def build_graph() -> StateGraph:
     graph.add_node("search_agent_node", search_agent_node)
     graph.add_node("default_agent_node", default_agent_node)
     graph.add_node("database_agent_node", database_agent_node)
+    graph.add_node("portfolio_advisor_node", portfolio_advisor_node)
 
     # --- Entry point ---
     graph.set_entry_point("entry_node")
@@ -92,6 +95,7 @@ def build_graph() -> StateGraph:
             "search_agent_node": "search_agent_node",
             "default_agent_node": "default_agent_node",
             "database_agent_node": "database_agent_node",
+            "portfolio_advisor_node": "portfolio_advisor_node",
         },
     )
 
@@ -108,6 +112,7 @@ def build_graph() -> StateGraph:
             "search_agent_node": "search_agent_node",
             "default_agent_node": "default_agent_node",
             "database_agent_node": "database_agent_node",
+            "portfolio_advisor_node": "portfolio_advisor_node",
         },
     )
 

src/graphs/nodes.py

@@ -48,6 +48,9 @@ from src.agents.staking.agent import StakingAgent
 from src.agents.staking.tools import staking_session
 from src.agents.staking.prompt import STAKING_AGENT_SYSTEM_PROMPT
 from src.agents.search.agent import SearchAgent
+from src.agents.portfolio.agent import PortfolioAdvisorAgent
+from src.agents.portfolio.tools import portfolio_session
+from src.agents.portfolio.prompt import PORTFOLIO_ADVISOR_SYSTEM_PROMPT
 from src.agents.database.client import is_database_available
 
 logger = logging.getLogger(__name__)
@@ -90,6 +93,8 @@ def initialize_agents() -> None:
     _agents["lending_agent"] = LendingAgent(llm).agent
     _agents["staking_agent"] = StakingAgent(llm).agent
 
+    _agents["portfolio_advisor"] = PortfolioAdvisorAgent(llm).agent
+
     if is_database_available():
         _agents["database_agent"] = DatabaseAgent(llm)
     else:
@@ -261,6 +266,7 @@ Available agents:
 - dca_agent: Dollar-cost averaging strategies.
 - lending_agent: Lending operations (supply, borrow, repay, withdraw).
 - staking_agent: Staking operations (stake ETH, unstake stETH via Lido).
+- portfolio_advisor: Portfolio analysis, risk assessment, wallet holdings, rebalancing advice.
 - search_agent: Web search for current events and factual lookups.
 - database_agent: Database queries and data analysis.
 - default_agent: General conversation, education, greetings.
@@ -287,7 +293,8 @@ def llm_router_node(state: AgentState) -> dict:
         # Validate
         valid_agents = {
             "crypto_agent", "swap_agent", "dca_agent", "lending_agent",
-            "staking_agent", "search_agent", "database_agent", "default_agent",
+            "staking_agent", "portfolio_advisor", "search_agent",
+            "database_agent", "default_agent",
         }
         if chosen not in valid_agents:
             chosen = "default_agent"
@@ -464,5 +471,53 @@ def default_agent_node(state: AgentState, config: RunnableConfig | None = None)
     return _invoke_simple_agent("default_agent", state, config)
 
 
+def portfolio_advisor_node(state: AgentState, config: RunnableConfig | None = None) -> dict:
+    """Invoke the portfolio advisor with wallet_address session context."""
+    user_id = state.get("user_id")
+    conversation_id = state.get("conversation_id")
+    wallet_address = state.get("wallet_address")
+    langchain_messages = list(state.get("langchain_messages", []))
+    nodes = list(state.get("nodes_executed", []))
+    nodes.append("portfolio_advisor_node")
+
+    agent = _agents.get("portfolio_advisor")
+    if not agent:
+        return {
+            "final_response": "Portfolio advisor is not available.",
+            "response_agent": "portfolio_advisor",
+            "response_metadata": {},
+            "raw_agent_messages": [],
+            "nodes_executed": nodes,
+        }
+
+    # Inject system prompt
+    scoped_messages = [SystemMessage(content=PORTFOLIO_ADVISOR_SYSTEM_PROMPT)]
+    scoped_messages.extend(langchain_messages)
+
+    try:
+        with portfolio_session(user_id=user_id, conversation_id=conversation_id, wallet_address=wallet_address):
+            response = agent.invoke({"messages": scoped_messages}, config=config)
+    except Exception:
+        logger.exception("Error invoking portfolio_advisor")
+        return {
+            "final_response": "Sorry, an error occurred while analyzing your portfolio.",
+            "response_agent": "portfolio_advisor",
+            "response_metadata": {},
+            "raw_agent_messages": [],
+            "nodes_executed": nodes,
+        }
+
+    agent_name, text, messages_out = extract_response_from_graph(response)
+    meta = build_metadata(agent_name or "portfolio_advisor", user_id, conversation_id, messages_out)
+
+    return {
+        "final_response": text,
+        "response_agent": agent_name or "portfolio_advisor",
+        "response_metadata": meta,
+        "raw_agent_messages": messages_out,
+        "nodes_executed": nodes,
+    }
+
+
 def database_agent_node(state: AgentState, config: RunnableConfig | None = None) -> dict:
     return _invoke_simple_agent("database_agent", state, config)

src/graphs/state.py

@@ -14,6 +14,7 @@ class AgentState(TypedDict, total=False):
     messages: List[Dict[str, Any]]          # Raw conversation messages from the gateway
     user_id: str
     conversation_id: str
+    wallet_address: Optional[str]           # EVM wallet address (from HTTP request)
 
     # --- Windowed context ---
     windowed_messages: List[Dict[str, Any]] # After context windowing