Add financial data plugin

.github/CODEOWNERS

@@ -5,6 +5,7 @@
 /src/autogpt_plugins/scenex @delgermurun 
 /src/autogpt_plugins/bing_search @ForestLinSen
 /src/autogpt_plugins/baidu_search @ForestLinSen
+/src/autogpt_plugins/financial_data @ForestLinSen
 /src/autogpt_plugins/news_search @PalAditya
 /src/autogpt_plugins/wikipedia_search @pierluigi-failla
 /src/autogpt_plugins/api_tools @sidewaysthought

README.md

@@ -74,6 +74,7 @@ You can also see the plugins here:
 | Twitter          | AutoGPT is capable of retrieving Twitter posts and other related content by accessing the Twitter platform via the v1.1 API using Tweepy.               | [autogpt_plugins/twitter](https://github.com/Significant-Gravitas/Auto-GPT-Plugins/tree/master/src/autogpt_plugins/twitter)           |
 | Wikipedia Search | This allows AutoGPT to use Wikipedia directly.                                                                    | [autogpt_plugins/wikipedia_search](https://github.com/Significant-Gravitas/Auto-GPT-Plugins/tree/master/src/autogpt_plugins/wikipedia_search) |
 | Telegram |  A smoothly working Telegram bot that gives you all the messages you would normally get through the Terminal. | [autogpt_plugins/telegram](https://github.com/Significant-Gravitas/Auto-GPT-Plugins/tree/master/src/autogpt_plugins/telegram) |
+| Financial Data      | Fetches, processes, and analyzes financial data for a given company                                                  | [autogpt_plugins/financial_data](https://github.com/Significant-Gravitas/Auto-GPT-Plugins/tree/master/src/autogpt_plugins/financial_data)       |
 
 
 Some third-party plugins have been created by contributors that are not included in this repository. For more information about these plugins, please visit their respective GitHub pages.

requirements.txt

@@ -21,4 +21,6 @@ colorama
 atproto
 requests
 bs4
-python-telegram-bot
+python-telegram-bot
+yfinance
+yahooquery

src/autogpt_plugins/financial_data/__init__.py

@@ -0,0 +1,105 @@
+"""This is the plugin for generating company financial data for Auto-GPT."""
+import os
+from typing import Any, Dict, List, Optional, Tuple, TypedDict, TypeVar
+from auto_gpt_plugin_template import AutoGPTPluginTemplate
+from .financial_analysis import generate_financial_data
+
+PromptGenerator = TypeVar("PromptGenerator")
+
+
+class Message(TypedDict):
+    role: str
+    content: str
+
+
+class AutoGPTFinancialAnalysis(AutoGPTPluginTemplate):
+    def __init__(self):
+        super().__init__()
+        self._name = "Financial-Analysis"
+        self._version = "0.1.0"
+        self.needAnalyse = False
+        self.symbol = ""
+        self._description = (
+            "Generate Company Financial Data for Analysis Plugin for Auto-GPT. "
+        )
+
+    def can_handle_post_prompt(self) -> bool:
+        return True
+
+    def post_prompt(self, prompt: PromptGenerator) -> PromptGenerator:
+        prompt.add_command(
+                "Generate Financial Data",
+                "generate_financial_data",
+                {"symbol": "<symbol>"},
+                generate_financial_data,
+            )
+        return prompt
+
+    def can_handle_pre_command(self) -> bool:
+        return True
+
+    def pre_command(
+        self, command_name: str, arguments: Dict[str, Any]
+    ) -> Tuple[str, Dict[str, Any]]:
+        if command_name == "generate_financial_data":
+            symbol = arguments["symbol"]
+            prompt = generate_financial_data(symbol)
+            # getcwd + auto_gpt_workspace
+            filename = os.getcwd() + f"/autogpt/auto_gpt_workspace/financial_analysis_{symbol}.txt"
+            print(filename)
+            return "write_to_file", {"filename": filename, "text": prompt}
+        return command_name, arguments
+
+    def can_handle_post_command(self) -> bool:
+        return False
+
+    def post_command(self, command_name: str, response: str) -> str:
+        pass
+
+    def can_handle_on_planning(self) -> bool:
+        return False
+
+    def on_planning(
+        self, prompt: PromptGenerator, messages: List[Message]
+    ) -> Optional[str]:
+        pass
+
+    def can_handle_on_response(self) -> bool:
+        return False
+
+    def on_response(self, response: str, *args, **kwargs) -> str:
+        pass
+
+    def can_handle_post_planning(self) -> bool:
+        return False
+
+    def post_planning(self, response: str) -> str:
+        pass
+
+    def can_handle_pre_instruction(self) -> bool:
+        return False
+
+    def pre_instruction(self, messages: List[Message]) -> List[Message]:
+        pass
+
+    def can_handle_on_instruction(self) -> bool:
+        return False
+
+    def on_instruction(self, messages: List[Message]) -> Optional[str]:
+        pass
+
+    def can_handle_post_instruction(self) -> bool:
+        return False
+
+    def post_instruction(self, response: str) -> str:
+        pass
+
+    def can_handle_chat_completion(
+        self, messages: Dict[Any, Any], model: str, temperature: float, max_tokens: int
+    ) -> bool:
+        return False
+
+    def handle_chat_completion(
+        self, messages: List[Message], model: str, temperature: float, max_tokens: int
+    ) -> str:
+        pass

src/autogpt_plugins/financial_data/financial_analysis.py

@@ -0,0 +1,138 @@
+import yfinance as yf
+import pandas as pd
+from yahooquery import Ticker
+from datetime import datetime 
+import requests
+from bs4 import BeautifulSoup
+import os
+
+def download_data(symbol, start_date, end_date):
+    data = yf.download(symbol, start_date, end_date)
+    return data
+
+def fetch_financial_data(symbol):
+    ticker = Ticker(symbol)
+
+    # Fetch income statement, balance sheet and cash flow
+    income_statement = ticker.income_statement('q')
+    balance_sheet = ticker.balance_sheet('q')
+    cash_flow = ticker.cash_flow('q')
+
+    income_statement = income_statement[income_statement.periodType == '3M']
+    balance_sheet = balance_sheet[balance_sheet.periodType == '3M']
+    cash_flow = cash_flow[cash_flow.periodType == '3M']
+
+    # Filter for key financial metrics
+    income_statement_keys = ['asOfDate', 'TotalRevenue', 'CostOfRevenue', 'GrossProfit', 'OperatingIncome', 'NetIncome', 'OperatingExpense', 'ResearchAndDevelopment']
+    balance_sheet_keys = ['asOfDate', 'TotalAssets', 'TotalLiabilitiesNetMinorityInterest', 'StockholdersEquity', 'CashAndCashEquivalents', 'LongTermDebt', 'CurrentAssets', 'CurrentLiabilities']
+    cash_flow_keys = ['asOfDate', 'OperatingCashFlow', 'InvestingCashFlow', 'FinancingCashFlow', 'FreeCashFlow', 'NetIncome']
+
+    # Filter financial data and prepare for presentation
+    income_statement = income_statement.loc[:, income_statement_keys]
+    balance_sheet = balance_sheet.loc[:, balance_sheet_keys]
+    cash_flow = cash_flow.loc[:, cash_flow_keys]
+
+    # Sort and get the latest 4 quarters
+    income_statement = income_statement.sort_values('asOfDate', ascending=False).head(4)
+    balance_sheet = balance_sheet.sort_values('asOfDate', ascending=False).head(4)
+    cash_flow = cash_flow.sort_values('asOfDate', ascending=False).head(4)
+
+    # Present data in a natural language format
+    result = f"Financial Analysis for {symbol}:\n\n"
+    for _, row in income_statement.iterrows():
+        result += f"\nAs of the quarter ending {row['asOfDate']}, {symbol} had:\n"
+        for key in income_statement_keys[1:]:
+            result += f"{key}: ${row[key]:,.2f}\n"
+
+    for _, row in balance_sheet.iterrows():
+        result += f"\nFor the quarter ending {row['asOfDate']}, {symbol}'s balance sheet was:\n"
+        for key in balance_sheet_keys[1:]:
+            result += f"{key}: ${row[key]:,.2f}\n"
+
+    for _, row in cash_flow.iterrows():
+        result += f"\nFor the quarter ending {row['asOfDate']}, {symbol}'s cash flow was:\n"
+        for key in cash_flow_keys[1:]:
+            result += f"{key}: ${row[key]:,.2f}\n"
+
+    return result
+
+def calc_stats(symbol, periods=[30, 90, 180, 365, 3 * 365, 5 * 365]):
+    descriptions = []
+    ticker = yf.Ticker(symbol)
+    ipo_date = ticker.info.get("ipoDate")
+
+    # Convert IPO date string to datetime object, if it exists
+    if ipo_date:
+        ipo_date = datetime.strptime(ipo_date, "%Y-%m-%d")
+
+    # Download all available data for this stock
+    all_data = download_data(symbol, ipo_date, datetime.today())
+
+    for period in periods:
+        # Check if the company has been listed for the specified number of days
+        if len(all_data.index) < period:
+            descriptions.append(
+                f"{symbol} has not been listed for {period} trading days."
+            )
+            continue
+
+        data = all_data.tail(period).copy()
+
+        start_date = data.index[0].strftime("%Y-%m-%d")
+        end_date = data.index[-1].strftime("%Y-%m-%d")
+        start_price = data["Adj Close"].iloc[0]
+        end_price = data["Adj Close"].iloc[-1]
+        pct_change = (end_price - start_price) / start_price * 100
+        high = data["High"].max()
+        high_date = data["High"].idxmax().strftime("%Y-%m-%d")
+        low = data["Low"].min()
+        low_date = data["Low"].idxmin().strftime("%Y-%m-%d")
+        avg_volume = data["Volume"].mean()
+        data["Return"] = data["Adj Close"].pct_change()
+        volatility = data["Return"].std()
+        description = f"For the last {period} days ({start_date} to {end_date}):\n"
+        description += (
+            f"Start price was {start_price:.2f} and end price was {end_price:.2f}. "
+        )
+        description += f"This is a percentage change of {pct_change:.2f}%.\n"
+        description += f"Highest price was {high:.2f} on {high_date}, "
+        description += f"and lowest price was {low:.2f} on {low_date}.\n"
+        description += f"Average volume was {avg_volume:.2f}, "
+        description += f"and the standard deviation of daily returns (volatility) was {volatility:.2f}."
+
+        descriptions.append(description)
+
+    return "\n\n".join(descriptions)
+
+def get_yahoo_finance_news(symbol):
+    url = f"https://feeds.finance.yahoo.com/rss/2.0/headline?s={symbol}&region=US&lang=en-US"
+
+    headers = {
+        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3'
+    }
+    response = requests.get(url, headers=headers)
+
+    soup = BeautifulSoup(response.content, features='xml')
+    items = soup.findAll('item')
+    result = f"Latest news for {symbol}:\n\n"
+    for item in items:
+        title = item.title.text
+        pubDate = item.pubDate.text
+        description = item.description.text
+
+        # Add the item's details to the result
+        result += f"Title: {title}\n"
+        result += f"Published: {pubDate}\n"
+        result += f"Description: {description}\n\n"
+
+    return result
+
+def generate_financial_data(symbol):
+        prompt = "As a senior analyst, your task is to write a comprehensive analysis report on a given stock using the provided information below. The report should offer a detailed overview of the company's financial condition, covering all essential aspects of its financial performance. Additionally, it should include your insights and a summary. Aim for a highly detailed report, consisting of approximately 3000 words."
+        prompt += "\n\n"
+        prompt += fetch_financial_data(symbol)
+        prompt += "\n\n"
+        prompt += calc_stats(symbol)
+        prompt += "\n\n"
+        prompt += get_yahoo_finance_news(symbol)
+        return prompt

src/autogpt_plugins/financial_data/test_auto_gpt_financial_analysis.py

@@ -0,0 +1,110 @@
+import unittest
+from unittest.mock import patch, Mock
+from datetime import datetime
+import yfinance as yf
+import pandas as pd
+import numpy as np
+from .financial_analysis import download_data, fetch_financial_data, calc_stats, get_yahoo_finance_news, generate_financial_data
+from . import AutoGPTFinancialAnalysis
+
+class TestAutoGPTFinancialAnalysis(unittest.TestCase):
+    def setUp(self):
+        self.plugin = AutoGPTFinancialAnalysis()
+        self.symbol = "AAPL"
+        self.dummy_data = pd.DataFrame({
+            'High': np.random.rand(10) * 100,
+            'Low': np.random.rand(10) * 100,
+            'Open': np.random.rand(10) * 100,
+            'Close': np.random.rand(10) * 100,
+            'Volume': np.random.rand(10) * 1000000,
+            'Adj Close': np.random.rand(10) * 100,
+        })
+
+    def test_can_handle_post_prompt(self):
+        self.assertTrue(self.plugin.can_handle_post_prompt())
+
+    def test_can_handle_pre_command(self):
+        self.assertTrue(self.plugin.can_handle_pre_command())
+
+    @patch('autogpt_plugins.financial_data.financial_analysis.yf.download')
+    def test_download_data(self, mock_download):
+        mock_download.return_value = self.dummy_data
+        data = download_data(self.symbol, "2022-01-01", "2023-01-01")
+        self.assertTrue(isinstance(data, pd.DataFrame))
+        self.assertEqual(len(data), 10)
+
+    @patch('autogpt_plugins.financial_data.financial_analysis.Ticker')
+    def test_fetch_financial_data(self, mock_ticker):
+        df_income = pd.DataFrame({
+            'periodType': ['3M', '3M', '6M', '6M'],
+            'asOfDate': pd.date_range(end='1/1/2022', periods=4),
+            'TotalRevenue': np.random.rand(4) * 1000,
+            'CostOfRevenue': np.random.rand(4) * 1000,
+            'GrossProfit': np.random.rand(4) * 1000,
+            'OperatingIncome': np.random.rand(4) * 1000,
+            'NetIncome': np.random.rand(4) * 1000,
+            'OperatingExpense': np.random.rand(4) * 1000,
+            'ResearchAndDevelopment': np.random.rand(4) * 1000,
+        })
+
+        df_balance = pd.DataFrame({
+            'periodType': ['3M', '3M', '6M', '6M'],
+            'asOfDate': pd.date_range(end='1/1/2022', periods=4),
+            'TotalAssets': np.random.rand(4) * 1000,
+            'TotalLiabilitiesNetMinorityInterest': np.random.rand(4) * 1000,
+            'StockholdersEquity': np.random.rand(4) * 1000,
+            'CashAndCashEquivalents': np.random.rand(4) * 1000,
+            'LongTermDebt': np.random.rand(4) * 1000,
+            'CurrentAssets': np.random.rand(4) * 1000,
+            'CurrentLiabilities': np.random.rand(4) * 1000,
+        })
+
+        df_cash_flow = pd.DataFrame({
+            'periodType': ['3M', '3M', '6M', '6M'],
+            'asOfDate': pd.date_range(end='1/1/2022', periods=4),
+            'OperatingCashFlow': np.random.rand(4) * 1000,
+            'InvestingCashFlow': np.random.rand(4) * 1000,
+            'FinancingCashFlow': np.random.rand(4) * 1000,
+            'FreeCashFlow': np.random.rand(4) * 1000,
+            'NetIncome': np.random.rand(4) * 1000,
+        })
+
+        mock_ticker.return_value.income_statement.return_value = df_income
+        mock_ticker.return_value.balance_sheet.return_value = df_balance
+        mock_ticker.return_value.cash_flow.return_value = df_cash_flow
+        result = fetch_financial_data(self.symbol)
+        self.assertTrue(isinstance(result, str))
+
+    @patch('autogpt_plugins.financial_data.financial_analysis.yf.Ticker')
+    @patch('autogpt_plugins.financial_data.financial_analysis.download_data')
+    def test_calc_stats(self, mock_download_data, mock_ticker):
+        mock_download_data.return_value = self.dummy_data
+        mock_ticker.return_value.info.get.return_value = "2022-01-01"
+        result = calc_stats(self.symbol)
+        self.assertTrue(isinstance(result, str))
+
+    @patch('autogpt_plugins.financial_data.financial_analysis.requests.get')
+    def test_get_yahoo_finance_news(self, mock_get):
+        mock_response = Mock()
+        mock_response.content = b'<rss><channel><item><title>Test news</title><pubDate>2023-05-19</pubDate><description>Test description</description></item></channel></rss>'
+        mock_get.return_value = mock_response
+        result = get_yahoo_finance_news(self.symbol)
+        self.assertTrue(isinstance(result, str))
+
+    @patch('autogpt_plugins.financial_data.financial_analysis.fetch_financial_data')
+    @patch('autogpt_plugins.financial_data.financial_analysis.calc_stats')
+    @patch('autogpt_plugins.financial_data.financial_analysis.get_yahoo_finance_news')
+    def test_generate_financial_data(self, mock_news, mock_stats, mock_fin_data):
+        mock_news.return_value = "News"
+        mock_stats.return_value = "Stats"
+        mock_fin_data.return_value = "Financial data"
+        result = generate_financial_data(self.symbol)
+        self.assertTrue(isinstance(result, str))
+
+    @patch('autogpt_plugins.financial_data.financial_analysis.os.getcwd')
+    def test_pre_command(self, mock_getcwd):
+        mock_getcwd.return_value = '/home'
+        command_name, arguments = self.plugin.pre_command(
+            "generate_financial_data", {"symbol": self.symbol}
+        )
+        self.assertEqual(command_name, "write_to_file")