added rich text to console output (#56)

Author: TexasCoding

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "py-alpaca-api"
-version = "2.1.4"
+version = "2.1.5"
 description = "Python package, for communicating with Alpaca Markets REST API."
 authors = ["TexasCoding <[email protected]>"]
 readme = "README.md"
@@ -18,6 +18,7 @@ tqdm = "^4.66.4"
 prophet = "^1.1.5"
 beautifulsoup4 = "^4.12.3"
 yfinance = "^0.2.40"
+rich = "^13.7.1"
 
 
 [tool.poetry.group.dev.dependencies]

pyproject.toml

@@ -1,4 +1,5 @@
 from typing import Dict
+from py_alpaca_api.stock.latest_quote import LatestQuote
 from py_alpaca_api.stock.predictor import Predictor
 from py_alpaca_api.trading.market import Market
 from py_alpaca_api.stock.assets import Assets
@@ -45,3 +46,4 @@ def _initialize_components(
             data_url=data_url, headers=headers, market=market, asset=self.assets
         )
         self.predictor = Predictor(history=self.history, screener=self.screener)
+        self.quotes = LatestQuote(headers=headers)

src/py_alpaca_api/stock/__init__.py

@@ -1,3 +1,45 @@
+import json
+from typing import Dict, List, Optional
+
+import pendulum
+
+from py_alpaca_api.http.requests import Requests
+
+
 class LatestQuote:
-    def __init__(self) -> None:
-        pass
+    def __init__(self, headers: Dict[str, str]) -> None:
+        self.headers = headers
+
+    def get(
+        self,
+        symbol: Optional[List[str] | str],
+        feed: str = "iex",
+        currency: str = "USD",
+    ) -> dict:
+        url = "https://data.alpaca.markets/v2/stocks/quotes/latest"
+
+        params = {"symbols": symbol, "feed": feed, "currency": currency}
+
+        response = json.loads(
+            Requests()
+            .request(method="GET", url=url, headers=self.headers, params=params)
+            .text
+        )
+
+        quotes = []
+
+        for key, value in response["quotes"].items():
+            quotes.append(
+                {
+                    "symbol": key,
+                    "timestamp": pendulum.parse(
+                        value["t"], tz="America/New_York"
+                    ).to_datetime_string(),
+                    "ask": value["ap"],
+                    "ask_size": value["as"],
+                    "bid": value["bp"],
+                    "bid_size": value["bs"],
+                }
+            )
+
+        print(quotes)

src/py_alpaca_api/stock/latest_quote.py

@@ -3,11 +3,35 @@
 import pandas as pd
 import pendulum
 from prophet import Prophet
-from tqdm import tqdm
+# from tqdm import tqdm
 
 from py_alpaca_api.stock.history import History
 from py_alpaca_api.stock.screener import Screener
 
+from rich.console import Console
+
+from rich.progress import (
+    BarColumn,
+    MofNCompleteColumn,
+    Progress,
+    TextColumn,
+    TimeElapsedColumn,
+    TimeRemainingColumn,
+)
+
+# Define custom progress bar
+progress_bar = Progress(
+    TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
+    BarColumn(),
+    MofNCompleteColumn(),
+    TextColumn("•"),
+    TimeElapsedColumn(),
+    TextColumn("•"),
+    TimeRemainingColumn(),
+)
+
+console = Console()
+
 yesterday = pendulum.now().subtract(days=1).format("YYYY-MM-DD")
 four_years_ago = pendulum.now().subtract(years=2).format("YYYY-MM-DD")
 
@@ -102,7 +126,7 @@ def get_losers_to_gainers(
         self,
         gain_ratio: float = 10.0,
         losers_to_scan: int = 200,
-        future_periods: int = 14,
+        future_periods: int = 5,
     ) -> list:
         """
         Predicts future gainers based on the previous day's losers using Prophet forecasting.
@@ -123,26 +147,37 @@ def get_losers_to_gainers(
 
         future_gainers = []
 
-        for i, ticker in tqdm(
-            enumerate(losers_list),
-            desc=f"• Predicting {len(losers_list)} future gainers with Prophet: ",
-        ):
-            try:
-                symbol_data = self.get_stock_data(ticker)
-                symbol_model = self.train_prophet_model(symbol_data)
-                symbol_forecast = self.generate_forecast(
-                    symbol_model, future_periods=future_periods
-                )
-                previous_price = previous_day_losers[
-                    previous_day_losers["symbol"] == ticker
-                ].iloc[0]["price"]
-                gain_prediction = round(
-                    ((symbol_forecast - previous_price) / previous_price) * 100, 2
-                )
-                if gain_prediction >= gain_ratio:
-                    future_gainers.append(ticker)
-            except Exception as e:
-                logger.error(f"Error predicting {ticker}: {e}")
-                continue
-
+        with progress_bar as progress:
+            # for i, ticker in tqdm(
+            #     enumerate(losers_list),
+            #     desc=f"• Predicting {len(losers_list)} future gainers with Prophet: ",
+            # ):
+            console.print(
+                f"Getting predictions for [bold]{len(losers_list)}[/bold] future gainers with Prophet: ",
+                style="green",
+            )
+            for i, ticker in progress.track(
+                enumerate(losers_list), total=len(losers_list)
+            ):
+                try:
+                    symbol_data = self.get_stock_data(ticker)
+                    symbol_model = self.train_prophet_model(symbol_data)
+                    symbol_forecast = self.generate_forecast(
+                        symbol_model, future_periods=future_periods
+                    )
+                    previous_price = previous_day_losers[
+                        previous_day_losers["symbol"] == ticker
+                    ].iloc[0]["price"]
+                    gain_prediction = round(
+                        ((symbol_forecast - previous_price) / previous_price) * 100, 2
+                    )
+                    if gain_prediction >= gain_ratio:
+                        future_gainers.append(ticker)
+                except Exception as e:
+                    logger.error(f"Error predicting {ticker}: {e}")
+                    continue
+        console.print(
+            f"Predicted [bold]{len(future_gainers)}[/bold] future gainers.",
+            style="yellow",
+        )
         return future_gainers

src/py_alpaca_api/stock/predictor.py

@@ -5,10 +5,14 @@
 import pandas as pd
 import pendulum
 
+from rich.console import Console
+
 from py_alpaca_api.http.requests import Requests
 from py_alpaca_api.trading.market import Market
 from py_alpaca_api.stock.assets import Assets
 
+console = Console()
+
 
 class Screener:
     def __init__(
@@ -111,14 +115,15 @@ def losers(
         Returns:
             pd.DataFrame: A filtered DataFrame containing stocks that meet the specified conditions for losers.
         """
-        return self.filter_stocks(
-            price_greater_than,
-            lambda df: df["change"] < change_less_than,
-            volume_greater_than,
-            trade_count_greater_than,
-            total_losers_returned,
-            ascending_order=True,
-        )
+        with console.status("[bold green]Getting previous day's losers..."):
+            return self.filter_stocks(
+                price_greater_than,
+                lambda df: df["change"] < change_less_than,
+                volume_greater_than,
+                trade_count_greater_than,
+                total_losers_returned,
+                ascending_order=True,
+            )
 
     ##################################################
     # //////////////// Get Gainers \\\\\\\\\\\\\\\\\ #
@@ -148,14 +153,15 @@ def gainers(
             pd.DataFrame: A Pandas DataFrame containing the stocks that satisfy the criteria for being gainers.
 
         """
-        return self.filter_stocks(
-            price_greater_than,
-            lambda df: df["change"] > change_greater_than,
-            volume_greater_than,
-            trade_count_greater_than,
-            total_gainers_returned,
-            ascending_order=False,
-        )
+        with console.status("[bold green]Getting previous day's biggest gainers..."):
+            return self.filter_stocks(
+                price_greater_than,
+                lambda df: df["change"] > change_greater_than,
+                volume_greater_than,
+                trade_count_greater_than,
+                total_gainers_returned,
+                ascending_order=False,
+            )
 
     ##################################################
     # /////////// Calculate Percentages \\\\\\\\\\\\ #

src/py_alpaca_api/stock/screener.py

@@ -8,6 +8,10 @@
 import yfinance as yf
 from py_alpaca_api.http.requests import Requests
 
+from rich.console import Console
+
+console = Console()
+
 logger = logging.getLogger("yfinance")
 logger.disabled = True
 logger.propagate = False
@@ -87,7 +91,7 @@ def truncate(text: str, length: int) -> str:
             else text
         )
 
-    def get_news(self, symbol: str, limit: int = 5) -> List[Dict[str, str]]:
+    def get_news(self, symbol: str, limit: int = 6) -> List[Dict[str, str]]:
         """
         Retrieves news articles related to a given symbol.
 
@@ -98,8 +102,10 @@ def get_news(self, symbol: str, limit: int = 5) -> List[Dict[str, str]]:
         Returns:
             list: A list of news articles, sorted by publish date in descending order.
         """
-        yahoo_news = self._get_yahoo_news(symbol=symbol, limit=limit)
-        benzinga_news = self._get_benzinga_news(symbol=symbol, limit=limit)
+        with console.status("[bold green]Getting latest news from Yahoo Finance..."):
+            yahoo_news = self._get_yahoo_news(symbol=symbol, limit=limit)
+        with console.status("[bold red]Getting latest news from Benzinga..."):
+            benzinga_news = self._get_benzinga_news(symbol=symbol, limit=limit)
 
         news = yahoo_news + benzinga_news