Implement endpoints for stocks snapshots

Author: pythonfortraders

cli/main.py

@@ -10,6 +10,16 @@
 
 Save output to a file:
    python main.py stocks historical eod-report AAPL 20240101 20240131 --output-file aapl_eod.csv
+
+Stocks Snapshot Data:
+Get real-time quotes:
+   python main.py stocks snapshot quotes AAPL
+
+Get real-time OHLC:
+   python main.py stocks snapshot ohlc NVDA
+
+Get real-time trades:
+   python main.py stocks snapshot trades TSLA
 """
 
 import typer
@@ -23,15 +33,21 @@
 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from src.stocks_historical import ThetaDataStocksHistorical
-from typing import Optional
+from src.stocks import ThetaDataStocksSnapshot
+from typing import Optional, List
 
 app = typer.Typer()
 stocks_app = typer.Typer()
+historical_app = typer.Typer()
+snapshot_app = typer.Typer()
 options_app = typer.Typer()
 app.add_typer(stocks_app, name="stocks")
+stocks_app.add_typer(historical_app, name="historical")
+stocks_app.add_typer(snapshot_app, name="snapshot")
 app.add_typer(options_app, name="options")
 
 historical_data = ThetaDataStocksHistorical(enable_logging=True, use_df=True)
+snapshot_data = ThetaDataStocksSnapshot(enable_logging=True, use_df=True)
 
 
 def save_output(result: pd.DataFrame | dict | None, output_file: Optional[str]):
@@ -61,7 +77,8 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
-@stocks_app.command(name="eod-report")
+# Historical commands
+@historical_app.command(name="eod-report")
 @with_spinner
 def eod_report(
     symbol: str, start_date: str, end_date: str, output_file: Optional[str] = None
@@ -71,45 +88,45 @@ def eod_report(
     save_output(result, output_file)
 
 
-@stocks_app.command(name="quotes")
+@historical_app.command(name="quotes")
 @with_spinner
-def quotes(
+def historical_quotes(
     symbol: str,
     start_date: str,
     end_date: str,
     interval: str = "900000",
     output_file: Optional[str] = None,
 ):
-    """Get quotes for a given symbol and date range."""
+    """Get historical quotes for a given symbol and date range."""
     result = historical_data.get_quotes(symbol, start_date, end_date, interval)
     save_output(result, output_file)
 
 
-@stocks_app.command(name="ohlc")
+@historical_app.command(name="ohlc")
 @with_spinner
-def ohlc(
+def historical_ohlc(
     symbol: str,
     start_date: str,
     end_date: str,
     interval: str = "900000",
     output_file: Optional[str] = None,
 ):
-    """Get OHLC data for a given symbol and date range."""
+    """Get historical OHLC data for a given symbol and date range."""
     result = historical_data.get_ohlc(symbol, start_date, end_date, interval)
     save_output(result, output_file)
 
 
-@stocks_app.command(name="trades")
+@historical_app.command(name="trades")
 @with_spinner
-def trades(
+def historical_trades(
     symbol: str, start_date: str, end_date: str, output_file: Optional[str] = None
 ):
     """Get historical trade data for a given symbol and date range."""
     result = historical_data.get_trades(symbol, start_date, end_date)
     save_output(result, output_file)
 
 
-@stocks_app.command(name="trade-quote")
+@historical_app.command(name="trade-quote")
 @with_spinner
 def trade_quote(
     symbol: str, start_date: str, end_date: str, output_file: Optional[str] = None
@@ -119,7 +136,7 @@ def trade_quote(
     save_output(result, output_file)
 
 
-@stocks_app.command(name="splits")
+@historical_app.command(name="splits")
 @with_spinner
 def splits(
     symbol: str, start_date: str, end_date: str, output_file: Optional[str] = None
@@ -129,7 +146,7 @@ def splits(
     save_output(result, output_file)
 
 
-@stocks_app.command(name="dividends")
+@historical_app.command(name="dividends")
 @with_spinner
 def dividends(
     symbol: str, start_date: str, end_date: str, output_file: Optional[str] = None
@@ -139,5 +156,50 @@ def dividends(
     save_output(result, output_file)
 
 
+# Snapshot commands
+@snapshot_app.command(name="quotes")
+@with_spinner
+def snapshot_quotes(
+    symbol: str, venue: Optional[str] = None, output_file: Optional[str] = None
+):
+    """Get real-time quotes for a given symbol."""
+    result = snapshot_data.get_quotes(symbol, venue)
+    save_output(result, output_file)
+
+
+@snapshot_app.command(name="bulk-quotes")
+@with_spinner
+def bulk_quotes(
+    symbols: List[str], venue: Optional[str] = None, output_file: Optional[str] = None
+):
+    """Get real-time quotes for multiple symbols."""
+    result = snapshot_data.get_bulk_quotes(symbols, venue)
+    save_output(result, output_file)
+
+
+@snapshot_app.command(name="ohlc")
+@with_spinner
+def snapshot_ohlc(symbol: str, output_file: Optional[str] = None):
+    """Get real-time OHLC data for a given symbol."""
+    result = snapshot_data.get_ohlc(symbol)
+    save_output(result, output_file)
+
+
+@snapshot_app.command(name="bulk-ohlc")
+@with_spinner
+def bulk_ohlc(symbols: List[str], output_file: Optional[str] = None):
+    """Get real-time OHLC data for multiple symbols."""
+    result = snapshot_data.get_bulk_ohlc(symbols)
+    save_output(result, output_file)
+
+
+@snapshot_app.command(name="trades")
+@with_spinner
+def snapshot_trades(symbol: str, output_file: Optional[str] = None):
+    """Get real-time trade data for a given symbol."""
+    result = snapshot_data.get_trades(symbol)
+    save_output(result, output_file)
+
+
 if __name__ == "__main__":
     app()

examples/get_historical_data.py examples/stocks_historical_examples.pyexamples/get_historical_data.py renamed to examples/stocks_historical_examples.py

@@ -0,0 +1,57 @@
+import sys
+import os
+from functools import wraps
+from stocks_historical_examples import example_runner
+
+# Add the parent directory to sys.path
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from src.stocks import ThetaDataStocksSnapshot
+
+snapshot_data = ThetaDataStocksSnapshot(enable_logging=True, use_df=True)
+
+
+@example_runner
+def apple_quotes_example():
+    return snapshot_data.get_quotes("AAPL")
+
+
+@example_runner
+def microsoft_quotes_nqb_example():
+    return snapshot_data.get_quotes("MSFT", venue="nqb")
+
+
+@example_runner
+def bulk_quotes_example():
+    return snapshot_data.get_bulk_quotes(["GOOGL", "AMZN", "TSLA"])
+
+
+@example_runner
+def nvidia_ohlc_example():
+    return snapshot_data.get_ohlc("NVDA")
+
+
+@example_runner
+def bulk_ohlc_example():
+    return snapshot_data.get_bulk_ohlc(["INTC", "AMD", "QCOM"])
+
+
+@example_runner
+def tesla_trades_example():
+    return snapshot_data.get_trades("TSLA")
+
+
+# Toggle example cases
+if __name__ == "__main__":
+    run_examples = {
+        "apple_quotes_example": True,
+        "microsoft_quotes_nqb_example": True,
+        "bulk_quotes_example": True,
+        "nvidia_ohlc_example": True,
+        "bulk_ohlc_example": True,
+        "tesla_trades_example": True,
+    }
+
+    for example, should_run in run_examples.items():
+        if should_run:
+            globals()[example]()