Delete core/plot.py

core/plot.py

@@ -1,377 +0,0 @@
-import matplotlib.pyplot as plt
-import seaborn as sns
-import pandas as pd
-import numpy as np
-import networkx as nx
-
-
-def plot_forecast(result):
-    """Modified to show only last day backtest comparison"""
-    forecast = result["forecast"]
-    actual = result["actual"]
-    next_pred = result.get("latest_prediction", [])
-
-    # 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]
-
-    fig, ax = plt.subplots(figsize=(10, 6))
-    
-    if len(actual) == 0 or len(forecast) == 0:
-        ax.text(0.5, 0.5, "No data available for plotting", ha='center', va='center')
-        ax.set_title("Forecast vs Actual")
-        return fig
-    
-    # Create time indices
-    time_indices = np.arange(len(actual))
-    
-    # Plot most of the data normally
-    if len(actual) > 1:
-        ax.plot(time_indices[:-1], actual[:-1], label="Historical", color="blue", linewidth=2, alpha=0.7)
-        ax.plot(time_indices[:-1], forecast[:-1], label="Historical Forecast", color="orange", linestyle="--", linewidth=1, alpha=0.7)
-    
-    # Highlight only the last day comparison (backtest)
-    if len(actual) > 1 and len(forecast) > 1:
-        last_idx = len(actual) - 1
-        
-        # Create line segments for last day
-        last_actual_segment = [float(actual[last_idx-1]), float(actual[last_idx])]
-        last_forecast_segment = [float(forecast[last_idx-1]), float(forecast[last_idx])]
-        last_time_segment = [time_indices[last_idx-1], time_indices[last_idx]]
-        
-        ax.plot(last_time_segment, last_actual_segment, 
-                color="blue", linewidth=3, label="Last Day Actual")
-        ax.plot(last_time_segment, last_forecast_segment, 
-                color="red", linewidth=3, linestyle="--", label="Last Day Predicted")
-        
-        # Add markers for the last day comparison
-        ax.scatter([last_idx], [float(actual[last_idx])], color="blue", s=100, zorder=5)
-        ax.scatter([last_idx], [float(forecast[last_idx])], color="red", s=100, zorder=5)
-        
-        # Add value annotations for last day
-        actual_val = float(actual[last_idx])
-        forecast_val = float(forecast[last_idx])
-        ax.annotate(f'Actual: {actual_val:.2f}', 
-                   xy=(last_idx, actual_val), 
-                   xytext=(10, 10), textcoords='offset points',
-                   bbox=dict(boxstyle='round,pad=0.3', facecolor='blue', alpha=0.7, edgecolor='none'),
-                   color='white', fontsize=9, fontweight='bold')
-        ax.annotate(f'Predicted: {forecast_val:.2f}', 
-                   xy=(last_idx, forecast_val), 
-                   xytext=(10, -20), textcoords='offset points',
-                   bbox=dict(boxstyle='round,pad=0.3', facecolor='red', alpha=0.7, edgecolor='none'),
-                   color='white', fontsize=9, fontweight='bold')
-    elif len(actual) == 1:
-        # Handle single point case
-        ax.scatter([0], [float(actual[0])], color="blue", s=100, label="Actual", zorder=5)
-        ax.scatter([0], [float(forecast[0])], color="red", s=100, label="Predicted", zorder=5)
-
-    ax.legend()
-    ax.set_title("Forecast vs Actual (Last Day Backtest Highlighted)")
-    ax.set_xlabel("Time Index")
-    ax.set_ylabel("Value")
-    ax.grid(True, alpha=0.3)
-    plt.tight_layout()
-    return fig
-
-
-def plot_future_forecast(df, result, future_df):
-    """Enhanced to show trend lines for predictions"""
-    fig, ax = plt.subplots(figsize=(12, 7))
-    
-    # Validate and convert data
-    if df.empty or 'Date' not in df.columns or 'value' not in df.columns:
-        ax.text(0.5, 0.5, "No valid historical data available", ha='center', va='center')
-        ax.set_title("Stock Prediction: Historical Data → Future Forecast")
-        return fig
-    
-    # Plot historical data
-    dates = pd.to_datetime(df['Date'])
-    values = np.array(df['value']).flatten()
-    ax.plot(dates, values, label="Historical Data", color="blue", linewidth=2)
-    
-    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])]
-            ax.plot(connection_dates, connection_values, color="orange", linestyle=":", linewidth=1, alpha=0.7)
-        
-        # Plot forecast with fine connecting lines
-        predictions_float = [float(p) for p in predictions]
-        ax.plot(future_dates, predictions_float, 
-               label="Forecast", color="orange", linewidth=2.5, 
-               linestyle="-", marker='o', markersize=6, markerfacecolor='orange', 
-               markeredgecolor='darkorange', markeredgewidth=1)
-        
-        # Add value annotations for predictions
-        for i, val in enumerate(predictions_float):
-            ax.annotate(f"{val:.2f}", 
-                       xy=(future_dates[i], val), 
-                       xytext=(0, 15), textcoords='offset points',
-                       ha='center', va='bottom',
-                       bbox=dict(boxstyle='round,pad=0.2', facecolor='orange', alpha=0.8, edgecolor='darkorange'),
-                       color='white', fontsize=8, fontweight='bold')
-        
-        # 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]
-            
-            # Connect to actual future values
-            ax.plot(actual_future_dates, actual_future_values_float, 
-                   label="Actual Future", color="green", linewidth=2.5,
-                   linestyle="-", marker='s', markersize=6, markerfacecolor='green',
-                   markeredgecolor='darkgreen', markeredgewidth=1)
-            
-            # Add value annotations for actuals
-            for i, val in enumerate(actual_future_values_float):
-                date_val = actual_future_dates.iloc[i] if hasattr(actual_future_dates, 'iloc') else actual_future_dates[i]
-                ax.annotate(f"{val:.2f}", 
-                           xy=(date_val, val), 
-                           xytext=(0, -20), textcoords='offset points',
-                           ha='center', va='top',
-                           bbox=dict(boxstyle='round,pad=0.2', facecolor='green', alpha=0.8, edgecolor='darkgreen'),
-                           color='white', fontsize=8, fontweight='bold')
-
-    ax.legend(loc='upper left')
-    ax.set_title("Stock Prediction: Historical Data → Future Forecast", fontsize=14, fontweight='bold')
-    ax.set_xlabel("Date", fontsize=12)
-    ax.set_ylabel("Stock Price", fontsize=12)
-    ax.grid(True, alpha=0.3)
-    
-    # Improve date formatting on x-axis
-    fig.autofmt_xdate(rotation=45)
-    plt.tight_layout()
-    return fig
-
-
-def plot_metrics_precision(result):
-    fig, ax = plt.subplots(figsize=(8, 5))
-    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:
-        ax.text(0.5, 0.5, "No valid precision metrics available", ha='center', va='center')
-        ax.set_title("Precision Metrics (Model Accuracy)")
-        return fig
-    
-    sns.barplot(x=list(metrics.keys()), y=list(metrics.values()), ax=ax, palette="Blues_d")
-    for i, v in enumerate(metrics.values()):
-        ax.text(i, v + 0.01 * max(metrics.values(), default=100), f"{v:.2f}%", ha='center', va='bottom', fontsize=10)
-    
-    max_val = max(metrics.values(), default=100)
-    min_val = min(metrics.values(), default=0)
-    ax.set_ylim(min(min_val - 5, -10), max_val + 10)
-    ax.set_ylabel("Value (%)")
-    ax.set_title("Precision Metrics (Model Accuracy)")
-    ax.grid(True, axis='y')
-    plt.tight_layout()
-    return fig
-
-
-def plot_metrics_risk(result):
-    fig, ax = plt.subplots(figsize=(8, 5))
-    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:
-        ax.text(0.5, 0.5, "No valid risk metrics available", ha='center', va='center')
-        ax.set_title("Risk Metrics (Error Magnitude)")
-        return fig
-    
-    sns.barplot(x=list(metrics.keys()), y=list(metrics.values()), ax=ax, palette="Reds_d")
-    for i, v in enumerate(metrics.values()):
-        ax.text(i, v + 0.01 * max(metrics.values(), default=1), f"{v:.2f}", ha='center', va='bottom', fontsize=10)
-    
-    max_val = max(metrics.values(), default=1)
-    ax.set_ylim(0, max_val + 0.2 * max_val)
-    ax.set_ylabel("Value")
-    ax.set_title("Risk Metrics (Error Magnitude)")
-    ax.grid(True, axis='y')
-    plt.tight_layout()
-    return fig
-
-
-def plot_loss_curve(result):
-    train_losses = result.get('train_loss', [])
-    val_losses = result.get('val_loss', [])
-    fig, ax = plt.subplots(figsize=(8, 5))
-    ax.plot(train_losses, label='Train Loss', color='blue', linewidth=2)
-    if val_losses:
-        ax.plot(val_losses, label='Validation Loss', color='orange', linewidth=2)
-    ax.set_title('Training and Validation Loss Curve')
-    ax.set_xlabel('Epoch')
-    ax.set_ylabel('Loss (MSE)')
-    ax.legend()
-    ax.grid(True)
-    plt.tight_layout()
-    return fig
-
-
-def plot_model_architecture(result):
-    fig, ax = plt.subplots(figsize=(10, 6))
-    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)
-        ax.set_title("Model Architecture")
-        plt.tight_layout()
-        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)  # Use first hidden layer size
-        units_label = f"{hidden_units[0]},{hidden_units[1]}"  # Show both sizes
-    elif model_name == "CNNModel":
-        hidden_nodes = 5  # Simplified for visualization
-        units_label = f"{hidden_units} filters"  # num_filters
-    elif model_name == "TransformerModel":
-        hidden_nodes = min(hidden_units, 5)  # d_model
-        units_label = f"{hidden_units}"
-    else:
-        hidden_nodes = min(hidden_units, 5)  # hidden_size
-        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  # Fixed for 3–4 blocks
-
-        # Define components for each complex model
-        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")
-            ]
-
-        # Add nodes and positions
-        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)
-
-        # Draw the graph
-        nx.draw(G, pos, ax=ax, with_labels=False, node_color='lightblue', edge_color='gray', 
-                node_size=2000, node_shape='s', arrowsize=10)
-
-        # Add labels for each block
-        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:
-        # Node-based graph for RNN, LSTM, GRU, BiLSTM, MLP, Transformer
-        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
-
-        # Input layer
-        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)])
-
-        # Hidden layers
-        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
-        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)
-
-        # Add edges between layers
-        for layer in range(len(nodes) - 1):
-            for src in nodes[layer]:
-                for dst in nodes[layer + 1]:
-                    G.add_edge(src, dst)
-
-        # Draw the graph
-        nx.draw(G, pos, ax=ax, with_labels=False, node_color='lightblue', edge_color='gray', 
-                node_size=500, arrowsize=10)
-
-        # Add layer labels
-        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 title and annotation
-    title = f"{model_name} Architecture"
-    details = f"Dropout: {dropout:.2f}\nBatch Size: {batch_size}"
-    ax.set_title(title, fontsize=12, pad=20)
-    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