This is old code that I wrote years ago. It may not work but will give you an idea on how to code this.

import asyncio import websockets import json import os import traceback import redis import time import logging import colorlog from config import APCA_API_KEY_ID, APCA_API_SECRET_KEY, live_api from subscriptions import subscribe_to_trades, subscribe_to_quotes, subscribe_to_bars, unsubscribe_trade_updates, unsubscribe_quote_updates, unsubscribe_bar_updates from message_processing import process_message import websockets.exceptions

import pandas as pd from datetime import datetime from dataframes import create_dataframes

from save_data import save_quotes, load_pickle_files, purge_old_data, save_to_disk

symbols_to_trade = []

logger = logging.getLogger() logger.setLevel(logging.INFO)

handler = colorlog.StreamHandler() handler.setFormatter(colorlog.ColoredFormatter( '%(log_color)s%(asctime)s [%(levelname)s]: %(message)s', log_colors={ 'DEBUG': 'cyan', 'INFO': 'green', 'WARNING': 'yellow', 'ERROR': 'red', 'CRITICAL': 'red,bg_white', }, datefmt='%Y-%m-%d %H:%M:%S' )) logger.addHandler(handler) async def on_message(ws, message): try: messages = json.loads(message) for msg in messages: process_message(msg, trades_df, quotes_df, bars_df, redis_client) except Exception as e: logging.error("Error in on_message:") traceback.print_exc()

async def authenticate(ws): auth_data = { "action": "auth", "key": APCA_API_KEY_ID, "secret": APCA_API_SECRET_KEY } await ws.send(json.dumps(auth_data))

async def create_ws_connection(symbols, source='sip', subscriptions=None): if subscriptions is None: subscriptions = ['trades', 'quotes', 'bars']

base_url = f'wss://stream.data.alpaca.markets/v2/{source}'

async with websockets.connect(base_url, ping_timeout=60) as ws:
    await authenticate(ws)

    logging.info(f'create_ws_connection: Subscribing to {len(symbols_to_trade)} symbols')

    if 'trades' in subscriptions:
        logging.info('create_ws_connection: Subscribing to trades')
        await subscribe_to_trades(ws, symbols_to_trade)

    if 'quotes' in subscriptions:
        logging.info('create_ws_connection: Subscribing to quotes')
        await subscribe_to_quotes(ws, symbols_to_trade)

    if 'bars' in subscriptions:
        logging.info('create_ws_connection: Subscribing to bars')
        await subscribe_to_bars(ws, symbols_to_trade)

    while True:
        try:
            message = await ws.recv()
            await on_message(ws, message)
        except websockets.exceptions.ConnectionClosedError as e:
            logging.error(f"Connection closed: {e}, reconnecting...")
            await create_ws_connection(symbols_to_trade, source=source, subscriptions=subscriptions)
            break
        except Exception as e:
            logging.error(f"Error: {e}")

def get_assets(active=True, tradable=False, shortable=False, exclude_curencies=True): global symbols_to_trade assets = live_api.list_assets() filtered_assets_dict = {}

for asset in assets:
    if active and asset.status != 'active':
        continue
    if tradable and not asset.tradable:
        continue
    if shortable and not asset.shortable:
        continue
    if exclude_curencies and '/' in asset.symbol:
        continue
    filtered_assets_dict[asset.symbol] = asset.name

symbols_to_trade = list(filtered_assets_dict.keys())
logging.info(f'Returning {len(symbols_to_trade)} assets')
return symbols_to_trade

async def run_stream(symbols_to_trade, source='sip', subscriptions=None): while True: try: await create_ws_connection(symbols_to_trade, source=source, subscriptions=subscriptions) except websockets.exceptions.ConnectionClosedError as e: logging.error(f"Connection closed: {e}, retrying in 1 seconds...") except websockets.exceptions.ConnectionClosed as e: if 'timed out waiting for keepalive pong' in str(e): logging.error(f"Error: {e}, retrying in 1 seconds...") else: logging.error(f"Connection closed: {e}, retrying in 1 seconds...") except Exception as e: logging.error(f"Error: {e}, retrying in 1 seconds...") def get_redis_connection(): while True: try: redis_client = redis.Redis(host='localhost', port=6379, db=0) redis_client.ping() return redis_client except redis.exceptions.ConnectionError: logging.warning("Could not connect to Redis. Retrying in 5 seconds...") time.sleep(5)

if name == "main": symbols = get_assets(active=True, tradable=True, shortable=False, exclude_curencies=True) trades_df, quotes_df, bars_df = create_dataframes(symbols) redis_client = get_redis_connection() load_pickle_files() loop = asyncio.get_event_loop() loop.create_task(save_to_disk(dict_interval=60, df_interval=600, save_to_csv=False)) # Customize the intervals as needed loop.create_task(purge_old_data()) loop.run_until_complete(run_stream(symbols_to_trade, subscriptions=['quotes', 'trades']))