Checking your Stocks with Alphavantage and Python

If you have not yet heard of Alphavantage, it is a Stock market API that gives you many stock charts in JSON. You can obtain a Key for Free, and so long as you keep your API calls limited to 5 API requests per minute; 500 API requests per day, you can use their API. Let’s learn how to use Alphavantage and Python to build a sock analysis application.

Getting Started

  1. Obtain a Key From Alphvantage.
  2. Open a new project in Pycharm.
  3. Open the provided project in another Pycharm window. (Optional)

Handling Alphavantage

First lets build out a stock handler that will allow us to make easy calls to Alphavantage and handle the JSON Data.

StockHandler.py


import numpy as np
import json
import requests


key = 'Put your alphavantage key here'
#Alphavantage specific variables
interval = {1:'1min', 5:'5min', 15:'15min', 30:'30min', 60:'60min', 'daily':'Daily', 'month': 'Monthly Time Series'} #min allowed: 1, 5, 15, 30, 60 > Daily: 'Daily'
function = {'intra' :'TIME_SERIES_INTRADAY', 'daily': 'TIME_SERIES_DAILY', 'month': 'TIME_SERIES_MONTHLY'}
i_type = {'intra': 0, 'none': 1}
#END Alphavantage specific variables

# A simple function that allows for easier transition between AlphaVantage's "intraday" 1,5,15,30,60 minute time series,
# and their "Daily" series
def alphavantage_time_series_url(function, ticker, interval, i_type):

    url = 'https://www.alphavantage.co/query?function=' + function

    switcher = {
        0: url + '&symbol=' + ticker + '&interval=' + interval + '&outputsize=full&apikey=' + key,
        1: url + '&symbol=' + ticker + '&apikey=' + key,
        2: url + '&symbol=' + ticker + '&interval=1min&apikey=' + key
    }
    return switcher.get(i_type, "Invalid type")

def load_data(function, ticker, interval, i_type):
    url = alphavantage_time_series_url(function, ticker,  interval, i_type)
    params = {"retina_name": "en_associative","start_index": 0, "max_results": 1, "sparsity": 1.0, "get_fingerprint": False}
    r = requests.get(url=url, params=params)
    data_str = r.json()

    ds = json.dumps(data_str)
    json_obj = json.loads(ds)
    try:
        if interval == 'Monthly Time Series' or interval == 'Technical Analysis: VWAP':
            dataset = json_obj[interval]
        else:
            dataset = json_obj["Time Series (" + interval + ")"]
    except:
        return None, None

    x = []
    ox = []
    y = []
    z = []
    count = 0
    for key in dataset:
        x.append(count)
        count += 1
        ox.append(key)
        try:
            y.append(average([float(dataset[key]["2. high"]), float(dataset[key]["3. low"])]))
            z.append(float(dataset[key]["5. volume"]))
        except:
            y.append(float(dataset[key]["VWAP"]))

    if len(y) % 2 != 0:
        y.pop(0)
        x.pop(0)
        try:
            z.pop(0)
        except:
            z = None

    #deviation = Reverse(cut_mean_from_elements(y))
    #zdev = Reverse(cut_mean_from_elements(z))

    deviation = Reverse(y)
    try:
        zdev = Reverse(z)
        np_z = np.array(zdev).reshape(len(z), 1)
    except:
        np_z = None

    np_x = np.array(x).reshape(len(x), 1)
    np_y = np.array(deviation).reshape(len(y), 1)

    #print(ox)
    return {'x': np_x, 'y': np_y, 'z': np_z}, {'ox': Reverse(ox), 'y': y}

def get_quotes(symbols):
    url = 'https://www.alphavantage.co/query?function=BATCH_STOCK_QUOTES&apikey=' + key + '&symbols=' + symbols
    params = {"retina_name": "en_associative", "start_index": 0, "max_results": 1, "sparsity": 1.0,
              "get_fingerprint": False}
    r = requests.get(url=url, params=params)
    print(r)
    data_str = r.json()

    ds = json.dumps(data_str)
    json_obj = json.loads(ds)

    dataset = json_obj["Stock Quotes"]

    return dataset

def Reverse(lst):
    return [ele for ele in reversed(lst)]


def average(set):
    count = len(set)
    total = 0
    for part in set:
        total += part

    return total / count


