How to pluck data every 30s from a stream in python

Question

My Question: How can I pluck data every 30's from a stream in python. I am streaming intraday data from a stockbroker, Alpaca Api. The alpaca api provides hundreds of messages every few seconds but I only want one data point every 30 seconds (to make analysis faster).

My approach: My approach has been to put my data handler to sleep for 30 seconds and then when it wakes up a for range statement prevents it from collecting more than one data point before it goes back to sleep again.

Actual vs Expected Results: The message handler works well at recording only one message but instead of recording a message from the second in which it is awake, it is recording what appears to be the next message in the queue, even if I received that message 30 seconds ago and there have been 1000's of messages since then.

Possabilities: Maybe there is a way to access only the last item in a stream?

#Create data handler for stream 
def on__message(message):

    #for range ensures only one message is recoreded when function is awake
    for i in range(1):

        with open("C:/Users/micha/github/trade_bot/intraday_data.csv", "a") as outfile:

            #pull variables from message
            timestamp = q.timestamp
            bid_price = q.bid_price
            ask_price = q.ask_price
        
            #create writer object & point at outfile
            writer = csv.writer(outfile)

            #write/append variables to outfile
            writer.writerow([timestamp,bid_price, ask_price])

    #put recorder to sleep for 30s
    time.sleep(30)


#Create stream object 
stream = Stream(APCA_API_KEY_ID,
            APCA_API_SECRET_KEY,
            APCA_API_DATA_URL,
            data_feed='iex')  


#Subscribe to stream and point stream to my handler - 'on_message' 
stream.subscribe_quotes(on_message, 'AAPL')
stream.run()

user_14872851 · Answer

One approach is to only process messages if the elapsed time of the new and last processed message is greater than your specified "sampling interval". Otherwise, the stream handler should ignore the update.

Here is the example code:

from datetime import time, datetime

class SamplingStreamReader:

    def __init__(self, sampling_interval):
        self.sampling_interval = sampling_interval
        self.last_processed_timestamp = datetime.now()

    async def on_message(self, message):
        time_delta = datetime.now() - self.last_processed_timestamp
        if time_delta.total_seconds() > self.sampling_interval:
            print(f"{datetime.now()} - {message}")
            self.last_processed_timestamp = datetime.now()

if __name__ == '__main__':
    # Create stream object
    stream = Stream(APCA_API_KEY_ID,
                    APCA_API_SECRET_KEY,
                    APCA_API_DATA_URL,
                    data_feed='iex')

    # Subscribe to stream and point stream to my handler - 'on_message'
    stream_reader = SamplingStreamReader(30)
    stream.subscribe_quotes(stream_reader.on_message, 'AAPL')
    stream.run()

Other options include:

polling (eg making a request every n of seconds) instead of streaming the data
subscribing and unsubscribing from the stream every n seconds (which I don't recommend)

How to pluck data every 30s from a stream in python

1 Answers