Add get_holdings method to MarketWatchClient

marketwatch/__init__.py

@@ -647,6 +647,68 @@ def get_ticker_info(self, ticker: str) -> dict:
         except Exception as err:
             raise MarketWatchException(f"Other error occurred: {err}")
 
+    def get_holdings(self, ticker: str) -> dict:
+        """
+        Get holdings information for a given fund ticker from MarketWatch.
+
+        :param ticker: Ticker symbol of the fund.
+        :return: Dictionary with holdings information.
+        """
+        try:
+            # Send a GET request to the MarketWatch holdings URL for the given ticker with follow_redirects=True
+            url = f"https://www.marketwatch.com/investing/fund/{ticker.lower()}/holdings"
+            response = self.session.get(url, follow_redirects=True)
+            response.raise_for_status()  # Will raise HTTPError for 4XX/5XX status
+
+            # Check if the URL has been redirected to a stock or index URL
+            if "investing/stock" in str(response.url) or "investing/index" in str(response.url):
+                raise MarketWatchException(f"The ticker {ticker.upper()} is not a fund, it is a stock or index.")
+
+
+            # Parse the HTML content using BeautifulSoup
+            soup = BeautifulSoup(response.text, "html.parser")
+
+            # Extract sector allocation
+            sector_allocation = {}
+            sector_table = soup.select_one('table.value-pairs.no-heading')
+            if sector_table:
+                for row in sector_table.select('tr.table__row'):
+                    cells = row.select('td.table__cell')
+                    if len(cells) == 2:
+                        sector = cells[0].get_text(strip=True)
+                        allocation = cells[1].get_text(strip=True)
+                        sector_allocation[sector] = allocation
+
+            # Extract top 25 holdings
+            top_holdings = []
+            holdings_table = soup.select_one('.element--table.holdings table.table--primary')
+            if holdings_table:
+                for row in holdings_table.select('tbody tr.table__row'):
+                    cells = row.select('td.table__cell')
+                    if len(cells) == 3:
+                        company = cells[0].get_text(strip=True)
+                        symbol = cells[1].get_text(strip=True)
+                        net_assets = cells[2].get_text(strip=True)
+                        top_holdings.append({
+                            "company": company,
+                            "symbol": symbol,
+                            "net_assets": net_assets
+                        })
+
+            # Compile the result
+            result = {
+                "ticker": ticker.upper(),
+                "sector_allocation": sector_allocation,
+                "top_holdings": top_holdings
+            }
+
+            return result
+
+        except httpx.HTTPError as http_err:
+            raise MarketWatchException(f"HTTP error occurred: {http_err}")
+        except Exception as err:
+            raise MarketWatchException(f"Other error occurred: {err}")
+
     @auth
     def get_portfolio(self, game_id: str):
         """

test/test_ticker.py

@@ -71,4 +71,51 @@ def test_get_ticker_info(authenticated_marketwatch):
         assert "1 Month" in performance
         assert "3 Month" in performance
         assert "YTD" in performance
-        assert "1 Year" in performance
+        assert "1 Year" in performance
+
+def test_get_holdings_with_fund_ticker(authenticated_marketwatch):
+    mw = authenticated_marketwatch
+    ticker = "aiq"
+    result = mw.get_holdings(ticker)
+
+    # Verificar que el resultado es un diccionario
+    assert result is not None
+    assert isinstance(result, dict)
+
+    # Verificar campos generales
+    assert "ticker" in result
+    assert result["ticker"] == "AIQ"
+    assert "sector_allocation" in result
+    assert isinstance(result["sector_allocation"], dict)
+    assert "top_holdings" in result
+    assert isinstance(result["top_holdings"], list)
+
+    # Verificar contenido de sector_allocation
+    sector_allocation = result["sector_allocation"]
+    assert "Technology" in sector_allocation
+    assert "Consumer Services" in sector_allocation
+    assert "Industrials" in sector_allocation
+
+    # Verificar contenido de top_holdings
+    top_holdings = result["top_holdings"]
+    assert len(top_holdings) > 0
+    assert all(isinstance(item, dict) for item in top_holdings)
+    assert all("company" in item for item in top_holdings)
+    assert all("symbol" in item for item in top_holdings)
+    assert all("net_assets" in item for item in top_holdings)
+
+def test_get_holdings_with_stock_ticker(authenticated_marketwatch):
+    mw = authenticated_marketwatch
+    ticker = "meli"
+
+    with pytest.raises(MarketWatchException) as exc_info:
+        mw.get_holdings(ticker)
+
+    assert str(exc_info.value) == "Other error occurred: The ticker MELI is not a fund, it is a stock or index."
+
+    ticker_index = "spx"
+
+    with pytest.raises(MarketWatchException) as exc_info:
+        mw.get_holdings(ticker_index)
+
+    assert str(exc_info.value) == "Other error occurred: The ticker SPX is not a fund, it is a stock or index."