Detecting Exponential Moving Average (EMA) crosses using Python

Repository With Sample Code

https://github.com/imwatsi/crypto-market-samples
Find the complete Python script on GitHub: ta_ema_cross.py

My Profile On GitHub

You can find code used in this tutorial as well as other tools and examples in the GitHub repositories under my profile:

https://github.com/imwatsi

What Will I Learn

• Calculate Exponential Moving Average using the bfxhfindicators
• Access previous EMA values using the EMA indicator's prev() function
• Use primary and secondary EMA periods to detect crosses in both directions

Requirements

• Python 3.6+
• Dependencies:
  • requests
  • bfxhfindicators
• An active internet connection

Difficulty

Intermediate

Definitions of Terms

Here are a few definitions of technical terms used in this tutorial:

• Exponential Moving Average (EMA): a type of moving average (MA) that places a greater weight and significance on the most recent data points
• EMA period: the number of data points used to calculate EMA, e.g. EMA(50) for 50-day moving average when using 1 day data points
• Candles: a representation of the price range for a particular period of time, in the format OHLC (open, high, low, close); sometimes volume is included to make OHLCV.

Tutorial Content

In this tutorial, we will use the bfxfhindicators technical analysis library to calculate EMA values for two periods (10 and 20) and then use a historical window of 10 periods to detect markets in which an EMA cross has occured. 10-period EMA will be the primary dataset and 20-period EMA will be the secondary dataset.

To get market data, we will use the HitBTC exchange API and extract 15 minute candle data for 20 USD symbols (BTCUSD, ETHUSD, etc...). Let's begin!

Install dependencies

Before we start writing code, let's make sure we have the necessary modules installed. Only two external dependencies need to be installed, the rest are already included as part of the standard Python library.

Install requests

On Linux, run the following command:

pip3 install requests

If you're on Windows or MacOS, the command is:

pip install requests

Install bfxhfindicators

For technical analysis, We will use bfxhfindicators, an open source module developed by Bitfinex. You can download the module from Bitfinex's GitHub Repository.

After downloading it, to install, you can either:

1. Extract the contents, "cd" into the directory on a terminal and run python3 setup.py install --user on Linux or just python setup.py install --user if on Windows or MacOS.

2. Extract the contents and copy the folder "bfxhfindicators" folder to your project's working directory.

Import the modules

The modules needed for our script to work can be imported like so:

import requests
import json
import time
from bfxhfindicators import EMA
from threading import Thread

• requests is used to access HitBTC's REST API to get market data

• json is used to parse JSON responses received from HitBTC

• time is used to create time delays

• bfxhfindicators contains the technical analysis library

• threading is used to perform tasks synchronous to speed requests up

Define variables and constants

After the imports, we can define a few global objects that we will need:

BASE_URL = 'https://api.hitbtc.com'

historic_window = 10
symbols = []
candles = {}
ema_values = {}
ema_periods = [10,20]
go_on = False

• historic_window: the number of historic EMA values to include when detecting EMA crosses

• symbols and candles store the market data (market symbols and 15m candles, respectively)

• ema_values stores the EMA values, by period and symbol

• ema_periods is where we set the primary and secondary EMA periods to use

• go_on boolean is used to create a delay till all candle data for all symbols is loaded

Function to get candles

Next, we define a function that will be used to extract candle data from HitBTC. Later on, we will write code to call it in a new thread for each symbol, to speed things up.

def import_candles(symbol):
    global candles
    # get candles
    resp = requests.get(BASE_URL + '/api/2/public/candles/%s?period=M15&limit=250'
                        %(symbol))
    raw_candles = json.loads(resp.content)
    # parse candles and save to memory
    parsed_candles = []
    for raw_c in raw_candles:
        new_candle = {
            'timestamp': raw_c['timestamp'],
            'close': float(raw_c['close']),
            'low': float(raw_c['min']),
            'high': float(raw_c['max'])
        }
        parsed_candles.append(new_candle)
    candles[symbol] = parsed_candles[:]

We declare candles as global to save data to the global variables we defined above. A request is made for 15min candles for the symbol in parameters, and the result is parsed and save in a format that's easy to work with.

Separate function to show progress

Next we define a function that will show us the progress of loading candles, for all symbols. It will also run in its own thread later on, checking the number of symbols loaded with the total, until all are complete.

def show_progress():
    global go_on
    #wait for symbols to load
    while True:
        time.sleep(0.2)
        print('Importing candles: %s/%s symbols loaded'
                %(len(candles), len(symbols)), end='\r')
        if len(candles) == len(symbols): # break when equal
            break
    go_on = True

Initial code

Next we write code that will execute first when the script is run. The first thing we need to do is get the 20 symbols from the exchange.

# get 20 USD symbols
print('Retrieving the first 20 USD symbols')
resp = requests.get(BASE_URL + '/api/2/public/symbol')
all_sym = json.loads(resp.content)
for x in all_sym:
    if 'USD' in x['id']:
        symbols.append(x['id'])
    if len(symbols) == 20:
        break
print('Found (%s) symbols.' %(len(symbols)))

A GET request is used to retrieve details of all symbols on the exchange, and a loop goes through each one, adding it to the list if it has USD as quote currency. A break in the loop is performed when the length of symbols is equal to 20: if len(symbols) == 20:.

Import candles for all symbols

Once we have the list of symbols, we can begin importing candle data for all of them. The code for this should create a new thread for each symbol, to make the process synchronous.

# import candles for each symbol
Thread(target=show_progress).start() # show progress
for sym in symbols:
    Thread(target=import_candles, args=(sym,)).start()

To stop the script from executing the next sections until all candle data has been loaded, we put a loop that waits for the boolean go_on to be True.

# wait until all candles are loaded
while go_on == False:
    time.sleep(1)
print('\nAll candles loaded.')

Calculate EMA values

The next step is to get the script to calculate EMA values for all symbols, in the two periods we want. Two for statements achieve this. One for the symbols and one for the EMA periods within each symbol.

# calculate EMA values
print('Calculating EMA values and scanning for crosses...', end='', flush=True)
for sym in symbols:
    for period in ema_periods:
        iEMA = EMA([period]) # define EMA object
        for candle in candles[sym]:
            iEMA.add(candle['close']) # add all close prices
        lst_ema = []
        lst_ema.append(iEMA.v()) # add current EMA value
        for i in range(historic_window):
            # add historic EMA values
            lst_ema.append(iEMA.prev(i+1))
        if sym not in ema_values: # add symbol key to dictionary
            ema_values[sym] = {}
        ema_values[sym][period] = lst_ema # save EMA values

All EMA values are saved in the ema_values variable, with symbol and EMA period as keys for the dictionary.

Identify EMA crosses

This is where all the logic needs to be perfect for this script to serve its purpose.

• For each market symbol, check to see what kind of cross we are looking for. By taking the oldest EMA values for both EMA periods and comparing them, we can determine whether we are looking for moments when the primary EMA crosses downwards to go below the secondary EMA, or vice versa.

• When the primary EMA is initially above the secondary, we are looking for a cross when the primary eventually reverses the situation and goes below the secondary, and vice versa for the other scenario.

• Scan all historic EMA values and filter out the symbols that meet the criteria for an EMA cross.

• Save the symbols in memory, according to the type of EMA cross detected.

# identify EMA crosses
ema_results = {
    'cross-downs': [],
    'cross-ups': []
}
for sym in symbols:
    # get primary and secondary EMA lists, and reverse for oldest first
    ema_first = ema_values[sym][ema_periods[0]][:]
    ema_second = ema_values[sym][ema_periods[1]][:]
    ema_first.reverse()
    ema_second.reverse()

    # determine type of cross to look for
    if ema_first[0] > ema_second[0]:
        look_for = 'cross-down'
    elif ema_first[0] < ema_second[0]:
        look_for = 'cross-up'

    # filter out symbols that meet criteria
    for i in range(1, historic_window + 1):
        if look_for == 'cross-down':
            if ema_first[i] < ema_second[i]:
                # primary EMA has gone below secondary
                tmp = ema_results['cross-downs']
                if sym not in tmp:
                    tmp.append(sym) # update list
                ema_results['cross-downs'] = tmp # save list
                del tmp
        elif look_for == 'cross-up':
            if ema_first[i] > ema_second[i]:
                # primary EMA has gone above secondary
                tmp = ema_results['cross-ups']
                if sym not in tmp:
                    tmp.append(sym) # update list
                ema_results['cross-ups'] = tmp # save list
                del tmp
print('done')

Print out the results

Once the process is complete for all symbols, it should then print out the results, categorized by type of EMA cross. To achieve this, we loop through ema_results and print the symbols found in each key. A combination of print statements and a little string manipulation will do the trick.

# print results
print('Primary EMA Period: %s' %(ema_periods[0]))
print('Secondary EMA Period: %s\n' %(ema_periods[1]))
print('EMA(%s) cross below EMA(%s):\n' %(ema_periods[0], ema_periods[1]))
for x_down in ema_results['cross-downs']:
    print(x_down)
print('\nEMA(%s) cross above EMA(%s):\n' %(ema_periods[0], ema_periods[1]))
for x_up in ema_results['cross-ups']:
    print(x_up)

And that's it! When you run the script, it should produce output similar to the one shown below. Different market conditions will produce different results.

Find the complete Python script on GitHub: ta_ema_cross.py

Other Tutorials In The Series

• Technical Analysis Using Python: Stochastic Oscillator (Basic)