Technical Analysis Graphs using Python Matplotlib

matplotlib.py

import requests
from datetime import datetime
import matplotlib.pyplot as plt


def cbpGetHistoricRates(market='BTC-GBP', granularity=86400, iso8601start='', iso8601end=''):
    api = 'https://api.pro.coinbase.com/products/' + market + '/candles?granularity=' + \
        str(granularity) + '&start=' + iso8601start + '&end=' + iso8601end
    resp = requests.get(api)
    if resp.status_code != 200:
        raise Exception('GET ' + api + ' {}'.format(resp.status_code))
    data = {}
    for price in reversed(resp.json()):
        # time, low, high, open, close, volume
        iso8601 = datetime.fromtimestamp(price[0])
        timestamp = datetime.strftime(iso8601, "%d/%m/%Y %H:%M:%S")
        data[timestamp] = price[4]
    return data


def simpleMovingAverage(data, num):
    if not isinstance(data, dict):
        raise Exception('Dictionary input expected')
    if not isinstance(num, int):
        raise Exception('Integer input expected')
    if num < 5 or num > 200:
        raise Exception('Unusual numeric input detected')
    if (num > len(data)):
        raise Exception('Insufficient data for calculation')

    data_keys = list(data.keys())
    data_list = list(data.values())

    result = {}
    for x in range(len(data_list) - num + 1):
        series = data_list[x:x + num]
        result[data_keys[x + num - 1]] = round(sum(series) / num)

    return result


def exponentialMovingAverage(data, num):
    if not isinstance(data, dict):
        raise Exception('Dictionary input expected')

    if not isinstance(num, int):
        raise Exception('Integer input expected')

    if num < 9 or num > 26:
        raise Exception('Unusual numeric input detected')

    if (num > len(data)):
        raise Exception('Insufficient data for calculation')

    data_keys = list(data.keys())
    data_list = list(data.values())

    last_sma = -1
    result = {}
    for x in range(len(data_list) - num + 1):
        series = data_list[x:x + num]

        if (last_sma == -1):
            result[data_keys[x + num - 1]] = round((sum(series) / num), 2)
        else:
            current_price = data[data_keys[x + num - 1]]
            result[data_keys[x + num - 1]] = round(
                current_price * 2 / (num + 1) + last_sma * (1 - 2 / (num + 1)), 2)

        last_sma = result[data_keys[x + num - 1]]

    return result


def movingAverageConvergenceDivergence(data):
    if not isinstance(data, dict):
        raise Exception('Dictionary input expected')

    if (26 > len(data)):
        raise Exception('Insufficient data for calculation')

    ema12_data = exponentialMovingAverage(data, 12)
    ema26_data = exponentialMovingAverage(data, 26)

    macd_data = {}
    data_keys = list(data.keys())
    for key in data_keys:
        ema12 = 0
        if key in ema12_data:
            ema12 = ema12_data[key]

        ema26 = 0
        if key in ema26_data:
            ema26 = ema26_data[key]

        if (ema12 > 0) and (ema26 == 0):
            macd_data[key] = 0
        else:
            macd_data[key] = ema12 - ema26

    signal_data = exponentialMovingAverage(macd_data, 9)

    result = {}
    for key in data_keys:
        price = 0
        if key in data:
            price = data[key]

        ema12 = 0
        if key in ema12_data:
            ema12 = ema12_data[key]

        ema26 = 0
        if key in ema26_data:
            ema26 = ema26_data[key]

        macd = 0
        if key in macd_data:
            macd = macd_data[key]

        signal = 0
        if key in signal_data:
            signal = signal_data[key]

        result[key] = {
            'price': price,
            'ema12': ema12,
            'ema26': ema26,
            'macd': macd,
            'signal': signal
        }

    return result


