Add gurufocus scraper

Author: doruirimescu

Trading/config/config.py

@@ -29,3 +29,4 @@
 RATIO_STOCKS_PATH = TMP_PATH.joinpath("ratio_stocks/")
 ALERTS_PATH = CURRENT_FILE_PATH.joinpath("../live/alert/alert.json")
 CACHING_PATH = TMP_PATH.joinpath("caching")
+GURUFOCUS_DOWNLOADS_PATH = TMP_PATH.joinpath("gurufocus")

Trading/stock/gurufocus/cli.py

@@ -1,9 +1,10 @@
 #!/usr/bin/env python3
 
-from Trading.stock.gurufocus.gurufocus import download_html, extract_stock_info
+from Trading.stock.gurufocus.gurufocus import GurufocusAnalyzer
 from Trading.utils.google_search import get_first_google_result
-from Trading.utils.cli import Named
+from Trading.utils.cli import Named, JsonFileWriter
 from Trading.utils.custom_logging import get_logger
+from Trading.config.config import GURUFOCUS_DOWNLOADS_PATH
 from typing import Optional, List
 import os
 import fire
@@ -13,9 +14,13 @@
 
 
 # DEBUG=true cli.py analyze --names '["pdco", "paypal", "johnson&johnson", "mcdonalds", "pepsi", "uniper", "palantir"]'
-class GuruFocusCLI(Named):
-    def __init__(self, name: Optional[str] = None, names: Optional[List[str]] = None):
+class GuruFocusCLI(Named, JsonFileWriter):
+    def __init__(self, name: Optional[str] = None, names: Optional[List[str]] = None, filename: Optional[str] = None):
         Named.__init__(self, name=name, names=names)
+        if filename is None:
+            filename = GURUFOCUS_DOWNLOADS_PATH / "gurufocus.json"
+        JsonFileWriter.__init__(self, filename=filename)
+
     def analyze(self):
         if not self.name and not self.names:
             LOGGER.error("Name is required")
@@ -27,18 +32,20 @@ def analyze(self):
                 urls.append(get_first_google_result("gurufocus summary " + name))
         LOGGER.debug(f"URLs: {urls}")
 
-        for url in urls:
-            LOGGER.info(f"Scraping {url}")
-            download_html(url, filename="gurufocus_page.html")
-            html_file_path = "gurufocus_page.html"
-            stock_info = extract_stock_info(html_file_path)
-            if os.path.exists(html_file_path):
-                os.remove(html_file_path)
-            if not stock_info:
-                LOGGER.error(f"Failed to extract stock info from {url}")
-                continue
-            LOGGER.info(stock_info)
-            time.sleep(0.5)
+        gf_analyzer = GurufocusAnalyzer(self.json_file_writer)
+        gf_analyzer.run(items=urls, data={})
+        # for url in urls:
+        #     LOGGER.info(f"Scraping {url}")
+        #     download_html(url, filename="gurufocus_page.html")
+        #     html_file_path = "gurufocus_page.html"
+        #     stock_info = extract_stock_info(html_file_path)
+        #     if os.path.exists(html_file_path):
+        #         os.remove(html_file_path)
+        #     if not stock_info:
+        #         LOGGER.error(f"Failed to extract stock info from {url}")
+        #         continue
+        #     LOGGER.info(stock_info)
+        #     time.sleep(0.1)
 
 if __name__ == "__main__":
     fire.Fire(GuruFocusCLI)

Trading/stock/gurufocus/gurufocus.py

@@ -1,10 +1,10 @@
-import requests
 import re
-from bs4 import BeautifulSoup
-from dataclasses import dataclass, asdict
+from Trading.utils.html import to_beautiful_soup
+from stateful_data_processor.processor import StatefulDataProcessor
+from pydantic import BaseModel
 
-@dataclass
-class GurufocusAnalysis:
+
+class GurufocusAnalysis(BaseModel):
     company_name: str = None
     ticker: str = None
     market_cap: str = None
@@ -14,26 +14,14 @@ class GurufocusAnalysis:
     gf_value: float = None
     altman_z_score: float = None
 
-    def __iter__(self):
-        return iter(asdict(self).items())
-def download_html(url, filename="gurufocus_page.html"):
-    headers = {"User-Agent": "Mozilla/5.0"}
-    response = requests.get(url, headers=headers)
-    response.raise_for_status()
-
-    with open(filename, "w", encoding="utf-8") as f:
-        f.write(response.text)
-
-    return filename
-
-
 
 MARKET_CAP_REGEX = re.compile(
     r"Market Cap\s*[:\-]?\s*\$?\s*([\d\.]+\s*[MBT]?)", re.IGNORECASE
 )
-def extract_stock_info(html_file_path: str) -> GurufocusAnalysis:
-    with open(html_file_path, "r", encoding="utf-8") as file:
-        soup = BeautifulSoup(file, "html.parser")
+
+
+def extract_stock_info(html: str) -> GurufocusAnalysis:
+    soup = to_beautiful_soup(html)
 
     data = {
         "company_name": None,
@@ -133,11 +121,12 @@ def extract_stock_info(html_file_path: str) -> GurufocusAnalysis:
     return GurufocusAnalysis(**data)
 
 
-# Example usage
-# if __name__ == "__main__":
-# #     url = "https://www.gurufocus.com/stock/GOGL/summary"
-# # html_file = download_html(url)
-# # print(f"Downloaded HTML to {html_file}")
-#     html_path = "gurufocus_page.html"  # Adjust path if needed
-#     info = extract_stock_info(html_path)
-#     print(info)
+class GurufocusAnalyzer(StatefulDataProcessor):
+    def __init__(self, json_file_writer=None, logger=None):
+        super().__init__(json_file_writer, logger=logger)
+        self.data = {}
+
+    def process_item(self, item, iteration_index, data):
+        url = item
+        stock_info = extract_stock_info(url)
+        self.data[stock_info.ticker] = stock_info.dict()

Trading/utils/html.py

@@ -0,0 +1,23 @@
+from bs4 import BeautifulSoup
+import requests
+
+def download_html(url, filename: str):
+    headers = {"User-Agent": "Mozilla/5.0"}
+    response = requests.get(url, headers=headers)
+    response.raise_for_status()
+
+    with open(filename, "w", encoding="utf-8") as f:
+        f.write(response.text)
+
+    return filename
+
+def load_html(url) -> str:
+    headers = {"User-Agent": "Mozilla/5.0"}
+    response = requests.get(url, headers=headers)
+    response.raise_for_status()
+
+    return response.text
+
+def to_beautiful_soup(url: str) -> BeautifulSoup:
+    html = load_html(url)
+    return BeautifulSoup(html, "html.parser")