Add support for watchlists

README.md

@@ -25,7 +25,7 @@ Install using pypi:
 pip install firstrade
 ```
 
-## Quikstart
+## Quickstart
 
 The code in `test.py` will:
 - Login and print account info.
@@ -36,6 +36,7 @@ The code in `test.py` will:
 - Get OHLC data
 - Get an option Dates, Quotes, and Greeks
 - Place a dry run option order
+- List watchlists, create one, add a symbol to it then delete it
 ---
 
 ## Implemented Features
@@ -51,6 +52,7 @@ The code in `test.py` will:
 - [x] Cancel placed orders
 - [x] Options (Orders, Quotes, Greeks)
 - [x] Order History
+- [x] Manage watchlists
 
 ## TO DO
 

firstrade/__init__.py

@@ -1,3 +1,3 @@
-from . import account, order, symbols, urls
+from . import account, order, symbols, urls, watchlist
 
-__all__ = ["account", "order", "symbols", "urls"]
+__all__ = ["account", "order", "symbols", "urls", "watchlist"]

firstrade/urls.py

@@ -88,6 +88,31 @@ def option_order() -> str:
     return "https://api3x.firstrade.com/private/option_order"
 
 
+def watchlists() -> str:
+    """Watchlist collection URL for FirstTrade API (list all / create)."""
+    return "https://api3x.firstrade.com/private/watchlists"
+
+
+def watchlist(list_id: int) -> str:
+    """Single watchlist URL for FirstTrade API (get / delete)."""
+    return f"https://api3x.firstrade.com/private/watchlists/{list_id}"
+
+
+def watchlist_items() -> str:
+    """All watchlist items URL for FirstTrade API."""
+    return "https://api3x.firstrade.com/private/all_watchlist_items"
+
+
+def watchlist_item(list_id: int) -> str:
+    """Add item to watchlist URL for FirstTrade API."""
+    return f"https://api3x.firstrade.com/private/watchlist/{list_id}"
+
+
+def watchlist_item_delete(watchlist_id: int) -> str:
+    """Delete a single watchlist item URL for FirstTrade API."""
+    return f"https://api3x.firstrade.com/private/watchlist/{watchlist_id}"
+
+
 def session_headers() -> dict[str, str]:
     """Session headers for FirstTrade API."""
     headers: dict[str, str] = {

firstrade/watchlist.py

@@ -0,0 +1,130 @@
+from firstrade import urls
+from firstrade.account import FTSession
+
+
+class Watchlist:
+    """Provides watchlist management for a Firstrade session.
+
+    Supports creating and deleting watchlists, adding and removing symbols
+    from watchlists, and retrieving watchlist contents.
+
+    Attributes:
+        ft_session (FTSession): The session object used for API requests.
+
+    """
+
+    def __init__(self, ft_session: FTSession) -> None:
+        """Initialize Watchlist with a Firstrade session.
+
+        Args:
+            ft_session (FTSession): An authenticated Firstrade session.
+
+        """
+        self.ft_session: FTSession = ft_session
+
+    # ------------------------------------------------------------------
+    # Watchlist CRUD
+    # ------------------------------------------------------------------
+
+    def get_watchlists(self) -> dict:
+        """Retrieve all watchlists for the current user.
+
+        Returns:
+            dict: API response containing a list of watchlists under ``items``.
+                  Each entry contains ``list_id``, ``name``, and ``isDefault``.
+
+        """
+        response = self.ft_session._request("get", url=urls.watchlists())
+        return response.json()
+
+    def create_watchlist(self, name: str) -> dict:
+        """Create a new watchlist.
+
+        Args:
+            name (str): Display name for the new watchlist.
+
+        Returns:
+            dict: API response containing ``result.list_id`` of the created watchlist.
+
+        """
+        data = {"name": name}
+        response = self.ft_session._request("post", url=urls.watchlists(), data=data)
+        return response.json()
+
+    def get_watchlist(self, list_id: int) -> dict:
+        """Retrieve the contents of a specific watchlist.
+
+        Args:
+            list_id (int): The ID of the watchlist to retrieve.
+
+        Returns:
+            dict: API response containing ``result.list_items`` with each symbol's
+                  quote data.
+
+        """
+        response = self.ft_session._request("get", url=urls.watchlist(list_id))
+        return response.json()
+
+    def delete_watchlist(self, list_id: int) -> dict:
+        """Delete a watchlist.
+
+        Args:
+            list_id (int): The ID of the watchlist to delete.
+
+        Returns:
+            dict: API response confirming deletion via ``result.result == "success"``.
+
+        """
+        response = self.ft_session._request("delete", url=urls.watchlist(list_id))
+        return response.json()
+
+    # ------------------------------------------------------------------
+    # Watchlist item management
+    # ------------------------------------------------------------------
+
+    def get_all_watchlist_items(self) -> dict:
+        """Retrieve every item across all watchlists.
+
+        Returns:
+            dict: API response with all watchlist symbols under ``items``.
+
+        """
+        response = self.ft_session._request("get", url=urls.watchlist_items())
+        return response.json()
+
+    def add_symbol(self, list_id: int, symbol: str, sec_type: str = "1") -> dict:
+        """Add a symbol to a watchlist.
+
+        Args:
+            list_id (int): The ID of the watchlist to add the symbol to.
+            symbol (str): The ticker symbol to add (e.g. ``"AAPL"``).
+            sec_type (str, optional): Security type. ``"1"`` for equities/ETFs.
+                Defaults to ``"1"``.
+
+        Returns:
+            dict: API response containing ``result.watchlist_id`` of the new item.
+
+        """
+        data = {"symbol": symbol, "sec_type": sec_type}
+        response = self.ft_session._request("post", url=urls.watchlist_item(list_id), data=data)
+        return response.json()
+
+    def remove_symbol(self, watchlist_id: int) -> dict:
+        """Remove a symbol from a watchlist by its watchlist item ID.
+
+        Note:
+            ``watchlist_id`` here refers to the per-item ID returned by
+            :meth:`add_symbol` (``result.watchlist_id``), **not** the
+            ``list_id`` of the watchlist itself.
+
+        Args:
+            watchlist_id (int): The item-level watchlist ID to remove.
+
+        Returns:
+            dict: API response confirming deletion via ``result.result == "success"``.
+
+        """
+        response = self.ft_session._request(
+            "delete", url=urls.watchlist_item_delete(watchlist_id)
+        )
+        return response.json()

test.py

@@ -2,7 +2,7 @@
 
 import json
 
-from firstrade import account, order, symbols
+from firstrade import account, order, symbols, watchlist
 
 # Create a session
 # mfa_secret is the secret key to generate TOTP (not the backup code), see:
@@ -160,5 +160,33 @@
 )
 print(f"Preview of an option order for {option_quote['items'][0]['opt_symbol']}: {json.dumps(option_order, indent=2)}")
 
+
+wl = watchlist.Watchlist(ft_ss)
+data = wl.get_watchlists()
+if data['statusCode'] != 200 or len(data['error']) > 0:
+    raise Exception("Error while fetching watchlists.")
+print("Watchlist(s):")
+print(*[f'#{i["list_id"]}: {i["name"]} (default={i["isDefault"]})' for i in data["items"]], sep="\n")
+
+result = wl.create_watchlist("My Watchlist")
+if result['statusCode'] != 200 or result['result']['result'] != 'success':
+    raise Exception("Cannot add a new watchlist.")
+print(f"Created a new watchlist with id: #{result['result']['list_id']}")
+
+list_id = result["result"]["list_id"]
+print(f"Adding symbol 'AAPL' to watchlist #{list_id}")
+data = wl.add_symbol(list_id, "AAPL")
+if data['statusCode'] != 200 or len(data['error']) > 0:
+    raise Exception("Error while adding a symbol to watchlist #{list_id}.")
+watchlist_content = wl.get_watchlist(list_id)
+if watchlist_content['statusCode'] != 200 or len(watchlist_content['error']) > 0:
+    raise Exception("Error while fetching content of watchlist #{list_id}.")
+watchlist_content = watchlist_content['result']
+print(f"Content of created watchlist ({watchlist_content['name']} #{watchlist_content['list_id']}): {json.dumps(watchlist_content['list_items'], indent=2)}")
+data = wl.delete_watchlist(list_id)
+if data['statusCode'] != 200 or len(data['error']) > 0:
+    raise Exception("Error while deleting a watchlist.")
+print(f"Deleted watchlist #{list_id}.")
+
 # Delete the session cookie
 # ft_ss.delete_cookies()