A skeleton code for the tools that the agents will utilize to achieve their goals:
from langchain.tools import BaseTool
from langchain.pydantic_v1 import BaseModel, Field
from typing import Optional, List, Dict
import yfinance as yf
import pandas as pd
import numpy as np
from scipy.optimize import minimize
class StockData(BaseModel):
symbol: str = Field(..., description="Stock ticker symbol")
start_date: str = Field(..., description="Start date for historical data (YYYY-MM-DD)")
end_date: str = Field(..., description="End date for historical data (YYYY-MM-DD)")
class PortfolioData(BaseModel):
stocks: List[str] = Field(..., description="List of stock ticker symbols")
weights: List[float] = Field(..., description="List of portfolio weights")
class FundamentalAnalysisTool(BaseTool):
name = "fundamental_analysis"
description = "Analyzes fundamental data for a given stock"
args_schema = StockData
def _run(self, symbol: str, start_date: str, end_date: str) -> Dict:
stock = yf.Ticker(symbol)
info = stock.info
financials = stock.financials
# Calculate key ratios
pe_ratio = info.get('trailingPE', None)
pb_ratio = info.get('priceToBook', None)
debt_to_equity = info.get('debtToEquity', None)
roe = info.get('returnOnEquity', None)
return {
"symbol": symbol,
"pe_ratio": pe_ratio,
"pb_ratio": pb_ratio,
"debt_to_equity": debt_to_equity,
"roe": roe,
# Add more fundamental metrics as needed
}
class MPTOptimizationTool(BaseTool):
name = "mpt_optimization"
description = "Optimizes portfolio using Modern Portfolio Theory"
args_schema = PortfolioData
def _run(self, stocks: List[str], weights: List[float]) -> Dict:
# Fetch historical data
data = yf.download(stocks, start="2020-01-01")['Adj Close']
# Calculate returns
returns = data.pct_change().dropna()
# Calculate expected returns and covariance
expected_returns = returns.mean()
cov_matrix = returns.cov()
# Perform portfolio optimization (this is a simplified version)
def portfolio_volatility(weights, cov_matrix):
return np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
def objective(weights, cov_matrix):
return portfolio_volatility(weights, cov_matrix)
constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
bounds = tuple((0, 1) for _ in range(len(stocks)))
result = minimize(objective, weights, args=(cov_matrix,), method='SLSQP', bounds=bounds, constraints=constraints)
optimized_weights = result.x
expected_return = np.sum(expected_returns * optimized_weights)
expected_volatility = portfolio_volatility(optimized_weights, cov_matrix)
return {
"optimized_weights": optimized_weights.tolist(),
"expected_return": expected_return,
"expected_volatility": expected_volatility
}
class RiskManagementTool(BaseTool):
name = "risk_management"
description = "Assesses portfolio risk"
args_schema = PortfolioData
def _run(self, stocks: List[str], weights: List[float]) -> Dict:
# Fetch historical data
data = yf.download(stocks, start="2020-01-01")['Adj Close']
returns = data.pct_change().dropna()
# Calculate portfolio return and volatility
portfolio_return = np.sum(returns.mean() * weights) * 252 # Annualized
portfolio_volatility = np.sqrt(np.dot(weights.T, np.dot(returns.cov() * 252, weights)))
# Calculate Value at Risk (VaR)
var_95 = np.percentile(np.sum(returns * weights, axis=1), 5)
return {
"portfolio_return": portfolio_return,
"portfolio_volatility": portfolio_volatility,
"var_95": var_95
}
class MarginOfSafetyTool(BaseTool):
name = "margin_of_safety"
description = "Calculates margin of safety for a stock"
args_schema = StockData
def _run(self, symbol: str, start_date: str, end_date: str) -> Dict:
stock = yf.Ticker(symbol)
info = stock.info
# This is a simplified calculation and should be more comprehensive in practice
book_value_per_share = info.get('bookValue', 0)
current_price = info.get('currentPrice', 0)
if book_value_per_share > 0 and current_price > 0:
margin_of_safety = (book_value_per_share - current_price) / book_value_per_share
else:
margin_of_safety = None
return {
"symbol": symbol,
"book_value_per_share": book_value_per_share,
"current_price": current_price,
"margin_of_safety": margin_of_safety
}
class MarketSentimentTool(BaseTool):
name = "market_sentiment"
description = "Analyzes market sentiment"
args_schema = StockData
def _run(self, symbol: str, start_date: str, end_date: str) -> Dict:
# This is a placeholder. In a real implementation, you would integrate
# with sentiment analysis APIs or use NLP models on news and social media data.
return {
"symbol": symbol,
"sentiment_score": 0.5, # Placeholder score
"sentiment": "Neutral" # Placeholder sentiment
}
# Create instances of the tools
fundamental_analysis_tool = FundamentalAnalysisTool()
mpt_optimization_tool = MPTOptimizationTool()
risk_management_tool = RiskManagementTool()
margin_of_safety_tool = MarginOfSafetyTool()
market_sentiment_tool = MarketSentimentTool()
# List of all tools
tools = [
fundamental_analysis_tool,
mpt_optimization_tool,
risk_management_tool,
margin_of_safety_tool,
market_sentiment_tool
]
# Example usage
if __name__ == "__main__":
# Fundamental Analysis
fa_result = fundamental_analysis_tool.run({"symbol": "AAPL", "start_date": "2023-01-01", "end_date": "2023-12-31"})
print("Fundamental Analysis Result:", fa_result)
# MPT Optimization
mpt_result = mpt_optimization_tool.run({"stocks": ["AAPL", "GOOGL", "MSFT"], "weights": [0.33, 0.33, 0.34]})
print("MPT Optimization Result:", mpt_result)
# Risk Management
risk_result = risk_management_tool.run({"stocks": ["AAPL", "GOOGL", "MSFT"], "weights": [0.33, 0.33, 0.34]})
print("Risk Management Result:", risk_result)
# Margin of Safety
mos_result = margin_of_safety_tool.run({"symbol": "AAPL", "start_date": "2023-01-01", "end_date": "2023-12-31"})
print("Margin of Safety Result:", mos_result)
# Market Sentiment
sentiment_result = market_sentiment_tool.run({"symbol": "AAPL", "start_date": "2023-01-01", "end_date": "2023-12-31"})
print("Market Sentiment Result:", sentiment_result)
A skeleton code for the tools that the agents will utilize to achieve their goals: