PythonAnywhere to Run a Simple Trading Algorithm
Using PythonAnywhere When Running an Algorithm Live in the Cloud
After first getting into quant trading and writing algos on your local machine, the biggest area of resistance for newbie quant traders is running an algorithm live in the cloud.
This is a code snippet for a sample trading algorithm that we use for this post.
There are many cloud services on the market, including AWS, Google Cloud, Heroku, Digital Ocean, and PythonAnywhere. This walkthrough focuses on setting up PythonAnywhere and running a sample algorithm in Python and the Alpaca Trade API.
Alpaca
PythonAnywhere has a free tier of accounts, but only with access to whitelisted URLs. To access the Alpaca endpoints, you will need a paid account ($5/month).
PythonAnywhere is a nice option because it gives you some manner of GUI for accessing consoles, files, notebooks, and even scheduling tasks that automatically restart if they go down (excellent for algo trading).
Let’s jump in with a fresh account.
Step 1.) Setting up PythonAnywhere.
Step A.) Make a new PythonAnywhere account.
https://www.pythonanywhere.com/
You’re welcomed with a dashboard like below.
Step B.) Install alpaca-trade-api.
Under “Consoles”, open a new Bash console.
Run the following:
pip3 install --user alpaca-trade-apiStep 2.) Set up algo.
Our sample algo for this walkthrough is a simple 5/20 period ema (exponential moving average) crossover on the 5-minute timeframe, checking every 1 minute for signals.
import alpaca_trade_api as tradeapi
import time
import datetime
from datetime import timedelta
from pytz import timezone
tz = timezone('EST')
api = tradeapi.REST('your key',
'your secret',
'https://paper-api.alpaca.markets')
import logging
logging.basicConfig(filename='./new_5min_ema.log', format='%(name)s - %(levelname)s - %(message)s')
logging.warning('{} logging started'.format(datetime.datetime.now().strftime("%x %X")))
def get_data_bars(symbols, rate, slow, fast):
data = api.get_barset(symbols, rate, limit=20).df
for x in symbols:
data.loc[:, (x, 'fast_ema')] = data[x]['close'].rolling(window=fast).mean()
data.loc[:, (x, 'slow_ema')] = data[x]['close'].rolling(window=slow).mean()
return data
def get_signal_bars(symbol_list, rate, ema_slow, ema_fast):
data = get_data_bars(symbol_list, rate, ema_slow, ema_fast)
signals = {}
for x in symbol_list:
if data[x].iloc[-1]['fast_ema'] > data[x].iloc[-1]['slow_ema']: signal = 1
else: signal = 0
signals[x] = signal
return signals
def time_to_open(current_time):
if current_time.weekday() <= 4:
d = (current_time + timedelta(days=1)).date()
else:
days_to_mon = 0 - current_time.weekday() + 7
d = (current_time + timedelta(days=days_to_mon)).date()
next_day = datetime.datetime.combine(d, datetime.time(9, 30, tzinfo=tz))
seconds = (next_day - current_time).total_seconds()
return seconds
def run_checker(stocklist):
print('run_checker started')
while True:
# Check if Monday-Friday
if datetime.datetime.now(tz).weekday() >= 0 and datetime.datetime.now(tz).weekday() <= 4:
# Checks market is open
print('Trading day')
if datetime.datetime.now(tz).time() > datetime.time(9, 30) and datetime.datetime.now(tz).time() <= datetime.time(15, 30):
signals = get_signal_bars(stocklist, '5Min', 20, 5)
for signal in signals:
if signals[signal] == 1:
if signal not in [x.symbol for x in api.list_positions()]:
logging.warning('{} {} - {}'.format(datetime.datetime.now(tz).strftime("%x %X"), signal, signals[signal]))
api.submit_order(signal, 1, 'buy', 'market', 'day')
# print(datetime.datetime.now(tz).strftime("%x %X"), 'buying', signals[signal], signal)
else:
try:
api.submit_order(signal, 1, 'sell', 'market', 'day')
logging.warning('{} {} - {}'.format(datetime.datetime.now(tz).strftime("%x %X"), signal, signals[signal]))
except Exception as e:
# print('No sell', signal, e)
pass
time.sleep(60)
else:
# Get time amount until open, sleep that amount
print('Market closed ({})'.format(datetime.datetime.now(tz)))
print('Sleeping', round(time_to_open(datetime.datetime.now(tz))/60/60, 2), 'hours')
time.sleep(time_to_open(datetime.datetime.now(tz)))
else:
# If not trading day, find out how much until open, sleep that amount
print('Market closed ({})'.format(datetime.datetime.now(tz)))
print('Sleeping', round(time_to_open(datetime.datetime.now(tz))/60/60, 2), 'hours')
time.sleep(time_to_open(datetime.datetime.now(tz)))
stocks = ['AA','AAL','AAPL','AIG','AMAT','AMC','AMD','AMGN','AMZN','APA','BA','BABA','BAC','BBY','BIDU','BP','C','CAT','CMG',]
run_checker(stocks)Upload the algo as a file, or go to “Open another file” on the dashboard.
Enter it as /home/yourusername/algoname.py
You can run this right from the terminal editor, and see if it works. If you’re missing any dependencies, install them in the Bash console.
If the algo runs, you can leave it up and check results in the log file.
Step 3.) Set as task.
Go to “Tasks”, and go down to “Always On Tasks”.
You can schedule your algo to automatically restart if it crashes by entering “python3.6 /home/yourusername/algoname.py”.
This works especially well with our sample algo, which dynamically finds the amount of time to sleep until next market open.
Checking our Alpaca account, using the below as an example, we can see it’s generating signals about every minute:
Please note that this example is for illustrative purposes only. Past performance is not indicative of future results.
Technology and services are offered by AlpacaDB, Inc. Brokerage services are provided by Alpaca Securities LLC (alpaca.markets), member FINRA/SIPC. Alpaca Securities LLC is a wholly-owned subsidiary of AlpacaDB, Inc.
You can find us @AlpacaHQ, if you use twitter.