def cut_mean_from_elements(set):
    mean = average(set)
    deviation = []
    for element in set:
        deviation.append(element - mean)

    return deviation

Please put your key in the ‘key’ variable.

Building Analyzers

Once we get the data we need to do something with it. Here I have built three different files.

DataManipulate.py

A simple snippet of code to use wherever it is needed. Simply take non-numeric data and make it numeric.

def pieces(l, n):
    n = max(1, n)
    return (l[i:i + n] for i in range(0, len(l), n))

LinearRegression.py

The simplest Stock visualization is a simple trend line. We will use Linear Regression to get a trend line.

import numpy as np
from sklearn.linear_model import LinearRegression
import Plotting.PlotFunctions as plt
from sklearn.svm import SVR
from sklearn.model_selection import GridSearchCV
import sklearn.metrics as metrics

class LR:

    model = None
    vmodel = None
    x = None
    y = None
    z = None
    accuracies = []


    def __init__(self, data, t):
        self.x = np.array(data['x']).reshape((-1, 1))
        self.y = np.array(data['y'])
        self.z = np.array(data['z'])
        self.ticker = t

        self.model = LinearRegression()
        self.vmodel = LinearRegression()
        self.model.fit(self.x, self.y)
        try:
            self.vmodel.fit(self.x, self.z)
        except:
            self.vmodel = None

    def get_trend(self):
        return self.model.coef_, self.model.intercept_


    def predict(self, x):
        return self.model.predict(x)

    def pieces(self, l, n):
        n = max(1, n)
        return (l[i:i + n] for i in range(0, len(l), n))

    def regressorOp(self, x, y):
        """
        This will optimize the parameters for the algo
        """
        regr_rbf = SVR(kernel="rbf", verbose=True, cache_size=1024)
        C = [10000, 6000, 5000, 1000, 10, 1]
        gamma = [0.005, 0.004, 0.003, 0.002, 0.001, 1, 2, 3, 4]
        epsilon = [0.1, 0.01, 0.05, 1]
        parameters = {"C": C, "gamma": gamma, "epsilon": epsilon}

        gs = GridSearchCV(regr_rbf, parameters, scoring="neg_mean_squared_error", cv=2, iid=True)
        gs.fit(x, y)

        #print("Best Estimator:\n", gs.best_estimator_)
        #print("Type: ", type(gs.best_estimator_))
        return gs.best_estimator_

    def view_trend_for_subset(self, sets):
        # Data is deviation, and originalData is true timeseries
        # (data.x = 0,1,2,...,n | originalData.x = 01-01-2020, ..., 02-01-2020, ..., n
        xPieces = self.pieces(self.x, int(len(self.x) / sets))
        yPieces = self.pieces(self.y, int(len(self.y) / sets))
        try:
            zPieces = self.pieces(self.z, int(len(self.z) / sets))
        except:
            zPieces = []
        xPieces = list(xPieces)
        yPieces = list(yPieces)
        zPieces = list(zPieces)

        xParts = []
        yParts = []
        zParts = []
        mParts = []
        bParts = []
        predictions = []
        # print(xPieces[3], yPieces[3], "\n",  xPieces[4], yPieces[4])
        for sub in range(sets):
            try:
                z = zPieces[sub]
            except:
                z = -1
            subdata = {'x': xPieces[sub], 'y': yPieces[sub], 'z': z}
            #print(subdata)
            self.model = LinearRegression()
            self.model.fit(subdata['x'], subdata['y'])
            #self.vmodel = SVR(kernel='poly', gamma='auto')
            #self.vmodel.fit(subdata['x'], subdata['z'].ravel())

            if z != -1:
                clf = self.regressorOp(subdata['x'], subdata['z'].ravel())
                clf.fit(subdata['x'], subdata['z'].ravel())

                y_pred = clf.predict(subdata['x'])

                accuracy = clf.score(subdata['x'], subdata['z'].ravel())
                variance = metrics.explained_variance_score(subdata['z'], y_pred)
                self.accuracies.append(accuracy)
                prediction = (clf.predict(subdata['x']))
                predictions.append(prediction)

            #print("prediction: ", prediction)

            xParts.append(subdata['x'])
            yParts.append(subdata['y'])
            zParts.append(subdata['z'])
            mParts.append(self.model.coef_)
            bParts.append(self.model.intercept_)

            #plt.plot_all(subdata['x'], {1: subdata['y'], 2: subdata['z']}, self.model.coef_, self.model.intercept_, prediction, self.ticker, 0)

            #print("PARAMS: ", clf.get_params(True))
        return mParts, bParts, xParts, yParts, zParts, predictions

GradientDescent.py

Another resourceful yet slightly more complex trend would be finding a non-linear trend with Gradient Descent.

import numpy as np
import scipy.linalg

def gradient_descent(X, Y, w, c, it, L):
    n = len(X)
    Y_pred = w * X + c  # The current predicted value of Y
    D_m = (-2 / n) * sum(X * (Y - Y_pred))  # Derivative wrt m
    D_c = (-2 / n) * sum(Y - Y_pred)  # Derivative wrt c
    m = w - L * D_m  # Update m
    c = c - L * D_c  # Update c

    return m, c


def batch_gradient_descent(x, y, w, eta):
    derivative = np.sum([-(y[d]-np.dot(w.T.copy(),x[d,:]))*(x[d,:]).reshape(np.shape(w)) for d in range(len(x))],axis=0)
    return eta*(1/len(x))*derivative


def mini_batch_gradient_descent(x, y, w, eta, batch):
    gradient_sum = np.zeros(shape=np.shape(w))
    for b in range(batch):
        choice = np.random.choice(list(range(len(x))))
        gradient_sum += -(y[choice]-np.dot(w.T,x[choice,:]))*x[choice,:].reshape(np.shape(w))
        return eta*(1/batch)*gradient_sum


# initialize variables
def trend_for(data, it, L, m, b):
    x = data['x']
    y = data['y']
    w = np.random.normal(size=(np.shape(x)[1], 1))
    c = 0.0
    # Update w
    w_s = []
    Error = []

    for i in range(it):
        # Calculate error
        error = (1 / 2) * np.sum([(y[i] - (b + np.dot(w.T, x[i, :]))) ** 2 for i in range(len(x))])
        #print(error)
        Error.append(error)
        if error == float('inf'):
            print('INFINITY')
            return x, y, m, b, Error

        m, b = gradient_descent(x, y, w, c, it, L)
        w = m
        c = b

        w_s.append(w.copy())

    return x, y, m, b, Error

Lets Plot Something

We can find all of this data, and build trends for the data, but now we need to see or visualize the data. Lets Plot it

PlotFunctions.py

import numpy as np
import matplotlib.pyplot as plt


def plot_function(x, y, w,b, Error, st, iterations):
    if len(Error) == iterations:
        fig, ax = plt.subplots(nrows=1,ncols=2,figsize=(40,10))
        ax[0].plot(x, ((x * w) + b), c='lightgreen', linewidth=3, zorder=0)
        ax[0].plot(x, y)
        ax[0].set_title(st)
        ax[1].scatter(range(iterations), Error)
        ax[1].set_title('Error')
        plt.show()
    else:
        print('Plotting Error: Iterations and Errors are different sizes')

def plot_all(x, set, w,b, vw, st, iterations):

    fig, ax = plt.subplots(nrows=1,ncols=2,figsize=(40,10))
    wb = [((xi * w) + b) for xi in x]
    print(wb.shape, " >> ", x.shape)
    wb = np.array(wb).reshape(len(x), 1)
    print(wb.shape, " >> ", x.shape)
    ax[0].plot(x, wb, c='lightgreen', linewidth=3, zorder=0)
    ax[0].plot(x, set[1])
    ax[0].set_title(st)
    try:
        ax[1].plot(x, set[2])
        ax[1].set_title('Volume')
        ax[1].plot(x, vw)
    except:
        ''
    plt.show()

Building an Alphavantage Service

Now we built a bunch of services, but no one want to recreate all of this every time they want to see a stock quote. Lets build a service that we can easily call one function and get an entire process all in one.

AlphavantageService.py

import GradientDescent.GD as gd
import Plotting.PlotFunctions as plt
import JsonHandler.StockHandler as sh
import Services.DataManipulate as dm
import LinearRegression.LR as lr

class AlphavantageService:
    data = None
    originalData = None
    ticker = None
    iterations = None
    L = None
    x = None
    y = None
    m = 0.0
    b = 0.0
    Error = None
    DataSet = True

    def __init__(self, t, i, l, stype, mins):
        self.ticker = t
        self.iterations = i
        self.L = l
        if stype == 0:
            self.data, self.originalData = sh.load_data(sh.function['daily'], t, sh.interval['daily'], sh.i_type['none'])
        elif stype == 1:
            try:
                self.data, self.originalData = sh.load_data(sh.function['intra'], t, sh.interval[mins],
                                                        sh.i_type['intra'])
            except:
                self.DataSet = False
        elif stype == 2:
            self.data, self.originalData = sh.load_data(sh.function['month'], t, sh.interval['month'],
                                                        sh.i_type['none'])
        elif stype == 3:
            self.data, self.originalData = sh.load_data('VWAP', t, 'Technical Analysis: VWAP',
                                                        2)

    def get_data(self):
        return self.data

    def set_data_switch(self, val):
        self.DataSet = val

    def originalX(self):
        return self.originalData['ox']


    def view_trend_for(self, l):
        self.L = l
        # Data is deviation, and originalData is true timeseries
        # (data.x = 0,1,2,...,n | originalData.x = 01-01-2020, ..., 02-01-2020, ..., n
        self.x, self.y, self.m, self.b, self.Error = gd.trend_for(self.data, self.iterations, self.L, self.m, self.b)
        return self.m, self.b, self.Error[len(self.Error) - 1] / len(self.x)

    def gradient_descent(self, l, sets, ticker):
        self.L = l
        # Data is deviation, and originalData is true timeseries
        # (data.x = 0,1,2,...,n | originalData.x = 01-01-2020, ..., 02-01-2020, ..., n
        xPieces = dm.pieces(self.data['x'], int(len(self.data['x']) / sets))
        yPieces = dm.pieces(self.data['y'], int(len(self.data['y']) / sets))

        xPieces = list(xPieces)
        yPieces = list(yPieces)
        xParts = []
        yParts = []
        mParts = []
        bParts = []
        # print(xPieces[3], yPieces[3], "\n",  xPieces[4], yPieces[4])
        for sub in range(sets):
            subdata = {'x': xPieces[sub], 'y': yPieces[sub]}
            xPart, yPart, mPart, bPart, ErrorPart = gd.trend_for(subdata, self.iterations, self.L, 0.0, 0.0)
            xParts.append(xPart)
            yParts.append(yPart)
            mParts.append(mPart)
            bParts.append(bPart)

            plt.plot_function(self.originalData['ox'], yPart, mPart, bPart, ErrorPart, self.ticker, self.iterations)

        return mParts, bParts, 0

    def linear_regression(self, sets):
        return lr.view_trend_for_subset(sets)

    def plot_trend(self, acceptedError, error):
        # Plot the predicted function and the Error
        if error < acceptedError:
            plt.plot_function(self.x, self.y, self.m, self.b, self.Error, self.ticker, self.iterations)

    def get_daily_trend_for(ticker, iterations, L):
        data, originalData = sh.load_data(sh.function['daily'], ticker, sh.interval['daily'], sh.i_type['none'])
        x, y, m, b, self.Error = gd.trend_for(data, iterations, L)
        return m, b, Error[len(Error) - 1]

    def get_trendline_for(ticker, iterations, L):
        data, originalData = sh.load_data(sh.function['intra'], ticker, sh.interval[60], sh.i_type['intra'])
        x, y, m, b, Error = gd.trend_for(data, iterations, L)
        return m, b, Error[len(Error) - 1]

Build for Future Enhancements

This is one API. The next article in this series will take this project and tie in another Free API Service that has different data sets. Lets build a service that we call once, and manages Alphavantage as well as other API’s

The service we build will have two key functions. It will have a deep analysis function, and it will also build a watch list.

Watch List

If you want to scan through a large group of tickers, and find stocks that fit your criteria, we can build a function to do this.

What this means is we can use a flat file with tickers, and a reader to read the tickers.

tickers.txt

AAPL
XOM
MSFT
BAC^I
IBM
CVX
GE
WMT
T
JNJ
PG
WFC
KO
PFE
JPM
GOOG
PM
VOD
ORCL
INTC
MRK
VZ

Then, something to read the file

FlatDataReader.py

def read(file):
    tickers = []
    with open(file) as f:
        for line in f:
            line = line.replace("\n", "")
            tickers.append(line)

    return tickers

Our Final Service

StockService.py

import LinearRegression.LR as lr
import Services.AlphavantageService as alphavantage
import numpy as np
from sklearn.linear_model import LinearRegression
import Plotting.PlotFunctions as plt
import JsonHandler.StockHandler as sh
import FlatData.FlatDataReader as fdr
import Services.DataManipulate as dm
import time

#Run: Used to set which tickers you want, some Ai Settings, and what type of dataset you want to work with

#Ai settings
iterations = 20000
L = 0.1
errortolerance = 20
trials = 1000 #set a trial limit
Error = []
#END Ai settings

#OK Now lets loop through and find the best Learning rate
not_found = True

def deep_analysis_of_stock(ticker):

    # cut data into many components
    # First lets Analyze longest trend
    stock_service = alphavantage.AlphavantageService(ticker, 10, 1, 3, 0)
    lineReg = lr.LR(stock_service.get_data(), ticker)
    mParts, bParts, xParts, yParts, zParts, prediction = lineReg.view_trend_for_subset(2)

    for i in range(len(xParts)):
        print(xParts[i].shape, " >> ", yParts[i].shape)
        print('ticker: ', ticker, 'm: ', mParts[i], 'b: ', bParts[i])
        plt.plot_all(xParts[i], {1: yParts[i], 2: zParts[i]},
                     mParts, bParts[i], None, ticker, 0)

    return mParts, bParts, xParts, yParts, zParts, prediction


def get_watchlist(min_price, max_price, wait, min_slope):

    tickers = fdr.read("C:\\Path\\To\\ticker.txt")
    ticker_batches = dm.pieces(tickers, 750)
    ticker_batches = list(ticker_batches)
    print('Batches: ', len(ticker_batches))

    in_budget_stocks = []
    for i in ticker_batches:
        print("batchSize: ", len(i))
        batchString = ','.join(i)
        print('batch: ', batchString)

        data = sh.get_quotes(batchString)

        for line in data:
            if max_price > float(line['2. price']) > min_price:
                print("price: ", line['2. price'], "symbol: ", line['1. symbol'])
                in_budget_stocks.append(line)
        time.sleep(wait)  # Throttle API Calls

    stock_data = []
    for stock in in_budget_stocks:
        print('Reviewing Symbol: ', stock['1. symbol'])
        stock_service = None
        try:
            stock_service = alphavantage.AlphavantageService(stock['1. symbol'], iterations, L, 0, 15)
        except:
            stock_service.set_data_switch(False)
            continue
        if stock_service is None or stock_service.get_data() is None:
            print('No Data Available for: ', stock['1. symbol'])
            continue
        lineReg = lr.LR(stock_service.get_data(), stock['1. symbol'] + " From " + stock_service.originalX()[0] + " TO "
                        + stock_service.originalX()[len(stock_service.originalX()) - 1])

        mParts, bParts, xParts, yParts, zParts, prediction = lineReg.view_trend_for_subset(1)

        analysis = {'ticker': stock['1. symbol'], 'mParts': mParts, 'bParts': bParts, 'xParts': xParts, 'yParts': yParts,
                    'zParts': zParts, 'pz': prediction}

        stock_data.append(analysis)
        time.sleep(wait)

    watch_list = []

    for analysis in stock_data:
        for i in range(len(analysis['xParts'])):
            print('ticker: ', analysis['ticker'], 'm: ', analysis['mParts'][i], 'b: ', analysis['bParts'][i])
            if analysis['mParts'][i] > min_slope:
                watch_list.append(analysis)
                plt.plot_all(analysis['xParts'][i], {1: analysis['yParts'][i], 2: analysis['zParts'][i]},
                             analysis['mParts'][i], analysis['bParts'][i], analysis['pz'][i], analysis['ticker'], 0)


    ##print('WatchList: ', watch_list)
    return watch_list

And Finally, Use your Stock Service

Run.py

import StockService as ss

wl = ss.get_watchlist(100, 105, 15, 0.1)
print(wl)
# Now, once we get are watch list, we can Do further analysis.
for analysis in wl:
    print('Deep Analysis', ss.deep_analysis_of_stock(analysis['ticker']))

Conclusion

This code is intended to show how to use Alphvantage and Python to begin the process of a Stock Analysis Tool. This Tool will almost surely not help you make decisions with your stock picks.

This Application will however give you a great head start on building a Stock service using Alphavantage and Python.

If you would like to view or download this code, it has been placed on Github.

Leave a Reply