APCA
AP Acquisition Corp
stock NYSE

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']))

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']))