Open SaudAltamimi opened 3 months ago
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: