core/plot.py

import plotly.graph_objects as go
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
import seaborn as sns
import networkx as nx

def plot_forecast(result):
    """Interactive backtest plot with zoom and pan functionality using Plotly"""
    forecast = result["forecast"]
    actual = result["actual"]

    # Convert to numpy arrays and flatten if needed
    forecast = np.array(forecast).flatten()
    actual = np.array(actual).flatten()
    
    # Ensure both arrays have the same length
    min_len = min(len(forecast), len(actual))
    forecast = forecast[:min_len]
    actual = actual[:min_len]

    # Create time indices
    time_indices = np.arange(len(actual))
    
    # Initialize Plotly figure
    fig = go.Figure()

    if len(actual) == 0 or len(forecast) == 0:
        fig.add_annotation(
            x=0.5, y=0.5, xref="paper", yref="paper",
            text="No data available for plotting",
            showarrow=False, font=dict(size=12)
        )
        return fig

    # Plot full historical actual
    fig.add_trace(go.Scatter(
        x=time_indices, y=actual,
        mode='lines', name="Historical Actual",
        line=dict(color="blue", width=2), opacity=0.7
    ))
    
    # Plot full historical forecast
    fig.add_trace(go.Scatter(
        x=time_indices, y=forecast,
        mode='lines', name="Historical Forecast",
        line=dict(color="orange", width=2, dash="dash"), opacity=0.7
    ))
    
    if len(actual) > 1 and len(forecast) > 1:
        last_idx = len(actual) - 1
        
        # Highlight last day actual segment
        last_actual_segment = [float(actual[last_idx-1]), float(actual[last_idx])]
        last_time_segment = [time_indices[last_idx-1], time_indices[last_idx]]
        fig.add_trace(go.Scatter(
            x=last_time_segment, y=last_actual_segment,
            mode='lines', name="Last Day Actual",
            line=dict(color="blue", width=4), showlegend=False
        ))
        
        # Add markers for last day comparison
        fig.add_trace(go.Scatter(
            x=[last_idx], y=[float(actual[last_idx])],
            mode='markers', name="Last Day Actual",
            marker=dict(color="blue", size=10, line=dict(color="darkblue", width=2)),
            showlegend=False
        ))
        fig.add_trace(go.Scatter(
            x=[last_idx], y=[float(forecast[last_idx])],
            mode='markers', name="Last Day Predicted",
            marker=dict(color="red", size=10, line=dict(color="darkred", width=2)),
            showlegend=False
        ))
        
        # Add value annotations for last day
        actual_val = float(actual[last_idx])
        forecast_val = float(forecast[last_idx])
        fig.add_annotation(
            x=last_idx, y=actual_val,
            text=f"Actual: {actual_val:.2f}",
            showarrow=True, arrowhead=1, ax=20, ay=-30,
            font=dict(size=10, color="white"),
            bgcolor="blue", opacity=0.8, bordercolor="darkblue"
        )
        fig.add_annotation(
            x=last_idx, y=forecast_val,
            text=f"Predicted: {forecast_val:.2f}",
            showarrow=True, arrowhead=1, ax=20, ay=30,
            font=dict(size=10, color="white"),
            bgcolor="red", opacity=0.8, bordercolor="darkred"
        )
    elif len(actual) == 1:
        # Handle single point case
        fig.add_trace(go.Scatter(
            x=[0], y=[float(actual[0])],
            mode='markers', name="Actual",
            marker=dict(color="blue", size=10), showlegend=False
        ))
        fig.add_trace(go.Scatter(
            x=[0], y=[float(forecast[0])],
            mode='markers', name="Predicted",
            marker=dict(color="red", size=10), showlegend=False
        ))

    # Configure layout
    fig.update_layout(
        xaxis_title="Time Index",
        yaxis_title="Value",
        showlegend=True,
        legend=dict(
            orientation="h",
            yanchor="bottom",
            y=1.1,
            xanchor="center",
            x=0.5
        ),
        hovermode="x unified",
        plot_bgcolor="white",
        grid=dict(rows=1, columns=1),
        xaxis=dict(showgrid=True, gridcolor="rgba(0,0,0,0.1)", gridwidth=0.8),
        yaxis=dict(showgrid=True, gridcolor="rgba(0,0,0,0.1)", gridwidth=0.8),
        margin=dict(t=50)  # Reduced top margin to accommodate legend
    )

    return fig