def relativeStrengthIndex(data, num):
    if not isinstance(data, dict):
        raise Exception('Dictionary input expected')

    if not isinstance(num, int):
        raise Exception('Integer input expected')

    if num < 7 or num > 21:
        raise Exception('Unusual numeric input detected')

    if (num > len(data)):
        raise Exception('Insufficient data for calculation')

    data_keys = list(data.keys())
    data_list = list(data.values())
    result = {}
    last_price = -1
    gains_losses_list = []
    for x in range(len(data_list)):
        if (last_price != -1):
            diff = round((data_list[x] - last_price), 2)

            if (diff > 0):
                gains_losses = [data_list[x], diff, 0]
            elif (diff < 0):
                gains_losses = [data_list[x], 0, abs(diff)]
            else:
                gains_losses = [data_list[x], 0, 0]

            gains_losses_list.append(gains_losses)

        sum_gains = 0
        sum_losses = 0
        avg_gains = 0
        avg_losses = 0
        if (x == num):
            series = gains_losses_list[-num::]

            for y in series:
                sum_gains += y[1]
                sum_losses += y[2]
            avg_gains = sum_gains / num
            avg_losses = sum_losses / num
            rs = avg_gains / avg_losses
            rsi = 100 - (100 / (1 + rs))
            last_gain_avg = avg_gains
            last_loss_avg = avg_losses
            result[data_keys[x]] = round(rsi, 2)
        if (x > num):
            current_list = gains_losses_list[-1::]
            current_gain = current_list[0][1]
            current_loss = current_list[0][2]
            current_gains_avg = (
                last_gain_avg * (num - 1) + current_gain) / num
            current_losses_avg = (
                last_loss_avg * (num - 1) + current_loss) / num
            rs = current_gains_avg / current_losses_avg
            rsi = 100 - (100 / (1 + rs))
            last_gain_avg = current_gains_avg
            last_loss_avg = current_losses_avg
            result[data_keys[x]] = round(rsi, 2)

        last_price = data_list[x]

    return result


data = cbpGetHistoricRates('BTC-GBP', 86400)
sma20 = simpleMovingAverage(data, 20)
sma50 = simpleMovingAverage(data, 50)
sma200 = simpleMovingAverage(data, 200)
ema12 = exponentialMovingAverage(data, 12)
ema26 = exponentialMovingAverage(data, 26)
macd = movingAverageConvergenceDivergence(data)
rsi14 = relativeStrengthIndex(data, 14)

data_values = list(data.values())
trim_data = data_values[(len(sma200) * -1) - 1:-1]

sma20_values = list(sma20.values())
trim_sma20 = sma20_values[(len(sma200) * -1) - 1:-1]

sma50_values = list(sma50.values())
trim_sma50 = sma50_values[(len(sma200) * -1) - 1:-1]

ema12_values = list(ema12.values())
trim_ema12 = ema12_values[(len(sma200) * -1) - 1:-1]

ema26_values = list(ema26.values())
trim_ema26 = ema26_values[(len(sma200) * -1) - 1:-1]

macd_list = []
signal_list = []
for macd_dict in macd.values():
    macd_list.append(macd_dict['macd'])
    signal_list.append(macd_dict['signal'])

trim_macd = macd_list[(len(sma200) * -1) - 1:-1]
trim_signal = signal_list[(len(sma200) * -1) - 1:-1]

rsi14_values = list(rsi14.values())
trim_rsi14 = rsi14_values[(len(sma200) * -1) - 1:-1]

ax1 = plt.subplot(411)
plt.plot(trim_data, label="price")
plt.plot(trim_sma20, label="sma20")
plt.plot(trim_sma50, label="sma50")
plt.plot(sma200.values(), label="sma200")
plt.legend()
plt.ylabel('Price')

plt.subplot(412, sharex=ax1)
plt.plot(trim_data, label="price")
plt.plot(trim_ema12, label="ema12")
plt.plot(trim_ema26, label="ema26")
plt.legend()
plt.ylabel('Price')

plt.subplot(413, sharex=ax1)
plt.plot(trim_macd, label="macd")
plt.plot(trim_signal, label="signal")
plt.legend()
plt.ylabel('Divergence')

plt.subplot(414, sharex=ax1)
plt.plot(trim_rsi14, label="rsi14")
plt.legend()
plt.ylabel('Percentage')
plt.xlabel('Days')

plt.show()