Spaces:
Sleeping
Sleeping
File size: 8,199 Bytes
99c70a5 53256d4 99c70a5 093b44b 99c70a5 093b44b 99c70a5 | 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 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | {
"cells": [
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from typing import Union, Dict, TypedDict, Annotated\n",
"\n",
"import dotenv\n",
"from IPython.display import Image, display\n",
"from langgraph.graph import StateGraph, START, END\n",
"from langchain_openai import ChatOpenAI\n",
"from langchain_core.messages import SystemMessage, AIMessage, HumanMessage\n",
"from langchain_core.tools import tool\n",
"from langgraph.prebuilt import ToolNode, tools_condition\n",
"from langgraph.graph.message import add_messages\n",
"\n",
"from src.technical_analysis import TechnicalAnalysis\n",
"from src.fundamental_analysis import FundamentalAnalysis\n",
"dotenv.load_dotenv(dotenv.find_dotenv())"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"@tool\n",
"def get_stock_prices(ticker: str) -> Union[Dict, str]:\n",
" \"\"\"Fetches historical stock price data and technical indicator for a given ticker.\"\"\"\n",
" df = TechnicalAnalysis(\n",
" ticker=ticker,\n",
" fetchperiodinweeks=12,\n",
" plot_ta=False,\n",
" debug=False).run()\n",
" if df.shape[0] > 0:\n",
" df['Date'] = df.index.astype(str)\n",
" # split the data into price and indicators, and take the last 10 days of data\n",
" dict_price = df[['Date','Close', 'High', 'Low', 'Open', 'Volume']].iloc[-10:,:].to_dict(orient='records')\n",
" indicators = df[['VWAP', 'RSI', 'StochOsc', 'MACD', 'MACDsig', 'MACDdif']].iloc[-10:,:].to_dict(orient='records')\n",
" return {'stock_price': dict_price, 'indicators': indicators}\n",
" else:\n",
" return f\"Error fetching technical data for ticker: {ticker}\"\n",
"\n",
"@tool\n",
"def get_financial_metrics(ticker: str) -> Union[Dict, str]:\n",
" \"\"\"Fetches key financial metrics for a given ticker.\"\"\"\n",
" dict_fundamentals = FundamentalAnalysis(\n",
" ticker=ticker).run()\n",
" if len(dict_fundamentals) > 0:\n",
" return dict_fundamentals\n",
" else:\n",
" return f\"Error fetching financial metrics for ticker: {ticker}\"\n",
"\n",
"def get_prompt(company: str):\n",
" prompt = \"\"\"\n",
" You are a stock analyst specializing in evaluating the performance of a given company (whose symbol is {company})\n",
" based on recent price data and technical indicators as well as financial metrics. \n",
" Your task is to provide a comprehensive summary of the technical and fundamental analysis for a given stock,\n",
" and based on the analysis, choose an action among the following options:\n",
" - strong sell: there are overwhelmingly bad signals for the stock, sell large quantities\n",
" - sell: there are some bad signals for the stock, sell medium quantities\n",
" - hold: there are either neutral signals, or good signals mixed with bad signals for the stock, hold\n",
" - buy: there are some good signals for the stock, buy medium quantities\n",
" - strong buy: there are overwhelmingly good signals for the stock, buy large quantities\n",
"\n",
" You have access to the following tools:\n",
" 1. **get_stock_prices**: Retrieves the latest stock price, historical price data and technical Indicators like VWAP, RSI, Stochastic Oscillator and MACD metrics. \n",
" 2. **get_financial_metrics**: Retrieves key financial metrics, such as revenue, earnings per share (EPS), price-to-earnings ratio (P/E), and debt-to-equity ratio.\n",
"\n",
" ### Your Task:\n",
" 1. **Input Stock Symbol**: Use the provided stock symbol to query the tools and gather the relevant information.\n",
" 2. **Analyze Data**: Evaluate the results from the tools and identify potential resistance, key trends, strengths, or concerns.\n",
" 3. **Provide Summary**: Write a concise, well-structured summary that highlights:\n",
" - Recent stock price movements, trends and potential resistance.\n",
" - Key insights from technical indicators (e.g., whether the stock is overbought or oversold).\n",
" - Financial health and performance based on financial metrics.\n",
"\n",
" ### Constraints:\n",
" - Use only the data provided by the tools.\n",
" - If any tool fails to provide data, clearly state that in your summary.\n",
" - Avoid speculative language; focus on observable data and trends.\n",
" - Ensure that your response is objective, concise, and actionable.\n",
"\n",
" ### Output Format:\n",
" Respond in the following format:\n",
" \"Stock\": \"<Stock symbol>\",\n",
" \"Price analysis\": \"<Detailed analysis of stock price trends>\",\n",
" \"Technical analysis\": \"<Detailed analysis of technical indicators>\",\n",
" \"Fundamental analysis\": \"<Detailed analysis of financial metrics>\",\n",
" \"Final summary\": \"<Conclusive summary of the analyses above>\"\n",
" \"Recommended action\": \"<Suggested action based on the above analyses among options: [strong sell, sell, hold, buy, strong buy]>\"\n",
" \"\"\"\n",
" return prompt.format(company=company)\n",
"\n",
"class State(TypedDict):\n",
" messages: Annotated[list, add_messages]\n",
" stock: str\n",
"\n",
"def build_graph():\n",
" graph_builder = StateGraph(State)\n",
"\n",
" tools = [get_stock_prices, get_financial_metrics]\n",
" llm = ChatOpenAI(model='gpt-4o-mini')\n",
" llm_with_tool = llm.bind_tools(tools)\n",
"\n",
" def stock_analyst(state: State):\n",
" \n",
" messages = [\n",
" SystemMessage(content=get_prompt(state['stock'])), \n",
" ] + state['messages']\n",
" return {\n",
" 'messages': llm_with_tool.invoke(messages)\n",
" }\n",
" \n",
" graph_builder.add_node('stock_analyst', stock_analyst)\n",
" graph_builder.add_edge(START, 'stock_analyst')\n",
" graph_builder.add_node(ToolNode(tools))\n",
" graph_builder.add_conditional_edges('stock_analyst', tools_condition)\n",
" graph_builder.add_edge('tools', 'stock_analyst')\n",
" \n",
" # graph_builder.add_edge('stock_analyst', END)\n",
"\n",
" graph = graph_builder.compile()\n",
" return graph\n",
"\n",
"def draw_graph(graph):\n",
" try:\n",
" display(Image(graph.get_graph().draw_mermaid_png()))\n",
" except Exception:\n",
" # This requires some extra dependencies and is optional\n",
" pass\n",
"\n",
"def get_stock_suggestion(graph, stock):\n",
" events = graph.stream({'messages':[('user', 'Should I buy this stock?')],\n",
" 'stock': stock}, stream_mode='values')\n",
" for event in events:\n",
" if 'messages' in event:\n",
" event['messages'][-1].pretty_print()\n",
" "
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"graph = build_graph()\n",
"draw_graph(graph)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"get_stock_suggestion(graph, 'GOOG')"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"get_stock_suggestion(graph, 'AAPL')"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "finagents_py311",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.1"
}
},
"nbformat": 4,
"nbformat_minor": 2
}
|