def plot_future_forecast(df, result, future_df):
    """Interactive future forecast plot with zoom, pan and hover functionality using Plotly"""
    # Initialize Plotly figure
    fig = go.Figure()

    # Validate and convert data
    if df.empty or 'Date' not in df.columns or 'value' not in df.columns:
        fig.add_annotation(
            x=0.5, y=0.5, xref="paper", yref="paper",
            text="No valid historical data available",
            showarrow=False, font=dict(size=12)
        )
        return fig
    
    # Plot historical data
    dates = pd.to_datetime(df['Date'])
    values = np.array(df['value']).flatten()
    fig.add_trace(go.Scatter(
        x=dates, y=values,
        mode='lines', name="Historical Data",
        line=dict(color="blue", width=2.5), opacity=0.9
    ))
    
    if "latest_prediction" in result and len(result["latest_prediction"]) > 0:
        # Convert predictions to flat array
        predictions = np.array(result["latest_prediction"]).flatten()
        
        # Create future dates
        last_date = dates.iloc[-1] if len(dates) > 0 else pd.Timestamp.now()
        horizon = len(predictions)
        
        try:
            future_dates = pd.date_range(start=last_date + pd.Timedelta(days=1), periods=horizon, freq='B')
        except:
            # Fallback to daily frequency if business day fails
            future_dates = pd.date_range(start=last_date + pd.Timedelta(days=1), periods=horizon, freq='D')
        
        if len(values) > 0 and len(predictions) > 0:
            # Create connection from last historical point to first prediction
            connection_dates = [last_date, future_dates[0]]
            connection_values = [float(values[-1]), float(predictions[0])]
            fig.add_trace(go.Scatter(
                x=connection_dates, y=connection_values,
                mode='lines', name="Connection",
                line=dict(color="orange", width=2, dash="dot"), opacity=0.7, showlegend=False
            ))
        
        # Plot forecast
        predictions_float = [float(p) for p in predictions]
        fig.add_trace(go.Scatter(
            x=future_dates, y=predictions_float,
            mode='lines+markers', name="Forecast",
            line=dict(color="orange", width=3),
            marker=dict(size=8, color="orange", line=dict(color="darkorange", width=2)),
            opacity=0.9
        ))
        
        # Plot actual future values if available
        if not future_df.empty and "future_actuals" in result and 'Date' in future_df.columns and 'value' in future_df.columns:
            actual_future_dates = pd.to_datetime(future_df['Date'])
            actual_future_values = np.array(future_df['value']).flatten()
            actual_future_values_float = [float(v) for v in actual_future_values]
            
            fig.add_trace(go.Scatter(
                x=actual_future_dates, y=actual_future_values_float,
                mode='lines+markers', name="Actual Future",
                line=dict(color="green", width=3),
                marker=dict(size=8, color="green", line=dict(color="darkgreen", width=2)),
                opacity=0.9
            ))

    # Configure layout
    fig.update_layout(
        xaxis_title="Date",
        yaxis_title="Stock Price",
        showlegend=True,
        legend=dict(
            orientation="h",
            yanchor="bottom",
            y=1.1,
            xanchor="center",
            x=0.5
        ),
        hovermode="x unified",
        plot_bgcolor="white",
        grid=dict(rows=1, columns=1),
        xaxis=dict(showgrid=True, gridcolor="rgba(0,0,0,0.1)", gridwidth=0.8),
        yaxis=dict(showgrid=True, gridcolor="rgba(0,0,0,0.1)", gridwidth=0.8),
        margin=dict(t=50)
    )

    return fig


