From 9bffd9c3e36650833ef790ed6bc089511e0dda58 Mon Sep 17 00:00:00 2001
From: camelpac <78740792+camelpac@users.noreply.github.com>
Date: Sat, 23 Oct 2021 19:10:58 +0300
Subject: [PATCH] update to alpaca data v2
---
README.md | 33 ++-
algo.py | 588 +++++++++++++++++++++++++++++------------------
config.yaml | 4 +
requirements.txt | 24 +-
4 files changed, 414 insertions(+), 235 deletions(-)
create mode 100644 config.yaml
diff --git a/README.md b/README.md
index 04ba531..e35c512 100644
--- a/README.md
+++ b/README.md
@@ -1,11 +1,38 @@
# Momentum-Trading-Example
-An example algorithm for a momentum-based day trading strategy. This script uses the API provided by [Alpaca](https://alpaca.markets/). A brokerage account with Alpaca, available to US customers, is required to access the [Polygon](https://polygon.io/) data stream used by this algorithm.
+An example algorithm for a momentum-based day trading strategy. This script uses
+the API provided by [Alpaca](https://alpaca.markets/). A brokerage account with
+Alpaca, is required. to access the
+[Polygon](https://polygon.io/) data stream used by this algorithm.
## Running the Script
+You must supply your Alpaca credentials to run this script.
+You do this by filling the supplied `config.yaml` file next to this script.
+You need to fill your key and secret => get it from the Alpaca Dashboard.
+Then, for paper trading leave `base_url` as is. For live trading this needs to be changed to `'https://api.alpaca.markets'`
+Lastly, select the data stream feed type:
+ * iex for free accounts
+ * sip for paid accounts
-Note that near the top of the file, there are placeholders for your API information - your key ID, your secret key, and the URL you want to connect to. You can get all that information from the Alpaca dashboard. Replace the placeholder strings with your own information, and the script is ready to run with `python algo.py`. **Please note that running with Python 3.6 is required.**
+Now the script is ready
+to run with `python algo.py`.
+**Please note that running with Python 3.6+ is required.**
## Algorithm Logic
-This algorithm may buy stocks during a 45 minute period each day, starting 15 minutes after market open. (The first 15 minutes are avoided, as the high volatility can lead to poor performance.) The signals it uses to determine if it should buy a stock are if it has gained 4% from the previous day's close, is above the highest point seen during the first 15 minutes of trading, and if its MACD is positive and increasing. (It also checks that the volume is strong enough to make trades on reliably.) It sells when a stock drops to a stop loss level or increases to a target price level. If there are open positions in your account at the end of the day on a symbol the script is watching, those positions will be liquidated at market. (Potentially holding positions overnight goes against the concept of risk that the algorithm uses, and must be avoided for it to be effective.)
+This algorithm may buy stocks during a 45 minute period each day, starting 15
+minutes after market open.
The first 15 minutes are avoided, as the high
+volatility can lead to poor performance.
The signals it uses to determine
+if it should buy a stock are:
+ * if a stock has gained 4% from the previous day's close
+ * and is above the highest point seen during the first 15 minutes of trading
+ * and if its MACD is positive and increasing.
+ (It also checks that the volume is strong
+enough to make trades on reliably.)
+
+It sells when a stock drops to a stop loss level or increases to a target price level.
+If there are open positions in your account at the end of the day on a symbol the script is watching, those positions
+will be liquidated at market price.
+(Potentially holding positions overnight goes against
+the concept of risk that the algorithm uses, and must be avoided for it to be
+effective.)
diff --git a/algo.py b/algo.py
index b52342a..8a89661 100644
--- a/algo.py
+++ b/algo.py
@@ -1,21 +1,44 @@
+import logging
+import asyncio
+
+import pytz
+import sys
+import pandas as pd
import alpaca_trade_api as tradeapi
import requests
import time
-from ta import macd
+
+import yaml
+from alpaca_trade_api.common import URL
+from alpaca_trade_api.entity import Order
+from alpaca_trade_api.rest import TimeFrame
+from alpaca_trade_api.rest_async import gather_with_concurrency
+from ta.trend import macd
import numpy as np
from datetime import datetime, timedelta
from pytz import timezone
+from loguru import logger
+from concurrent.futures import ThreadPoolExecutor
+
# Replace these with your API connection info from the dashboard
-base_url = 'Your API URL'
-api_key_id = 'Your API Key'
-api_secret = 'Your API Secret'
+with open("./config.yaml", mode='r') as f:
+ o = yaml.safe_load(f)
+ api_key_id = o.get("key_id")
+ api_secret = o.get("secret")
+ base_url = o.get("base_url")
+ feed = o.get("feed")
api = tradeapi.REST(
base_url=base_url,
key_id=api_key_id,
secret_key=api_secret
)
+api_async = tradeapi.AsyncRest(
+ key_id=api_key_id,
+ secret_key=api_secret
+)
+conn: tradeapi.Stream = None
session = requests.session()
@@ -24,44 +47,82 @@
max_share_price = 13.0
# Minimum previous-day dollar volume for a stock we might consider
min_last_dv = 500000
+daily_pct_change = 3.5
# Stop limit to default to
default_stop = .95
# How much of our portfolio to allocate to any one position
risk = 0.001
+if float(api.get_account().cash) < 0:
+ # we don't have money left, we'll wait until we liquidate some stocks
+ WAIT_FOR_LIQUIDATION = True
+else:
+ WAIT_FOR_LIQUIDATION = False
+
+
+async def get_historic_bars(symbols, start, end, timeframe: TimeFrame):
+ major = sys.version_info.major
+ minor = sys.version_info.minor
+ if major < 3 or minor < 6:
+ raise Exception('asyncio is not support in your python version')
+ msg = f"Getting Bars data for {len(symbols)} symbols"
+ msg += f", timeframe: {timeframe}" if timeframe else ""
+ msg += f" between dates: start={start}, end={end}"
+ logger.info(msg)
+
+ tasks = []
-def get_1000m_history_data(symbols):
- print('Getting historical data...')
- minute_history = {}
- c = 0
for symbol in symbols:
- minute_history[symbol] = api.polygon.historic_agg(
- size="minute", symbol=symbol, limit=1000
- ).df
- c += 1
- print('{}/{}'.format(c, len(symbols)))
- print('Success.')
+ args = [symbol, start, end, timeframe.value, 3000]
+ tasks.append(api_async.get_bars_async(*args))
+ start_time = time.time()
+ if minor >= 8:
+ results = await asyncio.gather(*tasks, return_exceptions=True)
+ else:
+ results = await gather_with_concurrency(500, *tasks)
+
+ logger.info(f"Total of {len(results)} Bars. It took "
+ f"{time.time() - start_time} seconds")
+ for symbol, df in results:
+ df.index = df.index.tz_convert(pytz.timezone('America/New_York'))
+
+ return {symbol: df.iloc[-1000:] for symbol, df in results}
+
+
+def get_1000m_history_data(symbols, start, end):
+ logger.info('Getting historical data...')
+ timeframe: TimeFrame = TimeFrame.Minute
+ loop = asyncio.get_event_loop()
+ minute_history = loop.run_until_complete(
+ get_historic_bars(symbols,
+ start.isoformat(),
+ end.isoformat(),
+ timeframe))
+
return minute_history
def get_tickers():
- print('Getting current ticker data...')
- tickers = api.polygon.all_tickers()
- print('Success.')
- assets = api.list_assets()
- symbols = [asset.symbol for asset in assets if asset.tradable]
- return [ticker for ticker in tickers if (
- ticker.ticker in symbols and
- ticker.lastTrade['p'] >= min_share_price and
- ticker.lastTrade['p'] <= max_share_price and
- ticker.prevDay['v'] * ticker.lastTrade['p'] > min_last_dv and
- ticker.todaysChangePerc >= 3.5
- )]
+ assets = api.list_assets(status="active")
+
+ snapshot = api.get_snapshots([el.symbol for el in assets])
+ tickers = {}
+ for symbol, data in snapshot.items():
+ try:
+ if data.latest_trade.p >= min_share_price:
+ if data.latest_trade.p <= max_share_price:
+ if data.daily_bar.v * data.latest_trade.p > min_last_dv:
+ if (data.daily_bar.c - data.prev_daily_bar.c) / \
+ data.prev_daily_bar.c * 100 >= daily_pct_change:
+ tickers[symbol] = data
+ except:
+ logger.warning(f"can't get data for: {symbol}")
+ return tickers
def find_stop(current_value, minute_history, now):
series = minute_history['low'][-100:] \
- .dropna().resample('5min').min()
+ .dropna().resample('5min').min()
series = series[now.floor('1D'):]
diff = np.diff(series.values)
low_index = np.where((diff[:-1] <= 0) & (diff[1:] > 0))[0] + 1
@@ -70,21 +131,29 @@ def find_stop(current_value, minute_history, now):
return current_value * default_stop
+
def run(tickers, market_open_dt, market_close_dt):
# Establish streaming connection
- conn = tradeapi.StreamConn(base_url=base_url, key_id=api_key_id, secret_key=api_secret)
+ feed = 'iex' # <- replace to sip if you have PRO subscription
+ # feed = 'sip' # <- replace to sip if you have PRO subscription
+ global conn
+ conn = tradeapi.Stream(key_id=api_key_id,
+ secret_key=api_secret,
+ base_url=URL('https://paper-api.alpaca.markets'),
+ data_feed=feed)
# Update initial state with information from tickers
volume_today = {}
prev_closes = {}
- for ticker in tickers:
- symbol = ticker.ticker
- prev_closes[symbol] = ticker.prevDay['c']
- volume_today[symbol] = ticker.day['v']
+ for symbol, data in tickers.items():
+ prev_closes[symbol] = data.prev_daily_bar.c
+ volume_today[symbol] = data.daily_bar.v
- symbols = [ticker.ticker for ticker in tickers]
- print('Tracking {} symbols.'.format(len(symbols)))
- minute_history = get_1000m_history_data(symbols)
+ symbols = tickers.keys()
+ logger.info('Tracking {} symbols.'.format(len(symbols)))
+ end_date = list(tickers.values())[0].daily_bar.t + timedelta(days=1)
+ start_date = end_date - timedelta(days=5)
+ minute_history = get_1000m_history_data(symbols, start_date, end_date)
portfolio_value = float(api.get_account().portfolio_value)
@@ -95,6 +164,7 @@ def run(tickers, market_open_dt, market_close_dt):
existing_orders = api.list_orders(limit=500)
for order in existing_orders:
if order.symbol in symbols:
+ logger.info(f"Cancelling open order for {order.symbol} on startup")
api.cancel_order(order.id)
stop_prices = {}
@@ -108,7 +178,7 @@ def run(tickers, market_open_dt, market_close_dt):
# Recalculate cost basis and stop price
latest_cost_basis[position.symbol] = float(position.cost_basis)
stop_prices[position.symbol] = (
- float(position.cost_basis) * default_stop
+ float(position.cost_basis) * default_stop
)
# Keep track of what we're buying/selling
@@ -116,8 +186,7 @@ def run(tickers, market_open_dt, market_close_dt):
partial_fills = {}
# Use trade updates to keep track of our portfolio
- @conn.on(r'trade_update')
- async def handle_trade_update(conn, channel, data):
+ async def handle_trade_update(data):
symbol = data.order['symbol']
last_order = open_orders.get(symbol)
if last_order is not None:
@@ -127,17 +196,17 @@ async def handle_trade_update(conn, channel, data):
if data.order['side'] == 'sell':
qty = qty * -1
positions[symbol] = (
- positions.get(symbol, 0) - partial_fills.get(symbol, 0)
+ positions.get(symbol, 0) - partial_fills.get(symbol, 0)
)
partial_fills[symbol] = qty
positions[symbol] += qty
- open_orders[symbol] = data.order
+ open_orders[symbol] = Order(data.order)
elif event == 'fill':
qty = int(data.order['filled_qty'])
if data.order['side'] == 'sell':
qty = qty * -1
positions[symbol] = (
- positions.get(symbol, 0) - partial_fills.get(symbol, 0)
+ positions.get(symbol, 0) - partial_fills.get(symbol, 0)
)
partial_fills[symbol] = 0
positions[symbol] += qty
@@ -146,229 +215,290 @@ async def handle_trade_update(conn, channel, data):
partial_fills[symbol] = 0
open_orders[symbol] = None
- @conn.on(r'A$')
- async def handle_second_bar(conn, channel, data):
- symbol = data.symbol
+ conn.subscribe_trade_updates(handle_trade_update)
- # First, aggregate 1s bars for up-to-date MACD calculations
- ts = data.start
- ts -= timedelta(seconds=ts.second, microseconds=ts.microsecond)
+ async def handle_second_bar(data):
try:
- current = minute_history[data.symbol].loc[ts]
- except KeyError:
- current = None
- new_data = []
- if current is None:
- new_data = [
- data.open,
- data.high,
- data.low,
- data.close,
- data.volume
- ]
- else:
- new_data = [
- current.open,
- data.high if data.high > current.high else current.high,
- data.low if data.low < current.low else current.low,
- data.close,
- current.volume + data.volume
- ]
- minute_history[symbol].loc[ts] = new_data
-
- # Next, check for existing orders for the stock
- existing_order = open_orders.get(symbol)
- if existing_order is not None:
- # Make sure the order's not too old
- submission_ts = existing_order.submitted_at.astimezone(
- timezone('America/New_York')
- )
- order_lifetime = ts - submission_ts
- if order_lifetime.seconds // 60 > 1:
- # Cancel it so we can try again for a fill
- api.cancel_order(existing_order.id)
- return
-
- # Now we check to see if it might be time to buy or sell
- since_market_open = ts - market_open_dt
- until_market_close = market_close_dt - ts
- if (
- since_market_open.seconds // 60 > 15 and
- since_market_open.seconds // 60 < 60
- ):
- # Check for buy signals
-
- # See if we've already bought in first
- position = positions.get(symbol, 0)
- if position > 0:
+ global WAIT_FOR_LIQUIDATION
+
+ symbol = data.symbol
+ if symbol not in symbols:
return
- # See how high the price went during the first 15 minutes
- lbound = market_open_dt
- ubound = lbound + timedelta(minutes=15)
- high_15m = 0
+ # First, aggregate 1s bars for up-to-date MACD calculations
+ ts = data.timestamp.replace(second=0, microsecond=0, nanosecond=0)
+ # ts -= timedelta(minutes=1)
try:
- high_15m = minute_history[symbol][lbound:ubound]['high'].max()
- except Exception as e:
- # Because we're aggregating on the fly, sometimes the datetime
- # index can get messy until it's healed by the minute bars
+ current = minute_history[data.symbol].loc[ts]
+ except KeyError:
+ current = None
+ if current is None:
+ new_data = [
+ data.bid_price, # open
+ data.ask_price, # high
+ data.bid_price, # low
+ data.ask_price, # close
+ data.ask_size + data.bid_size, # volume,
+ 0, # trade_count
+ 0, # vwap
+ ]
+ else:
+ new_data = [
+ current.open,
+ data.ask_price if data.ask_price > current.high else current.high,
+ data.bid_price if data.bid_price < current.low else current.low,
+ data.ask_price,
+ current.volume + data.ask_size + data.bid_size,
+ 0,
+ 0,
+ ]
+ minute_history[symbol].loc[ts] = new_data
+
+ # Next, check for existing orders for the stock
+ existing_order = open_orders.get(symbol)
+ if existing_order is not None:
+ # Make sure the order's not too old
+ if isinstance(existing_order, dict):
+ print("sdf")
+ existing_order = Order(existing_order)
+ submission_ts = existing_order.submitted_at.astimezone(
+ timezone('America/New_York')
+ )
+ order_lifetime = abs(ts - submission_ts)
+ if order_lifetime.seconds // 60 > 60:
+ # Cancel it so we can try again for a fill
+ logger.info(
+ f"Cancelling order for {existing_order.symbol} due to not filling in time")
+ api.cancel_order(existing_order.id)
return
- # Get the change since yesterday's market close
- daily_pct_change = (
- (data.close - prev_closes[symbol]) / prev_closes[symbol]
- )
+ # Now we check to see if it might be time to buy or sell
+ since_market_open = ts - market_open_dt
+ until_market_close = market_close_dt - ts
+ if (
+ since_market_open.seconds // 60 > 15 and
+ since_market_open.seconds // 60 < 60
+ ):
+
+ # Check for buy signals
+ if not WAIT_FOR_LIQUIDATION:
+
+ # See if we've already bought in first
+ position = positions.get(symbol, 0)
+ if position > 0:
+ return
+
+ # See how high the price went during the first 15 minutes
+ lbound = market_open_dt
+ ubound = lbound + timedelta(minutes=15)
+ high_15m = 0
+ try:
+ high_15m = minute_history[symbol][lbound:ubound][
+ 'high'].max()
+ except Exception as e:
+ # Because we're aggregating on the fly, sometimes the datetime
+ # index can get messy until it's healed by the minute bars
+ return
+
+ # Get the change since yesterday's market close
+ daily_pct_change = (data.ask_price - prev_closes[symbol]) / \
+ prev_closes[symbol]
+
+ if (
+ daily_pct_change > .04 and
+ data.ask_price > high_15m and
+ volume_today[symbol] > 30000
+ ):
+ # check for a positive, increasing MACD
+ hist = macd(
+ minute_history[symbol]['close'].dropna(),
+ window_fast=12,
+ window_slow=26
+ )
+ if (
+ hist[-1] < 0 or
+ not (hist[-3] < hist[-2] < hist[-1])
+ ):
+ return
+ hist = macd(
+ minute_history[symbol]['close'].dropna(),
+ window_fast=40,
+ window_slow=60
+ )
+ if hist[-1] < 0 or np.diff(hist)[-1] < 0:
+ return
+
+ # Stock has passed all checks; figure out how much to buy
+ stop_price = find_stop(
+ data.ask_price, minute_history[symbol], ts
+ )
+ stop_prices[symbol] = stop_price
+ target_prices[symbol] = data.ask_price + (
+ (data.ask_price - stop_price) * 3
+ )
+ shares_to_buy = portfolio_value * risk // (
+ data.ask_price - stop_price
+ )
+ if shares_to_buy == 0:
+ shares_to_buy = 1
+ shares_to_buy -= positions.get(symbol, 0)
+ if shares_to_buy <= 0:
+ return
+ if shares_to_buy * data.ask_price > float(
+ api.get_account().cash):
+ # logger.debug(f"not enough cash to buy {symbol}.")
+ return
+ logger.info(
+ 'Submitting buy for {} shares of {} at {}'.format(
+ shares_to_buy, symbol, data.ask_price
+ ))
+ try:
+ o = api.submit_order(
+ symbol=symbol, qty=str(shares_to_buy),
+ side='buy',
+ type='limit', time_in_force='day',
+ limit_price=str(data.ask_price)
+ )
+ open_orders[symbol] = o
+ latest_cost_basis[symbol] = data.ask_price
+ if float(api.get_account().cash) < 0:
+ # if float(api.get_account().cash) > 0:
+ # we don't have money left, we'll wait until we liquidate some stocks
+ WAIT_FOR_LIQUIDATION = True
+ except Exception as e:
+ logger.error(e)
+ return
if (
- daily_pct_change > .04 and
- data.close > high_15m and
- volume_today[symbol] > 30000
+ since_market_open.seconds // 60 >= 24 and
+ until_market_close.seconds // 60 > 15
):
- # check for a positive, increasing MACD
+ # Check for liquidation signals
+
+ # We can't liquidate if there's no position
+ position = positions.get(symbol, 0)
+ if position == 0:
+ return
+
+ # Sell for a loss if it's fallen below our stop price
+ # Sell for a loss if it's below our cost basis and MACD < 0
+ # Sell for a profit if it's above our target price
hist = macd(
minute_history[symbol]['close'].dropna(),
- n_fast=12,
- n_slow=26
+ window_fast=13,
+ window_slow=21
)
if (
- hist[-1] < 0 or
- not (hist[-3] < hist[-2] < hist[-1])
+ data.ask_price <= stop_prices[symbol] or
+ (data.ask_price >= target_prices[symbol] and hist[
+ -1] <= 0) or
+ (data.ask_price <= latest_cost_basis[symbol] and hist[
+ -1] <= 0)
):
- return
- hist = macd(
- minute_history[symbol]['close'].dropna(),
- n_fast=40,
- n_slow=60
- )
- if hist[-1] < 0 or np.diff(hist)[-1] < 0:
- return
-
- # Stock has passed all checks; figure out how much to buy
- stop_price = find_stop(
- data.close, minute_history[symbol], ts
- )
- stop_prices[symbol] = stop_price
- target_prices[symbol] = data.close + (
- (data.close - stop_price) * 3
- )
- shares_to_buy = portfolio_value * risk // (
- data.close - stop_price
- )
- if shares_to_buy == 0:
- shares_to_buy = 1
- shares_to_buy -= positions.get(symbol, 0)
- if shares_to_buy <= 0:
- return
-
- print('Submitting buy for {} shares of {} at {}'.format(
- shares_to_buy, symbol, data.close
- ))
- try:
- o = api.submit_order(
- symbol=symbol, qty=str(shares_to_buy), side='buy',
- type='limit', time_in_force='day',
- limit_price=str(data.close)
- )
- open_orders[symbol] = o
- latest_cost_basis[symbol] = data.close
- except Exception as e:
- print(e)
+ logger.info(
+ 'Submitting sell for {} shares of {} at {}'.format(
+ position, symbol, data.ask_price
+ ))
+ try:
+ o = api.submit_order(
+ symbol=symbol, qty=str(position), side='sell',
+ type='market', time_in_force='day'
+ )
+ WAIT_FOR_LIQUIDATION = False
+ open_orders[symbol] = o
+ latest_cost_basis[symbol] = data.ask_price
+ except Exception as e:
+ logger.error(e)
return
- if(
- since_market_open.seconds // 60 >= 24 and
- until_market_close.seconds // 60 > 15
- ):
- # Check for liquidation signals
-
- # We can't liquidate if there's no position
- position = positions.get(symbol, 0)
- if position == 0:
- return
-
- # Sell for a loss if it's fallen below our stop price
- # Sell for a loss if it's below our cost basis and MACD < 0
- # Sell for a profit if it's above our target price
- hist = macd(
- minute_history[symbol]['close'].dropna(),
- n_fast=13,
- n_slow=21
- )
- if (
- data.close <= stop_prices[symbol] or
- (data.close >= target_prices[symbol] and hist[-1] <= 0) or
- (data.close <= latest_cost_basis[symbol] and hist[-1] <= 0)
+ elif (
+ until_market_close.seconds // 60 <= 15
):
- print('Submitting sell for {} shares of {} at {}'.format(
- position, symbol, data.close
- ))
+ # Liquidate remaining positions on watched symbols at market
try:
- o = api.submit_order(
- symbol=symbol, qty=str(position), side='sell',
- type='limit', time_in_force='day',
- limit_price=str(data.close)
- )
- open_orders[symbol] = o
- latest_cost_basis[symbol] = data.close
+ position = api.get_position(symbol)
except Exception as e:
- print(e)
- return
- elif (
- until_market_close.seconds // 60 <= 15
- ):
- # Liquidate remaining positions on watched symbols at market
- try:
- position = api.get_position(symbol)
- except Exception as e:
- # Exception here indicates that we have no position
- return
- print('Trading over, liquidating remaining position in {}'.format(
- symbol)
- )
- api.submit_order(
- symbol=symbol, qty=position.qty, side='sell',
- type='market', time_in_force='day'
- )
- symbols.remove(symbol)
- if len(symbols) <= 0:
- conn.close()
- conn.deregister([
- 'A.{}'.format(symbol),
- 'AM.{}'.format(symbol)
- ])
+ # Exception here indicates that we have no position
+ return
+ logger.info(
+ 'Trading over, liquidating remaining position in {}'.format(
+ symbol)
+ )
+ api.submit_order(
+ symbol=symbol, qty=position.qty, side='sell',
+ type='market', time_in_force='day'
+ )
+ WAIT_FOR_LIQUIDATION = False
+ symbols.remove(symbol)
+ if len(symbols) <= 0:
+ conn.stop_ws()
+ conn.deregister([
+ 'A.{}'.format(symbol),
+ 'AM.{}'.format(symbol)
+ ])
+ except Exception as e:
+ print(e)
+
+ # conn.subscribe_quotes(handle_second_bar, *symbols)
+ conn.subscribe_quotes(handle_second_bar, "*")
# Replace aggregated 1s bars with incoming 1m bars
- @conn.on(r'AM$')
- async def handle_minute_bar(conn, channel, data):
- ts = data.start
- ts -= timedelta(microseconds=ts.microsecond)
+ async def handle_minute_bar(data):
+ ts = pd.Timestamp(data.timestamp).tz_localize(pytz.utc).tz_convert(
+ pytz.timezone('America/New_York'))
minute_history[data.symbol].loc[ts] = [
data.open,
data.high,
data.low,
data.close,
- data.volume
+ data.volume,
+ data.trade_count,
+ data.vwap
]
volume_today[data.symbol] += data.volume
- channels = ['trade_updates']
- for symbol in symbols:
- symbol_channels = ['A.{}'.format(symbol), 'AM.{}'.format(symbol)]
- channels += symbol_channels
- print('Watching {} symbols.'.format(len(symbols)))
- run_ws(conn, channels)
+ conn.subscribe_bars(handle_minute_bar, *symbols)
+
+ logger.info('Watching {} symbols.'.format(len(symbols)))
+ watchdog()
-# Handle failed websocket connections by reconnecting
-def run_ws(conn, channels):
+
+def conn_thread():
+ global conn
try:
- conn.run(channels)
- except Exception as e:
- print(e)
- conn.close()
- run_ws(conn, channels)
+ # make sure we have an event loop, if not create a new one
+ asyncio.get_event_loop()
+ except RuntimeError:
+ asyncio.set_event_loop(asyncio.new_event_loop())
+ conn.run()
+
+
+def watchdog():
+ global conn
+ pool = ThreadPoolExecutor(1)
+
+ # time.sleep(5) # give the ws time to open
+ while True:
+ if api.get_clock().is_open:
+ if not conn.is_open():
+ logger.info("Market is open, restarting websocket connection")
+ pool.submit(conn_thread)
+ else:
+ if conn:
+ if conn.is_open():
+ logger.info('waiting for market open')
+ loop = asyncio.get_event_loop()
+ loop.run_until_complete(conn.stop_ws())
+ else:
+ time.sleep(30)
+ time.sleep(15)
if __name__ == "__main__":
# Get when the market opens or opened today
+ alpaca_logger = logging.getLogger('alpaca_trade_api')
+ alpaca_logger.setLevel(logging.INFO)
+ alpaca_logger.addHandler(logging.StreamHandler())
+
nyc = timezone('America/New_York')
today = datetime.today().astimezone(nyc)
today_str = datetime.today().astimezone(nyc).strftime('%Y-%m-%d')
diff --git a/config.yaml b/config.yaml
new file mode 100644
index 0000000..0c668e2
--- /dev/null
+++ b/config.yaml
@@ -0,0 +1,4 @@
+key_id: ""
+secret: ""
+feed: iex
+base_url: https://paper-api.alpaca.markets
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index 9497206..98662b7 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,3 +1,21 @@
-alpaca-trade-api>=0.25
-ta
-sklearn
+alpaca-trade-api>=1.4.1
+certifi==2021.5.30
+charset-normalizer==2.0.4
+idna==3.2
+joblib==1.0.1
+msgpack==1.0.2
+numpy==1.21.1
+pandas==1.3.1
+python-dateutil==2.8.2
+pytz==2021.1
+requests==2.26.0
+scikit-learn==0.24.2
+scipy==1.7.1
+six==1.16.0
+ta==0.7.0
+threadpoolctl==2.2.0
+urllib3==1.26.6
+websocket-client==1.1.1
+websockets==9.1
+pyyaml==5.4.1
+loguru==0.5.3
\ No newline at end of file