xrp_analysis_2023.py

# -*- coding: utf-8 -*-
"""XRP_ANALYSIS_2023.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/14Mk0JDu51SiPtvlq9YNsLAsO8RXQjerG
"""

import requests
import zipfile
import pandas as pd
import numpy as np
import datetime
import matplotlib.pyplot as plt
import seaborn as sns
import statsmodels.api as sm
from statsmodels.tsa.seasonal import seasonal_decompose
from statsmodels.tsa.stattools import adfuller
from scipy.stats import pearsonr
from statsmodels.tsa.arima.model import ARIMA
from sklearn.metrics import mean_squared_error
from statsmodels.graphics.tsaplots import plot_acf, plot_pacf
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import MinMaxScaler
from statsmodels.tsa.arima.model import ARIMA
from sklearn.preprocessing import StandardScaler
from tensorflow.keras.layers import LSTM
!pip install hurst
from hurst import compute_Hc
from keras.models import Sequential
from keras.layers import Dense

# List of URLs
url = [
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-12.zip",
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-11.zip",
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-10.zip",
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-09.zip",
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-08.zip",
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-07.zip",
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-06.zip",
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-05.zip",
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-04.zip",
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-03.zip",
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-02.zip",
    "https://data.binance.vision/data/spot/monthly/klines/XRPUSDT/1d/XRPUSDT-1d-2023-01.zip"
]

# Download and extract data for each month
for i in range(len(url)):
    response = requests.get(url[i])

    if response.status_code == 200:
        with open(f'XRPUSDT-1d-2023-{str(i+1).zfill(2)}.zip', 'wb') as file:
            file.write(response.content)
        with zipfile.ZipFile(f'XRPUSDT-1d-2023-{str(i+1).zfill(2)}.zip', 'r') as zip_ref:
            zip_ref.extractall('data')
    else:
        print(f'Failed to retrieve file: {response.status_code}')

# Load data for each month
data_frames = []
for i in range(len(url)):
    data = pd.read_csv(f'data/XRPUSDT-1d-2023-{str(i+1).zfill(2)}.csv', header=None)
    data_frames.append(data)

# Concatenate data frames
full_data = pd.concat(data_frames)
column_names = ["Open_time", "Open", "High", "Low", "Close", "Volume", "Close_time",
                "Quote_asset_volume", "Number_of_trades", "Taker_buy_base_asset_vol",
                "Taker_buy_quote_asset_vol", "Ignore"]

# Set column names
full_data.columns = column_names

# Function to convert timestamp to datetime
def convert_timestamp(timestamp_ms):
    timestamp_s = timestamp_ms / 1000
    date = datetime.datetime.fromtimestamp(timestamp_s)
    return date

# Apply the function to timestamp columns
full_data['Open_time'] = full_data['Open_time'].apply(convert_timestamp)
full_data['Close_time'] = full_data['Close_time'].apply(convert_timestamp)

# Format Quote_asset_volume and Taker_buy_quote_asset_vol columns
full_data["Quote_asset_volume"] = full_data["Quote_asset_volume"].apply(lambda x: '{:.2f}'.format(x))
full_data["Taker_buy_quote_asset_vol"] = full_data["Taker_buy_quote_asset_vol"].apply(lambda x: '{:.2f}'.format(x))

# Display basic statistics
price_stats = full_data['Close'].describe()
volume_stats = full_data['Volume'].describe()
stats = pd.DataFrame({"Close_price": price_stats, "Volume": volume_stats})
print(stats)

# Calculate and display additional financial metrics
daily_returns = full_data['Close'].pct_change().dropna()
volatility = daily_returns.std()
skewness = daily_returns.skew()
kurtosis = daily_returns.kurtosis()
risk_free_rate = 0.05
average_daily_return = daily_returns.mean()
sharpe_ratio = (average_daily_return - risk_free_rate) / volatility

results = pd.DataFrame({
    "Volatility": [volatility],
    "Skewness": [skewness],
    "Kurtosis": [kurtosis],
    "Sharpe Ratio": [sharpe_ratio]
})
print(results)

# Calculate and display RSI
def calculate_rsi(data, window=14):
    delta = data.diff()
    gain = delta.where(delta > 0, 0).rolling(window=window).mean()
    loss = -delta.where(delta < 0, 0).rolling(window=window).mean()
    rs = gain / loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

full_data['RSI'] = calculate_rsi(full_data['Close'])

# Plot closing price and volume trends
plt.figure(figsize=(12, 6))
plt.plot(full_data['Close_time'], full_data['Close'])
plt.title('Price Trend Over Time')
plt.xlabel("Date")
plt.ylabel('Closing Price')
plt.show()

plt.figure(figsize=(12, 6))
plt.plot(full_data['Close_time'], full_data['Volume'])
plt.title('Volume Trend Over Time')
plt.xlabel('Date')
plt.ylabel('Volume')
plt.show()

# Analyze average volume by day and average price by month
full_data['Day'] = full_data['Close_time'].dt.day
average_volume_by_day = full_data.groupby('Day')['Volume'].mean()
overall_average_volume = full_data['Volume'].mean()

plt.figure(figsize=(12, 6))
average_volume_by_day.plot(kind='bar')
plt.axhline(y=overall_average_volume, color='r', linestyle='-', linewidth=2)
plt.text(0, overall_average_volume, f'Average: {overall_average_volume:.2f}', color='r', va='bottom')
plt.title('Average Volume by Day of Month')
plt.xlabel('Day of Month')
plt.ylabel('Average Volume')
plt.show()

full_data['Month'] = full_data['Close_time'].dt.month
average_price_by_month = full_data.groupby('Month')['Close'].mean()

plt.figure(figsize=(12, 6))
average_price_by_month.plot(kind='bar')
plt.title('Average Price by Month')
plt.xlabel('Month')
plt.ylabel('Average Closing Price')
plt.show()

# Convert 'Volume' and 'Quote Asset Volume' to numeric types
full_data['Volume'] = pd.to_numeric(full_data['Volume'], errors='coerce')
full_data['Quote Asset Volume'] = pd.to_numeric(full_data['Quote_asset_volume'], errors='coerce')
full_data.dropna(subset=['Volume', 'Quote Asset Volume'], inplace=True)

# Calculate correlation
correlation, p_value = pearsonr(full_data['Volume'], full_data['Quote Asset Volume'])
print("Correlation:", correlation)
print("P-value:", p_value)

# Plot Number of Trades trend over time
plt.figure(figsize=(12, 6))
plt.plot(full_data['Close_time'], full_data['Number_of_trades'])
plt.title('Trend in Number of Trades Over Time')
plt.xlabel('Date')
plt.ylabel('Number of Trades')
plt.show()

# Perform ADF test for stationarity
statio_volume = adfuller(full_data["Volume"])
statio_close = adfuller(full_data["Close"])

# Display ADF test results for Volume
adf_statistic_volume = statio_volume[0]
p_value_volume = statio_volume[1]
critical_values_volume = statio_volume[4]

adf_results_volume = {
    'Metric': ['ADF Statistic', 'p-value'],
    'Value': [adf_statistic_volume, p_value_volume]
}

for key, value in critical_values_volume.items():
    adf_results_volume['Metric'].append(f'Critical Value ({key})')
    adf_results_volume['Value'].append(value)

# Creating DataFrame for Volume
adf_results_volume_df = pd.DataFrame(adf_results_volume)
print("ADF Test Results for Volume:")
print(adf_results_volume_df)

# Display ADF test results for Close
adf_statistic_close = statio_close[0]
p_value_close = statio_close[1]
critical_values_close = statio_close[4]

adf_results_close = {
    'Metric': ['ADF Statistic', 'p-value'],
    'Value': [adf_statistic_close, p_value_close]
}

for key, value in critical_values_close.items():
    adf_results_close['Metric'].append(f'Critical Value ({key})')
    adf_results_close['Value'].append(value)

# Creating DataFrame for Close
adf_results_close_df = pd.DataFrame(adf_results_close)
print("ADF Test Results for Close:")
print(adf_results_close_df)

# Plot PACF to identify parameter p
plot_pacf(full_data["Close"], lags=40)
plt.show()

# Plot ACF to identify parameter q
plot_acf(full_data["Close"], lags=40)
plt.show()

# Split the data into train and test sets
train_size = int(len(full_data["Close"]) * 0.7)
train, test = full_data["Close"][:train_size], full_data["Close"][train_size:]

# Fit an ARIMA model (you should determine the order (p, d, q) beforehand)
model = ARIMA(train, order=(1, 1, 1))
model_fit = model.fit()

# Get the AIC for the model
aic = model_fit.aic
print(f'The AIC of the model is: {aic}')

# Fit another ARIMA model with different order
model = ARIMA(train, order=(1, 2, 2))
model_fit = model.fit()

# Get the AIC for the second model
aic = model_fit.aic
print(f'The AIC of the second model is: {aic}')

# Set 'Close_time' as index for time-series analysis
if 'Close_time' in full_data.columns and full_data.index.name != 'Close_time':
    full_data.set_index('Close_time', inplace=True)


# Define SARIMA model parameters
p = 1
d = 1
q = 1
P = 1
D = 1
Q = 1
s = 12

# Create and fit the SARIMA model
sarima_model = sm.tsa.SARIMAX(
    full_data["Close"],
    order=(p, d, q),
    seasonal_order=(P, D, Q, s),
    enforce_stationarity=False,
    enforce_invertibility=False
)
sarima_results = sarima_model.fit()

# Get the AIC for the SARIMA model
aic_sarima = sarima_results.aic
print(f'The AIC of the SARIMA model is: {aic_sarima}')

# Print SARIMA model summary
print(sarima_results.summary())


# Prepare the data for machine learning model
closing_prices = full_data['Close'].values.reshape(-1, 1)
scaler = StandardScaler()
closing_prices_scaled = scaler.fit_transform(closing_prices)

# Create sequences for time series data
def create_sequences(data, seq_length):
    sequences = []
    target = []
    for i in range(len(data) - seq_length):
        seq = data[i:i + seq_length]
        label = data[i + seq_length]
        sequences.append(seq)
        target.append(label)
    return np.array(sequences), np.array(target)

# Set sequence length
sequence_length = 10

# Create sequences and targets
X, y = create_sequences(closing_prices_scaled, sequence_length)

# Split the data into train and test sets
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Build the neural network model
model_nn = Sequential()
model_nn.add(LSTM(50, activation='relu', input_shape=(X_train.shape[1], X_train.shape[2])))
model_nn.add(Dense(1))
model_nn.compile(optimizer='adam', loss='mse')

# Train the model
model_nn.fit(X_train, y_train, epochs=50, batch_size=32, validation_split=0.1)

# Evaluate the model
mse = model_nn.evaluate(X_test, y_test)
print(f'Mean Squared Error on Test Data: {mse}')

# Make predictions
predictions = model_nn.predict(X_test)

# Inverse transform the predictions to get actual closing prices
predicted_prices = scaler.inverse_transform(predictions)
actual_prices = scaler.inverse_transform(y_test.reshape(-1, 1))

# Get corresponding dates for test data
test_dates = full_data.index[-len(y_test):].date

# Plot actual vs predicted prices with dates
plt.figure(figsize=(12, 6))
plt.plot(test_dates, actual_prices, label='Actual Prices')
plt.plot(test_dates, predicted_prices, label='Predicted Prices')
plt.title('Actual vs Predicted Prices')
plt.xlabel('Date')
plt.ylabel('Closing Prices')
plt.legend()
plt.show()


# Calculate the Hurst exponent
hurst_exponent, _, _ = compute_Hc(full_data['Close'], kind='price', simplified=True)

print(f'Hurst Exponent: {hurst_exponent}')

# If the Hurst exponent suggests trending behavior, you can use a simple moving average for prediction
if hurst_exponent > 0.5:
    # Calculate the simple moving average
    window_size = 30  # You can adjust the window size
    sma = full_data['Close'].rolling(window=window_size).mean()

    # Plot the original closing prices and the simple moving average
    plt.figure(figsize=(12, 6))
    plt.plot(full_data.index, full_data['Close'], label='Closing Prices')
    plt.plot(full_data.index, sma, label=f'SMA ({window_size} days)')
    plt.title('Closing Prices and Simple Moving Average')
    plt.xlabel('Date')
    plt.ylabel('Closing Prices')
    plt.legend()
    plt.show()

    # Extend the time axis for prediction
    prediction_dates = pd.date_range(full_data.index[-1], periods=90, freq='D')[1:]

    # Make predictions for the first three months of 2024 using the simple moving average
    predicted_prices_sma = sma[-1]  # Assuming the last value of the SMA continues into the future

    # Introduce randomness to simulate fluctuations
    np.random.seed(42)
    noise = np.random.normal(0, 10, len(prediction_dates))  # Adjust the parameters as needed
    predicted_prices_sma_with_noise = np.full_like(prediction_dates, fill_value=predicted_prices_sma)
    predicted_prices_sma_with_noise = predicted_prices_sma_with_noise.astype(float) + noise

    # Plot predicted prices for the first three months of 2024
    plt.figure(figsize=(12, 6))
    plt.plot(full_data.index, full_data['Close'], label='Closing Prices')
    plt.plot(prediction_dates, predicted_prices_sma_with_noise, label='Predicted Prices (SMA with Noise)')
    plt.title('Predicted Prices for the First Three Months of 2024')
    plt.xlabel('Date')
    plt.ylabel('Closing Prices')
    plt.legend()
    plt.show()

else:
    print('The Hurst exponent suggests random walk behavior. Consider other analysis methods for predictions.')