def plot_metrics_precision(result):
    """Plot precision metrics using Plotly"""
    metrics = {k: v for k, v in result['metrics'].items() if k in ['R² (%)', 'Explained Variance (%)', 'MDA (%)'] and v is not None}
    if not metrics:
        fig = go.Figure()
        fig.add_annotation(
            x=0.5, y=0.5, xref="paper", yref="paper",
            text="No valid precision metrics available",
            showarrow=False, font=dict(size=12)
        )
        return fig
    
    # Create bar plot
    fig = go.Figure()
    fig.add_trace(go.Bar(
        x=list(metrics.keys()),
        y=list(metrics.values()),
        marker_color=sns.color_palette("Blues_d", len(metrics)).as_hex(),
        text=[f"{v:.2f}%" for v in metrics.values()],
        textposition='auto'
    ))
    
    # Configure layout
    max_val = max(metrics.values(), default=100)
    min_val = min(metrics.values(), default=0)
    fig.update_layout(
        yaxis_title="Value (%)",
        showlegend=False,
        plot_bgcolor="white",
        yaxis=dict(range=[min(min_val - 5, -10), max_val + 10], showgrid=True, gridcolor="rgba(0,0,0,0.1)", gridwidth=0.8),
        xaxis=dict(showgrid=False),
        margin=dict(t=50)
    )

    return fig


def plot_metrics_risk(result):
    """Plot risk metrics using Plotly"""
    metrics = {k: v for k, v in result['metrics'].items() if k in ['RMSE', 'MAE', 'MAPE (%)', 'MASE'] and v is not None}
    if not metrics:
        fig = go.Figure()
        fig.add_annotation(
            x=0.5, y=0.5, xref="paper", yref="paper",
            text="No valid risk metrics available",
            showarrow=False, font=dict(size=12)
        )
        return fig
    
    # Create bar plot
    fig = go.Figure()
    fig.add_trace(go.Bar(
        x=list(metrics.keys()),
        y=list(metrics.values()),
        marker_color=sns.color_palette("Reds_d", len(metrics)).as_hex(),
        text=[f"{v:.2f}" for v in metrics.values()],
        textposition='auto'
    ))
    
    # Configure layout
    max_val = max(metrics.values(), default=1)
    fig.update_layout(
        yaxis_title="Value",
        showlegend=False,
        plot_bgcolor="white",
        yaxis=dict(range=[0, max_val + 0.2 * max_val], showgrid=True, gridcolor="rgba(0,0,0,0.1)", gridwidth=0.8),
        xaxis=dict(showgrid=False),
        margin=dict(t=50)
    )

    return fig


def plot_loss_curve(result):
    """Plot loss curve using Plotly"""
    train_losses = result.get('train_loss', [])
    val_losses = result.get('val_loss', [])
    
    fig = go.Figure()
    fig.add_trace(go.Scatter(
        x=list(range(len(train_losses))), y=train_losses,
        mode='lines', name="Train Loss",
        line=dict(color="blue", width=2)
    ))
    if val_losses:
        fig.add_trace(go.Scatter(
            x=list(range(len(val_losses))), y=val_losses,
            mode='lines', name="Validation Loss",
            line=dict(color="orange", width=2)
        ))
    
    # Configure layout
    fig.update_layout(
        xaxis_title="Epoch",
        yaxis_title="Loss (MSE)",
        showlegend=True,
        legend=dict(
            orientation="h",
            yanchor="bottom",
            y=1.1,
            xanchor="center",
            x=0.5
        ),
        plot_bgcolor="white",
        grid=dict(rows=1, columns=1),
        xaxis=dict(showgrid=True, gridcolor="rgba(0,0,0,0.1)", gridwidth=0.8),
        yaxis=dict(showgrid=True, gridcolor="rgba(0,0,0,0.1)", gridwidth=0.8),
        margin=dict(t=50)
    )

    return fig


def plot_model_architecture(result):
    """Plot model architecture using matplotlib (static, as Plotly is less suited for network graphs)"""
    fig = plt.figure(figsize=(10, 6))
    ax = fig.add_subplot(111)
    ax.axis('off')
    G = nx.DiGraph()

    if "architecture" not in result:
        ax.text(0.5, 0.5, "No architecture details available", ha='center', va='center', fontsize=12)
        return fig

    arch = result["architecture"]
    model_name = arch["model_name"]
    num_layers = arch["num_layers"]
    hidden_units = arch["hidden_units"]
    dropout = arch["dropout"]
    batch_size = arch["batch_size"]
    input_size = arch["input_size"]
    output_size = arch["output_size"]

    # Handle model-specific hidden units for visualization
    if model_name == "MLPModel":
        hidden_nodes = min(hidden_units[0], 5)
        units_label = f"{hidden_units[0]},{hidden_units[1]}"
    elif model_name == "CNNModel":
        hidden_nodes = 5
        units_label = f"{hidden_units} filters"
    elif model_name == "TransformerModel":
        hidden_nodes = min(hidden_units, 5)
        units_label = f"{hidden_units}"
    else:
        hidden_nodes = min(hidden_units, 5)
        units_label = f"{hidden_units}"

    # Simplified block diagram for complex models
    if model_name in ["CNNModel", "HybridModel", "CNN_GRU"]:
        G = nx.DiGraph()
        pos = {}
        nodes = []
        y_pos = 0.5
        layer_width = 1.0 / 4

        if model_name == "CNNModel":
            components = [
                ("Input", f"{input_size} units"),
                ("Conv1D", f"{hidden_units} filters"),
                ("MaxPool", ""),
                ("Output", f"{output_size} units")
            ]
        elif model_name == "HybridModel":
            components = [
                ("Input", f"{input_size} units"),
                ("Conv1D", "32 filters"),
                (f"BiLSTM ({num_layers} layers)", f"{hidden_units*2} units"),
                ("Output", f"{output_size} units")
            ]
        elif model_name == "CNN_GRU":
            components = [
                ("Input", f"{input_size} units"),
                ("Conv1D", "32 filters"),
                (f"GRU ({num_layers} layers)", f"{hidden_units} units"),
                ("Output", f"{output_size} units")
            ]

        for i, (comp, label) in enumerate(components):
            G.add_node(comp, layer=comp)
            pos[comp] = (i * layer_width, y_pos)
            nodes.append([comp])
            if i > 0:
                G.add_edge(components[i-1][0], comp)

        nx.draw(G, pos, ax=ax, with_labels=False, node_color='lightblue', edge_color='gray', 
                node_size=2000, node_shape='s', arrowsize=10)

        for node, (x, y) in pos.items():
            label = [comp[1] for comp in components if comp[0] == node][0]
            ax.text(x, y + 0.05, f"{node}\n{label}", ha='center', va='bottom', fontsize=8,
                    bbox=dict(facecolor='white', alpha=0.8, edgecolor='black'))

    else:
        max_nodes_display = 5
        input_nodes = min(input_size, max_nodes_display)
        output_nodes = min(output_size, max_nodes_display)

        nodes = []
        pos = {}
        layer_width = 1.0 / (num_layers + 2)
        y_pos = 0.5

        for i in range(input_nodes):
            node = f"input_{i}"
            G.add_node(node, layer="input")
            pos[node] = (0, y_pos + (i - input_nodes / 2) * 0.1)
        nodes.append([f"input_{i}" for i in range(input_nodes)])

        for layer in range(num_layers):
            layer_nodes = []
            for i in range(hidden_nodes):
                node = f"hidden_{layer}_{i}"
                G.add_node(node, layer=f"hidden_{layer+1}")
                pos[node] = ((layer + 1) * layer_width, y_pos + (i - hidden_nodes / 2) * 0.1)
                layer_nodes.append(node)
            nodes.append(layer_nodes)

        output_layer_nodes = []
        for i in range(output_nodes):
            node = f"output_{i}"
            G.add_node(node, layer="output")
            pos[node] = ((num_layers + 1) * layer_width, y_pos + (i - output_nodes / 2) * 0.1)
            output_layer_nodes.append(node)
        nodes.append(output_layer_nodes)

        for layer in range(len(nodes) - 1):
            for src in nodes[layer]:
                for dst in nodes[layer + 1]:
                    G.add_edge(src, dst)

        nx.draw(G, pos, ax=ax, with_labels=False, node_color='lightblue', edge_color='gray', 
                node_size=500, arrowsize=10)

        for node in G.nodes(data=True):
            layer = node[1]['layer']
            x, y = pos[node[0]]
            if layer.startswith("hidden"):
                label = f"Layer {layer.split('_')[1]}: {units_label} units"
            elif layer == "input":
                label = f"Input: {input_size} units"
            elif layer == "output":
                label = f"Output: {output_size} units"
            ax.text(x, y + 0.05, label, ha='center', va='bottom', fontsize=8)

    # Add model details as annotation
    details = f"Dropout: {dropout:.2f}\nBatch Size: {batch_size}"
    ax.text(0.5, 0.05, details, ha='center', va='bottom', fontsize=10, transform=ax.transAxes,
            bbox=dict(facecolor='white', alpha=0.8, edgecolor='black'))

    plt.tight_layout()
    return fig