Investment Portfolio Multi-Agent System
Build an intelligent portfolio management system with specialized agents for market analysis, risk assessment, and investment recommendations.
Overview
This example demonstrates how to build a multi-agent financial system that can:
- Analyze market trends and financial indicators
- Assess portfolio risk and performance
- Recommend investment strategies based on goals
- Execute trades and rebalance portfolios
- Monitor regulatory compliance
- Generate financial reports and insights
Perfect for: Investment firms, robo-advisors, wealth management platforms, and personal finance applications.
How It Works
Workflow Steps
- Client Profiling: Manager agent analyzes investor profile and goals
- Market Analysis: Research agents analyze market conditions
- Risk Assessment: Risk management agents evaluate portfolio exposure
- Strategy Formulation: Investment strategists develop recommendations
- Trade Execution: Trading agents execute buy/sell orders
- Portfolio Monitoring: Continuous monitoring and rebalancing
- Reporting: Generate performance reports and insights
Agent Hierarchy
Install AgentRouter
# Install AgentRouter SDK
!pip install agentrouter -q
# Import required libraries
from agentrouter import ManagerAgent, WorkerAgent, tool
from agentrouter.visualization import PipelineInspector
import asyncio
import nest_asyncio
from typing import Dict, Any, List, Optional
from datetime import datetime, timedelta
import json
import random
nest_asyncio.apply()
Complete Code
# ============================================
# INVESTMENT PORTFOLIO MULTI-AGENT SYSTEM
# ============================================
# Configuration
API_KEY = "your-api-key-here" # Replace with your actual API key
# ============================================
# TOOL DEFINITIONS
# ============================================
@tool(
schema={
"type": "function",
"function": {
"name": "analyze_market",
"description": "Performs real-time technical and fundamental market analysis across asset classes",
"parameters": {
"type": "object",
"properties": {
"asset_class": {
"type": "string",
"enum": ["stocks", "bonds", "commodities", "crypto", "forex"],
"description": "Asset class to analyze"
},
"timeframe": {
"type": "string",
"enum": ["1D", "1W", "1M", "3M", "1Y"],
"description": "Analysis timeframe"
},
"indicators": {
"type": "array",
"items": {"type": "string"},
"description": "Technical indicators to use"
}
},
"required": ["asset_class"]
}
}
}
)
async def analyze_market(asset_class: str, timeframe: str = "1M", indicators: List[str] = None) -> Dict[str, Any]:
"""Analyze market conditions"""
# Simulated market data
market_conditions = {
"stocks": {
"trend": "bullish",
"volatility": "moderate",
"sp500": {"price": 4750, "change": "+1.2%", "pe_ratio": 24.5},
"nasdaq": {"price": 15200, "change": "+1.8%", "pe_ratio": 32.1},
"sectors": {
"technology": "outperforming",
"healthcare": "neutral",
"finance": "underperforming"
}
},
"bonds": {
"trend": "bearish",
"volatility": "low",
"yield_10y": 4.25,
"yield_curve": "inverted",
"credit_spreads": "widening"
},
"crypto": {
"trend": "volatile",
"volatility": "high",
"bitcoin": {"price": 45000, "change": "+5.3%", "dominance": "48%"},
"ethereum": {"price": 2800, "change": "+7.1%", "gas": "25 gwei"}
}
}
analysis = market_conditions.get(asset_class, {})
return {
"asset_class": asset_class,
"timeframe": timeframe,
"analysis_date": datetime.now().isoformat(),
"market_data": analysis,
"indicators_used": indicators or ["RSI", "MACD", "Moving Averages"],
"recommendation": "hold" if analysis.get("volatility") == "high" else "buy",
"confidence": random.uniform(0.65, 0.95)
}
@tool(
schema={
"type": "function",
"function": {
"name": "assess_portfolio_risk",
"description": "Calculates VaR, Sharpe ratio, beta, and other critical risk metrics for portfolios",
"parameters": {
"type": "object",
"properties": {
"portfolio_id": {
"type": "string",
"description": "Portfolio identifier"
},
"risk_metrics": {
"type": "array",
"items": {"type": "string"},
"description": "Risk metrics to calculate (VaR, Sharpe, Beta, etc.)"
}
},
"required": ["portfolio_id"]
}
}
}
)
async def assess_portfolio_risk(portfolio_id: str, risk_metrics: List[str] = None) -> Dict[str, Any]:
"""Calculate portfolio risk metrics"""
# Simulated risk assessment
return {
"portfolio_id": portfolio_id,
"assessment_date": datetime.now().isoformat(),
"risk_metrics": {
"value_at_risk": {"1_day": 2.5, "1_week": 5.8, "1_month": 12.3, "unit": "percent"},
"sharpe_ratio": 1.45,
"beta": 1.12,
"standard_deviation": 15.6,
"max_drawdown": -18.2,
"correlation_sp500": 0.85
},
"risk_level": "moderate",
"diversification_score": 7.5,
"concentration_risk": {
"top_holding": "AAPL - 8.5%",
"sector_concentration": "Technology - 35%"
},
"recommendations": [
"Consider reducing technology sector exposure",
"Add international equity diversification",
"Consider adding bonds to reduce volatility"
]
}
@tool(
schema={
"type": "function",
"function": {
"name": "get_portfolio_details",
"description": "Retrieves complete portfolio holdings, asset allocation, and performance metrics",
"parameters": {
"type": "object",
"properties": {
"portfolio_id": {
"type": "string",
"description": "Portfolio identifier"
},
"include_performance": {
"type": "boolean",
"description": "Include performance metrics"
}
},
"required": ["portfolio_id"]
}
}
}
)
async def get_portfolio_details(portfolio_id: str, include_performance: bool = True) -> Dict[str, Any]:
"""Get portfolio details"""
return {
"portfolio_id": portfolio_id,
"account_value": 1250000,
"cash_balance": 50000,
"invested_amount": 1200000,
"holdings": [
{"symbol": "AAPL", "shares": 500, "value": 95000, "weight": 7.6, "gain_loss": "+12.5%"},
{"symbol": "MSFT", "shares": 300, "value": 120000, "weight": 9.6, "gain_loss": "+18.2%"},
{"symbol": "GOOGL", "shares": 400, "value": 110000, "weight": 8.8, "gain_loss": "+15.1%"},
{"symbol": "SPY", "shares": 200, "value": 95000, "weight": 7.6, "gain_loss": "+8.3%"},
{"symbol": "BND", "shares": 1000, "value": 75000, "weight": 6.0, "gain_loss": "-2.1%"},
{"symbol": "GLD", "shares": 300, "value": 55000, "weight": 4.4, "gain_loss": "+5.2%"},
{"symbol": "VTI", "shares": 400, "value": 90000, "weight": 7.2, "gain_loss": "+10.1%"}
],
"performance": {
"day_change": "+1.2%",
"week_change": "+2.8%",
"month_change": "+5.3%",
"year_change": "+15.7%",
"total_return": "+22.3%",
"annualized_return": "+12.5%"
} if include_performance else None,
"asset_allocation": {
"stocks": 75,
"bonds": 15,
"commodities": 5,
"cash": 5
}
}
@tool(
schema={
"type": "function",
"function": {
"name": "recommend_trades",
"description": "Generates AI-powered trade recommendations aligned with investment strategy and risk profile",
"parameters": {
"type": "object",
"properties": {
"portfolio_id": {
"type": "string",
"description": "Portfolio identifier"
},
"strategy": {
"type": "string",
"enum": ["conservative", "moderate", "aggressive", "income", "growth"],
"description": "Investment strategy"
},
"amount": {
"type": "number",
"description": "Amount to invest/trade"
}
},
"required": ["portfolio_id", "strategy"]
}
}
}
)
async def recommend_trades(portfolio_id: str, strategy: str, amount: float = None) -> Dict[str, Any]:
"""Generate trade recommendations"""
strategies = {
"conservative": {
"trades": [
{"action": "buy", "symbol": "BND", "amount": 0.4, "reason": "Increase bond allocation"},
{"action": "buy", "symbol": "VIG", "amount": 0.3, "reason": "Dividend aristocrats"},
{"action": "sell", "symbol": "ARKK", "amount": 1.0, "reason": "Reduce high-risk exposure"}
]
},
"aggressive": {
"trades": [
{"action": "buy", "symbol": "NVDA", "amount": 0.3, "reason": "AI growth opportunity"},
{"action": "buy", "symbol": "TSLA", "amount": 0.2, "reason": "EV market leader"},
{"action": "buy", "symbol": "PLTR", "amount": 0.15, "reason": "Data analytics growth"}
]
},
"moderate": {
"trades": [
{"action": "buy", "symbol": "VTI", "amount": 0.3, "reason": "Broad market exposure"},
{"action": "buy", "symbol": "VXUS", "amount": 0.2, "reason": "International diversification"},
{"action": "rebalance", "symbol": "portfolio", "amount": 0.5, "reason": "Maintain target allocation"}
]
}
}
selected_strategy = strategies.get(strategy, strategies["moderate"])
return {
"portfolio_id": portfolio_id,
"strategy": strategy,
"recommendations": selected_strategy["trades"],
"total_amount": amount or 10000,
"execution_priority": "limit_orders",
"time_horizon": "30_days",
"expected_impact": {
"risk_change": "-5%" if strategy == "conservative" else "+8%",
"expected_return": "6-8%" if strategy == "conservative" else "12-18%"
},
"generated_at": datetime.now().isoformat()
}
@tool(
schema={
"type": "function",
"function": {
"name": "execute_trade",
"description": "Executes buy/sell orders with market, limit, or stop-loss instructions",
"parameters": {
"type": "object",
"properties": {
"portfolio_id": {
"type": "string",
"description": "Portfolio identifier"
},
"action": {
"type": "string",
"enum": ["buy", "sell"],
"description": "Trade action"
},
"symbol": {
"type": "string",
"description": "Stock symbol"
},
"quantity": {
"type": "integer",
"description": "Number of shares"
},
"order_type": {
"type": "string",
"enum": ["market", "limit", "stop_loss"],
"description": "Order type"
},
"limit_price": {
"type": "number",
"description": "Limit price for limit orders"
}
},
"required": ["portfolio_id", "action", "symbol", "quantity", "order_type"]
}
}
}
)
async def execute_trade(portfolio_id: str, action: str, symbol: str, quantity: int,
order_type: str, limit_price: float = None) -> Dict[str, Any]:
"""Execute trade order"""
order_id = f"ORD{datetime.now().strftime('%Y%m%d%H%M%S')}"
# Simulated trade execution
return {
"order_id": order_id,
"portfolio_id": portfolio_id,
"status": "executed" if order_type == "market" else "pending",
"action": action,
"symbol": symbol,
"quantity": quantity,
"order_type": order_type,
"executed_price": 150.25 if order_type == "market" else None,
"limit_price": limit_price,
"commission": 0.99,
"total_value": quantity * 150.25,
"execution_time": datetime.now().isoformat() if order_type == "market" else None,
"settlement_date": (datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d")
}
@tool(
schema={
"type": "function",
"function": {
"name": "check_compliance",
"description": "Validates trades against SEC, FINRA, and internal compliance rules",
"parameters": {
"type": "object",
"properties": {
"portfolio_id": {
"type": "string",
"description": "Portfolio identifier"
},
"trade_details": {
"type": "object",
"description": "Trade details to check"
}
},
"required": ["portfolio_id"]
}
}
}
)
async def check_compliance(portfolio_id: str, trade_details: Dict = None) -> Dict[str, Any]:
"""Check regulatory compliance"""
return {
"portfolio_id": portfolio_id,
"compliance_status": "compliant",
"checks_performed": [
{"rule": "Pattern Day Trading", "status": "pass", "details": "Under 4 trades per week"},
{"rule": "Wash Sale", "status": "pass", "details": "No repurchase within 30 days"},
{"rule": "Position Limits", "status": "pass", "details": "Within 10% per position"},
{"rule": "Margin Requirements", "status": "pass", "details": "Maintenance margin met"},
{"rule": "Tax Lot Accounting", "status": "pass", "details": "FIFO method applied"}
],
"warnings": [],
"restrictions": [],
"last_audit": datetime.now().isoformat()
}
# ============================================
# AGENT SYSTEM SETUP
# ============================================
class InvestmentPortfolioSystem:
"""Complete investment portfolio multi-agent system"""
def __init__(self, api_key: str):
self.api_key = api_key
self.manager = None
self.agents = {}
def setup(self):
"""Setup the multi-agent hierarchy"""
# Create Manager Agent - Portfolio Manager
self.manager = ManagerAgent(
name="portfolio_manager",
api_key=self.api_key,
model="usf-mini",
backstory="""I am a seasoned Portfolio Manager with 25 years of experience
managing institutional and high-net-worth portfolios. I have expertise in
asset allocation, risk management, and investment strategy development.""",
goal="Maximize risk-adjusted returns while meeting client investment objectives",
instruction="""
1. Analyze client investment goals and risk tolerance
2. Develop comprehensive investment strategies
3. Monitor market conditions and portfolio performance
4. Coordinate with research analysts and traders
5. Ensure regulatory compliance and risk management
6. Generate regular performance reports and insights
7. Rebalance portfolios based on market conditions
""",
knowledge_cutoff="15 January 2025",
temperature=0.4, # Lower temperature for financial decisions
max_iterations=30
)
# Register manager tools
self.manager.register_tool(get_portfolio_details)
self.manager.register_tool(recommend_trades)
# Create Level 1 Workers
# Market Research Analyst
market_analyst = self.manager.create_worker(
name="Market_Analyst",
role="Analyzes market trends, economic indicators, and identifies investment opportunities",
backstory="Expert in market analysis with deep understanding of economic indicators",
goal="Provide accurate market insights and investment opportunities",
instruction="Analyze market trends, economic data, and identify opportunities"
)
market_analyst.register_tool(analyze_market)
self.agents["market_analyst"] = market_analyst
# Risk Manager
risk_manager = self.manager.create_worker(
name="Risk_Manager",
role="Manages portfolio risk, monitors compliance, and maintains optimal risk-return profiles",
backstory="Specialist in portfolio risk assessment and management",
goal="Maintain optimal risk-return profile and ensure compliance",
instruction="Monitor portfolio risk metrics and ensure regulatory compliance"
)
risk_manager.register_tool(assess_portfolio_risk)
risk_manager.register_tool(check_compliance)
self.agents["risk_manager"] = risk_manager
# Investment Strategist
investment_strategist = self.manager.create_worker(
name="Investment_Strategist",
role="Develops and implements investment strategies across all asset classes",
backstory="Expert in developing investment strategies across asset classes",
goal="Create optimal investment strategies based on market conditions",
instruction="Develop and implement investment strategies aligned with goals"
)
investment_strategist.register_tool(recommend_trades)
self.agents["investment_strategist"] = investment_strategist
# Create Level 2 Workers (Traders under Investment Strategist)
# Equity Trader
equity_trader = investment_strategist.create_worker(
name="Equity_Trader",
role="Executes equity trades, manages stock positions, and optimizes order execution",
backstory="Specialized in equity markets with 15 years trading experience",
goal="Execute equity trades efficiently with minimal market impact"
)
equity_trader.register_tool(execute_trade)
# Bond Trader
bond_trader = investment_strategist.create_worker(
name="Bond_Trader",
role="Manages fixed income portfolio, trades bonds, and controls duration risk",
backstory="Expert in bond markets and interest rate derivatives",
goal="Manage fixed income portfolio and duration risk"
)
bond_trader.register_tool(execute_trade)
# Crypto Trader
crypto_trader = investment_strategist.create_worker(
name="Crypto_Trader",
role="Trades digital assets, manages crypto allocations, and monitors DeFi protocols",
backstory="Specialist in cryptocurrency markets and DeFi protocols",
goal="Manage digital asset allocation and execute crypto trades"
)
crypto_trader.register_tool(execute_trade)
# Create Level 2 Workers (Analysts under Market Analyst)
# Technical Analyst
technical_analyst = market_analyst.create_worker(
name="Technical_Analyst",
role="Identifies trading opportunities using chart patterns and technical indicators",
backstory="Expert in chart patterns and technical indicators",
goal="Identify trading opportunities using technical analysis"
)
technical_analyst.register_tool(analyze_market)
# Fundamental Analyst
fundamental_analyst = market_analyst.create_worker(
name="Fundamental_Analyst",
role="Evaluates companies through financial statements and valuation metrics",
backstory="Expert in company valuation and financial analysis",
goal="Evaluate investment opportunities based on fundamentals"
)
fundamental_analyst.register_tool(analyze_market)
# Create Level 2 Workers (under Risk Manager)
# Compliance Officer
compliance_officer = risk_manager.create_worker(
name="Compliance_Officer",
role="Ensures regulatory compliance for all trading activities and reporting",
backstory="Expert in financial regulations and compliance",
goal="Ensure all trading activities meet regulatory requirements"
)
compliance_officer.register_tool(check_compliance)
# Tax Advisor
tax_advisor = risk_manager.create_worker(
name="Tax_Advisor",
role="Optimizes portfolio tax efficiency and manages tax-loss harvesting strategies",
backstory="Expert in investment taxation and tax-efficient strategies",
goal="Optimize portfolio for tax efficiency"
)
print("✅ Investment Portfolio System initialized successfully!")
# Visualize hierarchy
inspector = PipelineInspector(self.manager)
print("\n📊 INVESTMENT TEAM HIERARCHY:")
print(inspector.visualize(format='mermaid'))
print("\n🔧 FINANCIAL TOOLS AVAILABLE:")
# Get all tools from inspector
for agent in inspector.pipeline_data['agents']:
if agent['tools']:
print(f" {agent['name']}:")
for tool_name in agent['tools']:
print(f" • {tool_name}")
async def manage_portfolio(self, request: str, portfolio_id: str = "PORT001"):
"""Process portfolio management request"""
messages = [
{
"role": "system",
"content": f"Portfolio ID: {portfolio_id}\nDate: {datetime.now().strftime('%Y-%m-%d %H:%M')}\nMarket Hours: Open"
},
{
"role": "user",
"content": request
}
]
print(f"\n💼 Client ({portfolio_id}): {request}")
print("📈 Processing with investment AI system...")
try:
response = await self.manager.execute(messages)
if response and "choices" in response:
result = response["choices"][0]["message"]["content"]
print(f"\n✅ Investment Recommendation:\n{result}")
return {
"success": True,
"portfolio_id": portfolio_id,
"request": request,
"recommendation": result,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
print(f"❌ Error: {str(e)}")
return {
"success": False,
"portfolio_id": portfolio_id,
"request": request,
"error": str(e),
"timestamp": datetime.now().isoformat()
}
# ============================================
# EXAMPLE USAGE
# ============================================
async def run_investment_example():
"""Run investment portfolio scenarios"""
# Initialize system
portfolio_system = InvestmentPortfolioSystem(API_KEY)
portfolio_system.setup()
# Test scenarios
test_cases = [
{
"portfolio_id": "PORT001",
"request": "My portfolio is worth $1.25M with 75% stocks. Given current market conditions, should I rebalance? I'm 45 years old with moderate risk tolerance."
},
{
"portfolio_id": "PORT002",
"request": "I want to invest $100,000 in growth stocks. What are your recommendations for the technology sector?"
}
]
print("\n" + "="*60)
print("📈 RUNNING INVESTMENT PORTFOLIO SCENARIOS")
print("="*60)
results = []
for test_case in test_cases[:1]: # Run first case for demo
result = await portfolio_system.manage_portfolio(
test_case["request"],
test_case["portfolio_id"]
)
results.append(result)
print("\n" + "-"*60)
return results
# ============================================
# MAIN EXECUTION
# ============================================
# Run the example
results = await run_investment_example()
# Display results
print("\n📋 PORTFOLIO MANAGEMENT RESULTS:")
for result in results:
print(json.dumps(result, indent=2))
Expected Output
✅ Investment Portfolio System initialized successfully!
📊 INVESTMENT TEAM HIERARCHY:
graph TB
Portfolio_Manager["Portfolio Manager"]
Market_Analyst["Market Analyst"]
Risk_Manager["Risk Manager"]
Investment_Strategist["Investment Strategist"]
Equity_Trader["Equity Trader"]
Bond_Trader["Bond Trader"]
Crypto_Trader["Crypto Trader"]
Technical_Analyst["Technical Analyst"]
Fundamental_Analyst["Fundamental Analyst"]
Compliance_Officer["Compliance Officer"]
Tax_Advisor["Tax Advisor"]
Portfolio_Manager --> Market_Analyst
Portfolio_Manager --> Risk_Manager
Portfolio_Manager --> Investment_Strategist
Investment_Strategist --> Equity_Trader
Investment_Strategist --> Bond_Trader
Investment_Strategist --> Crypto_Trader
Market_Analyst --> Technical_Analyst
Market_Analyst --> Fundamental_Analyst
Risk_Manager --> Compliance_Officer
Risk_Manager --> Tax_Advisor
🔧 FINANCIAL TOOLS AVAILABLE:
• get_portfolio_details
• recommend_trades
• analyze_market
• assess_portfolio_risk
• check_compliance
• execute_trade
============================================================
📈 RUNNING INVESTMENT PORTFOLIO SCENARIOS
============================================================
💼 Client (PORT001): My portfolio is worth $1.25M with 75% stocks. Given current market conditions, should I rebalance? I'm 45 years old with moderate risk tolerance.
📈 Processing with investment AI system...
✅ Investment Recommendation:
Based on my analysis of your $1.25M portfolio and current market conditions, here are my recommendations:
PORTFOLIO ASSESSMENT:
• Current allocation: 75% stocks, 15% bonds, 5% commodities, 5% cash
• Risk level: Moderate-High (above target for your profile)
• Age-appropriate allocation: 60-65% stocks (110 - age rule)
MARKET ANALYSIS:
• Equities: Currently bullish but showing moderate volatility
• Bonds: Bearish trend with inverted yield curve signaling caution
• Technology sector: Overweight at 35% (concentration risk)
REBALANCING RECOMMENDATIONS:
1. Reduce equity allocation from 75% to 65%
- Sell $125,000 in stocks (focus on overweight tech positions)
- Take profits on MSFT (+18.2%) and GOOGL (+15.1%)
2. Increase defensive positions:
- Add $75,000 to bonds (target 20% allocation)
- Add $50,000 to dividend aristocrats (VIG)
3. Risk Management:
- Current Sharpe ratio: 1.45 (good)
- Expected post-rebalance: 1.55 (improved)
- Reduces portfolio volatility by ~20%
ACTION PLAN:
✓ Immediate: Trim tech exposure (AAPL, MSFT)
✓ This week: Add bond allocation (BND, AGG)
✓ This month: Review and adjust international exposure
This rebalancing aligns with your moderate risk tolerance and time horizon of 20+ years to retirement.
📋 PORTFOLIO MANAGEMENT RESULTS:
{
"success": true,
"portfolio_id": "PORT001",
"request": "My portfolio is worth $1.25M with 75% stocks...",
"recommendation": "Based on my analysis...",
"timestamp": "2025-01-12T11:45:30.123456"
}
Customization Options
1. Add Alternative Investment Strategies
# Add Alternative Investment Specialist
alt_specialist = investment_strategist.create_worker(
name="Alternative_Investment_Specialist",
role="Alternative Assets Manager",
backstory="Expert in private equity, hedge funds, and real assets",
goal="Diversify portfolio with alternative investments"
)
2. Integrate Real-Time Market Data
@tool(schema={...})
async def get_realtime_quotes(symbols: List[str]):
"""Fetch real-time market quotes"""
# Integration with market data APIs
return {"quotes": market_data, "timestamp": timestamp}
3. Add Portfolio Optimization
@tool(schema={...})
async def optimize_portfolio(constraints: Dict, target_return: float):
"""Run portfolio optimization using modern portfolio theory"""
# Markowitz optimization implementation
return {"optimal_weights": weights, "efficient_frontier": data}
Production Considerations
- Market Data Integration: Connect to real-time market data feeds (Bloomberg, Reuters)
- Regulatory Compliance: Implement SEC, FINRA, and MiFID II compliance checks
- Risk Management: Real-time VaR calculations and stress testing
- Order Management: Integration with broker APIs for trade execution
- Audit Trail: Comprehensive logging of all investment decisions
- Security: Encryption of sensitive financial data and secure API access
- Disaster Recovery: Backup systems for continuous trading operations
- Performance Monitoring: Track slippage, execution quality, and alpha generation