Stock Stream Unsubscribing

Alpaca Market Data
1

Does anyone know how to unsubscribe from a quote? My current code subscribes successfully but cant seem to unsubscribe properly.

import asyncio
from datetime import datetime
import pytz
from alpaca.data.live import StockDataStream
from alpaca.data.enums import DataFeed
async def test_subscribe_unsubscribe():
    stock_stream = StockDataStream(API_KEY, SECRET_KEY, feed=DataFeed.IEX)
    ticker = "NVDA"
    received_quotes = False

    async def handle_quote(quote):
        nonlocal received_quotes
        received_quotes = True
        eastern = pytz.timezone('US/Eastern')
        now = datetime.now(eastern)
        print(f"[{now}] Received quote for {ticker}: {quote.ask_price}")

    eastern = pytz.timezone('US/Eastern')
    print(f"[{datetime.now(eastern)}] Subscribing to stock stream for {ticker}")
    
    # Subscribe to quotes
    stock_stream.subscribe_quotes(handle_quote, ticker)
    
    # Start the stream
    stream_task = asyncio.create_task(stock_stream._run_forever())
    
    # Wait for some quotes to come in
    start_time = datetime.now(eastern)
    timeout = 5  # seconds
    
    while not received_quotes and (datetime.now(eastern) - start_time).total_seconds() < timeout:
        await asyncio.sleep(0.1)
        
    # Once we've received quotes or timed out, proceed with unsubscribe
    try:
        print(f"[{datetime.now(eastern)}] Attempting to unsubscribe from stock stream for {ticker}")
        
        # Attempt to unsubscribe
        await stock_stream.unsubscribe_quotes(ticker)
        await asyncio.sleep(1)  # Give some time for unsubscribe to process
        
        print(f"[{datetime.now(eastern)}] Unsubscribed successfully")
    except Exception as e:
        print(f"[{datetime.now(eastern)}] Error during unsubscribe: {e}")
    
    # Clean up by canceling the stream task directly
    print(f"[{datetime.now(eastern)}] Closing connection")
    stream_task.cancel()
    try:
        await stream_task
    except asyncio.CancelledError:
        print(f"[{datetime.now(eastern)}] Stream task cancelled")
    await stock_stream.close()
    print(f"[{datetime.now(eastern)}] Connection closed")

if __name__ == "__main__":
    asyncio.run(test_subscribe_unsubscribe())
2

Hope this helps.

You’re using stock_stream._run_forever(), which is an internal/private method (_run_forever). This can cause issues, especially with unsubscribing and closing connections cleanly.

Also, you’re calling: stock_stream.subscribe_quotes(handle_quote, ticker)
But this is not awaited, and the unsubscribe is awaited:
await stock_stream.unsubscribe_quotes(ticker)

This inconsistency in how you’re interacting with the stream can cause lifecycle bugs.

Try this updated script:

import asyncio
from datetime import datetime
import pytz
from alpaca.data.live import StockDataStream
from alpaca.data.enums import DataFeed

API_KEY = “YOUR_API_KEY”
SECRET_KEY = “YOUR_SECRET_KEY”

async def test_subscribe_unsubscribe():
stock_stream = StockDataStream(API_KEY, SECRET_KEY, feed=DataFeed.IEX)
ticker = “NVDA”
received_quotes = asyncio.Event()

async def handle_quote(quote):
    eastern = pytz.timezone('US/Eastern')
    now = datetime.now(eastern)
    print(f"[{now}] Received quote for {ticker}: {quote.ask_price}")
    received_quotes.set()

eastern = pytz.timezone('US/Eastern')
print(f"[{datetime.now(eastern)}] Subscribing to stock stream for {ticker}")

await stock_stream.subscribe_quotes_async(handle_quote, ticker)

# Start the stream (public method)
stream_task = asyncio.create_task(stock_stream.run())

try:
    # Wait for quote or timeout
    try:
        await asyncio.wait_for(received_quotes.wait(), timeout=5)
    except asyncio.TimeoutError:
        print(f"[{datetime.now(eastern)}] Timeout waiting for quotes.")

    print(f"[{datetime.now(eastern)}] Attempting to unsubscribe from {ticker}")
    await stock_stream.unsubscribe_quotes_async(ticker)
    await asyncio.sleep(1)
    print(f"[{datetime.now(eastern)}] Unsubscribed successfully")

finally:
    print(f"[{datetime.now(eastern)}] Closing connection")
    await stock_stream.stop()
    stream_task.cancel()
    try:
        await stream_task
    except asyncio.CancelledError:
        print(f"[{datetime.now(eastern)}] Stream task cancelled")
    print(f"[{datetime.now(eastern)}] Connection closed")

if name == “main”:
asyncio.run(test_subscribe_unsubscribe())