Oliviervha

import numpy as np

def TrainValidSplit(X, y):
    
    total_sequences = len(X)

    # define train, val, test sizes from config
    train_size  = int(total_sequences * 0.7)
    val_size    = train_size + int(total_sequences * 0.3)

Oliviervha

import tensorflow as tf

def trainModel(X_train, X_valid, y_train, Y_valid, n_features = 6, sequence_length = 5):

    model = tf.keras.models.Sequential() # initialize sequential model
    
    model.add(tf.keras.layers.SimpleRNN(64, input_shape = (sequence_length, n_features), activation = 'sigmoid', return_sequences = False)) # add input layer
    model.add(tf.keras.layers.Dense(64, activation="sigmoid")) # add dense layer with X internal units (neurons)
    model.add(tf.keras.layers.Dense(1, activation="sigmoid")) # add dense layer as output layer (sigmoid activation as we're interested in the probability)

Oliviervha

from sklearn.model_selection import TimeSeriesSplit

def TimeSeriesSplitter(norm_input_data, n_features = 6, sequence_length = 5, output_length = 1):
    '''
    # Time series split
    Provides train/test indices to split time series data samples that are observed at fixed time intervals, in train/test sets. 
    In each split, test indices must be higher than before, and thus shuffling in cross validator is inappropriate.
    '''
    
    X,y = [],[]

Oliviervha

import pandas as pd
from IPython.display import display

input_data = pd.read_csv("crypto_data.csv")
  
display(data.head(5))

Oliviervha

from sklearn.preprocessing import MinMaxScaler

scaler = MinMaxScaler()
scaler.fit(input_data)
norm_input_data = pd.DataFrame(scaler.transform(input_data), columns = input_data.columns)

display(norm_input_data.head(5))

Oliviervha

from textblob import TextBlob

def getPolarity(sentence):

    polarity = TextBlob(sentence).sentiment.polarity
    
    return polarity

Oliviervha

from python_bitvavo_api.bitvavo import Bitvavo

bitvavo = Bitvavo({
                    'APIKEY': 'your-api-key',
                    'APISECRET': 'your-api-secret-key',
                    'RESTURL': 'https://api.bitvavo.com/v2',
                    'WSURL': 'wss://ws.bitvavo.com/v2/',
                    'ACCESSWINDOW': 10000,
                    'DEBUGGING': False
                })

Oliviervha

def predict_crypto():
    
    crypto_data = getCryptoData(symbols = ['vethor-token', 'vechain', 'bitcoin'])

    model = keras.models.load_model('.\models')
    
    # predict next day price sign fluctuation based on last 5 timesteps
    proba = model.predict(crypto_data)  

    return proba # e.g. 0.74 

Oliviervha

import requests
import json 
from textblob import TextBlob

def getPolarity(sentence):

    polarity = TextBlob(sentence).sentiment.polarity
    
    return polarity

Oliviervha

from requests import Session
from requests.exceptions import ConnectionError, Timeout, TooManyRedirects
import json

def getCryptoData(symbols = ['vethor-token', 'vechain', 'bitcoin']):

    crypto_data = []

    for symbol in symbols: