Spaces:
Sleeping
Sleeping
File size: 6,216 Bytes
e5e35a3 cc8beab e5e35a3 6710fbe e5e35a3 6710fbe e5e35a3 cc8beab e5e35a3 cc8beab e5e35a3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | import logging
from typing import Any, Dict, List, Tuple
from src.core.AgentCommand import AgentCommand
from src.core.FinanceState import FinanceState
from src.core.errors import add_error
from src.rag.StockMarketRag import StockMarketRag
logger = logging.getLogger(__name__)
class PortfolioAgent(AgentCommand):
def __init__(self, state: FinanceState):
self.state = state
self.yf = StockMarketRag()
def process(self):
trace_id = str(self.state.get("trace_id") or "")
logger.info("[trace=%s] PortfolioAgent.start", trace_id)
holdings = self.state.get("portfolio") or []
if not holdings:
self.state["response"] = (
"Educational only, not financial advice.\n\n"
"I can analyze a portfolio if you provide holdings with quantities.\n"
"Example: 'My portfolio is 10 AAPL, 5 MSFT, 2 VTI'."
)
return self.state
normalized: List[Tuple[str, float]] = []
for h in holdings:
try:
symbol = str(h.get("symbol") or "").strip().upper()
qty_raw = h.get("quantity")
qty = float(qty_raw) if qty_raw is not None else 0.0
except Exception:
continue
if symbol and qty > 0:
normalized.append((symbol, qty))
if not normalized:
self.state["response"] = (
"Educational only, not financial advice.\n\n"
"I couldn't parse any valid holdings. Please provide symbols and quantities."
)
return self.state
logger.info("[trace=%s] PortfolioAgent.holdings=%d", trace_id, len(normalized))
rows: List[Dict[str, Any]] = []
errors: List[str] = []
total_value = 0.0
for symbol, qty in normalized:
details = self.yf.get_stock_details(symbol)
if details.get("error"):
add_error(
self.state,
code="portfolio_quote_error",
message=str(details.get("error")),
agent="portfolio_agent",
detail={"symbol": symbol, "provider": details.get("provider")},
)
errors.append(f"{symbol}: {details.get('error')}")
continue
price = details.get("currentPrice")
try:
price_f = float(price) if price is not None else None
except Exception:
price_f = None
if price_f is None:
add_error(
self.state,
code="portfolio_quote_error",
message="Missing current price",
agent="portfolio_agent",
detail={"symbol": symbol, "provider": details.get("provider")},
)
errors.append(f"{symbol}: missing current price")
continue
value = price_f * qty
total_value += value
rows.append(
{
"symbol": symbol,
"quantity": qty,
"price": price_f,
"value": value,
"sector": details.get("sector"),
"industry": details.get("industry"),
}
)
if not rows:
self.state["response"] = (
"Educational only, not financial advice.\n\n"
"I couldn't retrieve prices for your holdings. "
+ (f"Errors: {', '.join(errors[:5])}" if errors else "")
)
return self.state
# Compute weights and concentration.
rows.sort(key=lambda r: r["value"], reverse=True)
for r in rows:
r["weight"] = (r["value"] / total_value) if total_value > 0 else 0.0
top1 = float(rows[0]["weight"])
top3 = float(sum(r["weight"] for r in rows[:3]))
n = len(rows)
concentration_note = "Moderate concentration."
if top1 >= 0.5:
concentration_note = "High concentration in a single holding."
elif top3 >= 0.8:
concentration_note = "High concentration across the top 3 holdings."
# Sector breakdown (best-effort).
sector_totals: Dict[str, float] = {}
for r in rows:
sector = r.get("sector") if isinstance(r.get("sector"), str) else None
key = (sector or "Unknown").strip() or "Unknown"
sector_totals[key] = sector_totals.get(key, 0.0) + float(r["weight"])
top_sectors = sorted(sector_totals.items(), key=lambda kv: kv[1], reverse=True)[
:5
]
lines: List[str] = []
lines.append("Educational only, not financial advice.\n")
lines.append(
f"Portfolio snapshot (approx): ${total_value:,.2f} across {n} holdings."
)
lines.append(
f"Concentration: top holding {top1:.0%}, top 3 holdings {top3:.0%}. {concentration_note}"
)
lines.append("\nHoldings (by value):")
for r in rows:
lines.append(
f"- {r['symbol']}: {r['quantity']:.4g} shares x ${r['price']:.2f} "
f"= ${r['value']:,.2f} ({float(r['weight']):.0%})"
)
if top_sectors:
lines.append("\nSector exposure (best-effort):")
for sector, w in top_sectors:
lines.append(f"- {sector}: {w:.0%}")
lines.append("\nWhat to consider next (general education):")
lines.append(
"- Diversification: consider whether any single holding dominates outcomes."
)
lines.append(
"- Time horizon and risk: align stock-heavy exposure with your ability to tolerate volatility."
)
lines.append(
"- Rebalancing: consider simple rules (e.g., annual review) rather than reacting to short-term moves."
)
if errors:
lines.append("\nData issues:")
for e in errors[:8]:
lines.append(f"- {e}")
self.state["response"] = "\n".join(lines).strip()
return self.state
|