319 lines
11 KiB
Python
319 lines
11 KiB
Python
import numpy as np
|
|
import dash
|
|
from dash import dcc, html
|
|
import plotly.graph_objs as go
|
|
import threading
|
|
|
|
|
|
def display_actual_vs_predicted(y_test, test_preds, timestamps, n_plot=200):
|
|
import plotly.offline as pyo
|
|
n_plot = min(n_plot, len(y_test))
|
|
plot_indices = timestamps[:n_plot]
|
|
actual = y_test[:n_plot]
|
|
predicted = test_preds[:n_plot]
|
|
|
|
trace_actual = go.Scatter(x=plot_indices, y=actual, mode='lines', name='Actual')
|
|
trace_predicted = go.Scatter(x=plot_indices, y=predicted, mode='lines', name='Predicted')
|
|
data = [trace_actual, trace_predicted]
|
|
layout = go.Layout(
|
|
title='Actual vs. Predicted BTC Close Prices (Test Set)',
|
|
xaxis={'title': 'Timestamp'},
|
|
yaxis={'title': 'BTC Close Price'},
|
|
legend={'x': 0, 'y': 1},
|
|
margin={'l': 40, 'b': 40, 't': 40, 'r': 10},
|
|
hovermode='closest'
|
|
)
|
|
fig = go.Figure(data=data, layout=layout)
|
|
pyo.plot(fig, auto_open=False)
|
|
|
|
def plot_target_distribution(y_train, y_test):
|
|
import plotly.offline as pyo
|
|
trace_train = go.Histogram(
|
|
x=y_train,
|
|
nbinsx=100,
|
|
opacity=0.5,
|
|
name='Train',
|
|
marker=dict(color='blue')
|
|
)
|
|
trace_test = go.Histogram(
|
|
x=y_test,
|
|
nbinsx=100,
|
|
opacity=0.5,
|
|
name='Test',
|
|
marker=dict(color='orange')
|
|
)
|
|
data = [trace_train, trace_test]
|
|
layout = go.Layout(
|
|
title='Distribution of Target Variable (Close Price)',
|
|
xaxis=dict(title='BTC Close Price'),
|
|
yaxis=dict(title='Frequency'),
|
|
barmode='overlay'
|
|
)
|
|
fig = go.Figure(data=data, layout=layout)
|
|
pyo.plot(fig, auto_open=False)
|
|
|
|
def plot_predicted_vs_actual_log_returns(y_test, test_preds, timestamps=None, n_plot=200):
|
|
import plotly.offline as pyo
|
|
import plotly.graph_objs as go
|
|
n_plot = min(n_plot, len(y_test))
|
|
actual = y_test[:n_plot]
|
|
predicted = test_preds[:n_plot]
|
|
if timestamps is not None:
|
|
x_axis = timestamps[:n_plot]
|
|
x_label = 'Timestamp'
|
|
else:
|
|
x_axis = list(range(n_plot))
|
|
x_label = 'Index'
|
|
|
|
# Line plot: Actual vs Predicted over time
|
|
trace_actual = go.Scatter(x=x_axis, y=actual, mode='lines', name='Actual')
|
|
trace_predicted = go.Scatter(x=x_axis, y=predicted, mode='lines', name='Predicted')
|
|
data_line = [trace_actual, trace_predicted]
|
|
layout_line = go.Layout(
|
|
title='Actual vs. Predicted Log Returns (Test Set)',
|
|
xaxis={'title': x_label},
|
|
yaxis={'title': 'Log Return'},
|
|
legend={'x': 0, 'y': 1},
|
|
margin={'l': 40, 'b': 40, 't': 40, 'r': 10},
|
|
hovermode='closest'
|
|
)
|
|
fig_line = go.Figure(data=data_line, layout=layout_line)
|
|
pyo.plot(fig_line, filename='charts/log_return_line_plot.html', auto_open=False)
|
|
|
|
# Scatter plot: Predicted vs Actual
|
|
trace_scatter = go.Scatter(
|
|
x=actual,
|
|
y=predicted,
|
|
mode='markers',
|
|
name='Predicted vs Actual',
|
|
opacity=0.5
|
|
)
|
|
# Diagonal reference line
|
|
min_val = min(np.min(actual), np.min(predicted))
|
|
max_val = max(np.max(actual), np.max(predicted))
|
|
trace_diag = go.Scatter(
|
|
x=[min_val, max_val],
|
|
y=[min_val, max_val],
|
|
mode='lines',
|
|
name='Ideal',
|
|
line=dict(dash='dash', color='red')
|
|
)
|
|
data_scatter = [trace_scatter, trace_diag]
|
|
layout_scatter = go.Layout(
|
|
title='Predicted vs Actual Log Returns (Scatter)',
|
|
xaxis={'title': 'Actual Log Return'},
|
|
yaxis={'title': 'Predicted Log Return'},
|
|
showlegend=True,
|
|
margin={'l': 40, 'b': 40, 't': 40, 'r': 10},
|
|
hovermode='closest'
|
|
)
|
|
fig_scatter = go.Figure(data=data_scatter, layout=layout_scatter)
|
|
pyo.plot(fig_scatter, filename='charts/log_return_scatter_plot.html', auto_open=False)
|
|
|
|
def plot_predicted_vs_actual_prices(actual_prices, predicted_prices, timestamps=None, n_plot=200):
|
|
import plotly.offline as pyo
|
|
import plotly.graph_objs as go
|
|
n_plot = min(n_plot, len(actual_prices))
|
|
actual = actual_prices[:n_plot]
|
|
predicted = predicted_prices[:n_plot]
|
|
if timestamps is not None:
|
|
x_axis = timestamps[:n_plot]
|
|
x_label = 'Timestamp'
|
|
else:
|
|
x_axis = list(range(n_plot))
|
|
x_label = 'Index'
|
|
|
|
# Line plot: Actual vs Predicted over time
|
|
trace_actual = go.Scatter(x=x_axis, y=actual, mode='lines', name='Actual Price')
|
|
trace_predicted = go.Scatter(x=x_axis, y=predicted, mode='lines', name='Predicted Price')
|
|
data_line = [trace_actual, trace_predicted]
|
|
layout_line = go.Layout(
|
|
title='Actual vs. Predicted BTC Prices (Test Set)',
|
|
xaxis={'title': x_label},
|
|
yaxis={'title': 'BTC Price'},
|
|
legend={'x': 0, 'y': 1},
|
|
margin={'l': 40, 'b': 40, 't': 40, 'r': 10},
|
|
hovermode='closest'
|
|
)
|
|
fig_line = go.Figure(data=data_line, layout=layout_line)
|
|
pyo.plot(fig_line, filename='charts/price_line_plot.html', auto_open=False)
|
|
|
|
# Scatter plot: Predicted vs Actual
|
|
trace_scatter = go.Scatter(
|
|
x=actual,
|
|
y=predicted,
|
|
mode='markers',
|
|
name='Predicted vs Actual',
|
|
opacity=0.5
|
|
)
|
|
# Diagonal reference line
|
|
min_val = min(np.min(actual), np.min(predicted))
|
|
max_val = max(np.max(actual), np.max(predicted))
|
|
trace_diag = go.Scatter(
|
|
x=[min_val, max_val],
|
|
y=[min_val, max_val],
|
|
mode='lines',
|
|
name='Ideal',
|
|
line=dict(dash='dash', color='red')
|
|
)
|
|
data_scatter = [trace_scatter, trace_diag]
|
|
layout_scatter = go.Layout(
|
|
title='Predicted vs Actual Prices (Scatter)',
|
|
xaxis={'title': 'Actual Price'},
|
|
yaxis={'title': 'Predicted Price'},
|
|
showlegend=True,
|
|
margin={'l': 40, 'b': 40, 't': 40, 'r': 10},
|
|
hovermode='closest'
|
|
)
|
|
fig_scatter = go.Figure(data=data_scatter, layout=layout_scatter)
|
|
pyo.plot(fig_scatter, filename='charts/price_scatter_plot.html', auto_open=False)
|
|
|
|
def plot_prediction_error_distribution(predicted_prices, actual_prices, nbins=100, prefix=""):
|
|
"""
|
|
Plots the distribution of signed prediction errors between predicted and actual prices,
|
|
coloring negative errors (under-prediction) and positive errors (over-prediction) differently.
|
|
"""
|
|
import plotly.offline as pyo
|
|
import plotly.graph_objs as go
|
|
errors = np.array(predicted_prices) - np.array(actual_prices)
|
|
|
|
# Separate negative and positive errors
|
|
neg_errors = errors[errors < 0]
|
|
pos_errors = errors[errors >= 0]
|
|
|
|
# Calculate common bin edges
|
|
min_error = np.min(errors)
|
|
max_error = np.max(errors)
|
|
bin_edges = np.linspace(min_error, max_error, nbins + 1)
|
|
xbins = dict(start=min_error, end=max_error, size=(max_error - min_error) / nbins)
|
|
|
|
trace_neg = go.Histogram(
|
|
x=neg_errors,
|
|
opacity=0.75,
|
|
marker=dict(color='blue'),
|
|
name='Negative Error (Under-prediction)',
|
|
xbins=xbins
|
|
)
|
|
trace_pos = go.Histogram(
|
|
x=pos_errors,
|
|
opacity=0.75,
|
|
marker=dict(color='orange'),
|
|
name='Positive Error (Over-prediction)',
|
|
xbins=xbins
|
|
)
|
|
layout = go.Layout(
|
|
title='Distribution of Prediction Errors (Signed)',
|
|
xaxis=dict(title='Prediction Error (Predicted - Actual)'),
|
|
yaxis=dict(title='Frequency'),
|
|
barmode='overlay',
|
|
bargap=0.05
|
|
)
|
|
fig = go.Figure(data=[trace_neg, trace_pos], layout=layout)
|
|
filename = f'charts/{prefix}_prediction_error_distribution.html'
|
|
pyo.plot(fig, filename=filename, auto_open=False)
|
|
|
|
def plot_directional_accuracy(actual_prices, predicted_prices, timestamps=None, n_plot=200):
|
|
"""
|
|
Plots the directional accuracy of predictions compared to actual price movements.
|
|
Shows whether the predicted direction matches the actual direction of price movement.
|
|
|
|
Args:
|
|
actual_prices: Array of actual price values
|
|
predicted_prices: Array of predicted price values
|
|
timestamps: Optional array of timestamps for x-axis
|
|
n_plot: Number of points to plot (default 200, plots last n_plot points)
|
|
"""
|
|
import plotly.graph_objs as go
|
|
import plotly.offline as pyo
|
|
import numpy as np
|
|
|
|
# Calculate price changes
|
|
actual_changes = np.diff(actual_prices)
|
|
predicted_changes = np.diff(predicted_prices)
|
|
|
|
# Determine if directions match
|
|
actual_direction = np.sign(actual_changes)
|
|
predicted_direction = np.sign(predicted_changes)
|
|
correct_direction = actual_direction == predicted_direction
|
|
|
|
# Get last n_plot points
|
|
actual_changes = actual_changes[-n_plot:]
|
|
predicted_changes = predicted_changes[-n_plot:]
|
|
correct_direction = correct_direction[-n_plot:]
|
|
|
|
if timestamps is not None:
|
|
x_values = timestamps[1:] # Skip first since we took diff
|
|
x_values = x_values[-n_plot:] # Get last n_plot points
|
|
else:
|
|
x_values = list(range(len(actual_changes)))
|
|
|
|
# Create traces for correct and incorrect predictions
|
|
correct_trace = go.Scatter(
|
|
x=np.array(x_values)[correct_direction],
|
|
y=actual_changes[correct_direction],
|
|
mode='markers',
|
|
name='Correct Direction',
|
|
marker=dict(color='green', size=8)
|
|
)
|
|
|
|
incorrect_trace = go.Scatter(
|
|
x=np.array(x_values)[~correct_direction],
|
|
y=actual_changes[~correct_direction],
|
|
mode='markers',
|
|
name='Incorrect Direction',
|
|
marker=dict(color='red', size=8)
|
|
)
|
|
|
|
# Calculate accuracy percentage
|
|
accuracy = np.mean(correct_direction) * 100
|
|
|
|
layout = go.Layout(
|
|
title=f'Directional Accuracy (Overall: {accuracy:.1f}%)',
|
|
xaxis=dict(title='Time' if timestamps is not None else 'Sample'),
|
|
yaxis=dict(title='Price Change'),
|
|
showlegend=True
|
|
)
|
|
|
|
fig = go.Figure(data=[correct_trace, incorrect_trace], layout=layout)
|
|
pyo.plot(fig, filename='charts/directional_accuracy.html', auto_open=False)
|
|
|
|
def plot_direction_transition_heatmap(actual_prices, predicted_prices, prefix=""):
|
|
"""
|
|
Plots a heatmap showing the frequency of each (actual, predicted) direction pair.
|
|
"""
|
|
import numpy as np
|
|
import plotly.graph_objs as go
|
|
import plotly.offline as pyo
|
|
|
|
# Calculate directions
|
|
actual_direction = np.sign(np.diff(actual_prices))
|
|
predicted_direction = np.sign(np.diff(predicted_prices))
|
|
|
|
# Build 3x3 matrix: rows=actual, cols=predicted, values=counts
|
|
# Map -1 -> 0, 0 -> 1, 1 -> 2 for indexing
|
|
mapping = {-1: 0, 0: 1, 1: 2}
|
|
matrix = np.zeros((3, 3), dtype=int)
|
|
for a, p in zip(actual_direction, predicted_direction):
|
|
matrix[mapping[a], mapping[p]] += 1
|
|
|
|
# Axis labels
|
|
directions = ['Down (-1)', 'No Change (0)', 'Up (+1)']
|
|
|
|
# Plot heatmap
|
|
heatmap = go.Heatmap(
|
|
z=matrix,
|
|
x=directions, # predicted
|
|
y=directions, # actual
|
|
colorscale='Viridis',
|
|
colorbar=dict(title='Count')
|
|
)
|
|
layout = go.Layout(
|
|
title='Direction Prediction Transition Matrix',
|
|
xaxis=dict(title='Predicted Direction'),
|
|
yaxis=dict(title='Actual Direction')
|
|
)
|
|
fig = go.Figure(data=[heatmap], layout=layout)
|
|
filename = f'charts/{prefix}_direction_transition_heatmap.html'
|
|
pyo.plot(fig, filename=filename, auto_open=False)
|
|
|