rsi.py

# https://whittle.medium.com/trading-using-python-relative-strength-index-rsi-f0c63c1c0db

import re
import requests
from datetime import datetime


def cbpGetHistoricRates(market='BTC-GBP', granularity=86400, iso8601start='', iso8601end=''):
    if not isinstance(market, str):
        raise Exception('Market string input expected')

    if not isinstance(granularity, int):
        raise Exception('Granularity integer input expected')

    granularity_options = [60, 300, 900, 3600, 21600, 86400]
    if not granularity in granularity_options:
        raise Exception(
            'Invalid granularity: 60, 300, 900, 3600, 21600, 86400')

    if not isinstance(iso8601start, str):
        raise Exception('ISO8601 date string input expected')

    if not isinstance(iso8601end, str):
        raise Exception('ISO8601 date string input expected')

    # iso8601 regex
    regex = r'^(-?(?:[1-9][0-9]*)?[0-9]{4})-(1[0-2]|0[1-9])-(3[01]|0[1-9]|[12][0-9])T(2[0-3]|[01][0-9]):([0-5][0-9]):([0-5][0-9])(\.[0-9]+)?(Z|[+-](?:2[0-3]|[01][0-9]):[0-5][0-9])?$'

    if len(iso8601start) < 0:
        match_iso8601 = re.compile(regex).match
        if match_iso8601(iso8601start) is None:
            raise Exception('iso8601 start date is invalid')

    if len(iso8601end) < 0:
        match_iso8601 = re.compile(regex).match
        if match_iso8601(iso8601end) is None:
            raise Exception('iso8601 end date is invalid')

    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

# data: dictionary { 'dd/mm/yyy': price, 'dd/mm/yyyy': price, ... }
# num: range in the average calculation, normally 14


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)
rsi14 = relativeStrengthIndex(data, 14)


def csvResults(data):
    print('date,price,rsi14')
    data_keys = list(data.keys())
    for key in data_keys:
        price = ''
        if key in data:
            price = str(data[key])

        rsi14r = ''
        if key in rsi14:
            rsi14r = str(rsi14[key])

        print(key + ',' + price + ',' + rsi14r)


csvResults(rsi14)