df = pd.read_csv('/workspaces/Data-Miscellany-Forum/src/多变量时序预测系列-股票预测/多输入+单输出+多步/data.csv', parse_dates=["Date"], index_col=[0])df = pd.DataFrame(df)var_num = len(df.columns)print(df)

test_split=round(len(df)*0.20)df_for_training=df[:-test_split]df_for_testing=df[-test_split:]

scaler = MinMaxScaler(feature_range=(0,1))df_for_training_scaled = scaler.fit_transform(df_for_training)df_for_testing_scaled=scaler.transform(df_for_testing)

train_dataset = TimeSeriesDataset(df_for_training_scaled, seq_len=seq_len, pred_len=pred_len)test_dataset = TimeSeriesDataset(df_for_testing_scaled, seq_len=seq_len, pred_len=pred_len)train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

class CNN_LSTM_Transformer(nn.Module):    def __init__(self, input_dim=5, cnn_channels=16, lstm_hidden=32, transformer_dim=32,                 transformer_heads=4, transformer_layers=1, pred_len=5):        super().__init__()        # CNN        self.cnn = nn.Conv1d(in_channels=input_dim, out_channels=cnn_channels, kernel_size=3, padding=1)        self.cnn_relu = nn.ReLU()        # LSTM        self.lstm = nn.LSTM(input_size=cnn_channels, hidden_size=lstm_hidden, batch_first=True)        # Transformer Encoder         encoder_layer = nn.TransformerEncoderLayer(d_model=transformer_dim, nhead=transformer_heads, batch_first=True)        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=transformer_layers)        # Projection layers        self.proj_lstm = nn.Linear(lstm_hidden, transformer_dim)        self.pred_len = pred_len        self.fc_out = nn.Linear(transformer_dim, pred_len)    def forward(self, x):        # x: [batch, seq_len, 1]        batch_size, seq_len, _ = x.shape        # CNN expects [batch, channels, seq_len]        cnn_out = self.cnn_relu(self.cnn(x.transpose(1,2)))  # [B, C, T]        cnn_out = cnn_out.transpose(1,2)  # [B, T, C]        # LSTM        lstm_out, _ = self.lstm(cnn_out)  # [B, T, hidden]        lstm_proj = self.proj_lstm(lstm_out)  # [B, T, transformer_dim]        # Transformer        trans_out = self.transformer(lstm_proj)  # [B, T, transformer_dim]        # 取最后时间步输出预测        out = self.fc_out(trans_out[:, -1, :])  # [B, pred_len]        return out.unsqueeze(-1)  # [B, pred_len, 1]

def train_model(model, dataloader, num_epochs=50, learning_rate=1e-3, device='cpu'):    optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)    criterion = nn.MSELoss()    model.train()    loss_history = []    for epoch in range(num_epochs):        epoch_losses = []        for batch_data, batch_targets in dataloader:            batch_data = batch_data.to(device)            batch_targets = batch_targets.to(device)            optimizer.zero_grad()            outputs = model(batch_data)            loss = criterion(outputs, batch_targets)            loss.backward()            optimizer.step()            epoch_losses.append(loss.item())        avg_loss = np.mean(epoch_losses)        loss_history.append(avg_loss)        if (epoch + 1) % 10 == 0:            print(f"Epoch [{epoch + 1}/{num_epochs}], Loss: {avg_loss:.4f}")    return loss_history

def evaluate_model(model, dataloader, device='cpu'):    model.eval()    preds = []    trues = []    with torch.no_grad():        for batch_data, batch_targets in dataloader:            batch_data = batch_data.to(device)            outputs = model(batch_data)            preds.append(outputs.cpu().numpy())            trues.append(batch_targets.cpu().numpy())    # print(preds.shape)    # print(trues.shape)    preds = np.concatenate(preds, axis=0).squeeze()    trues = np.concatenate(trues, axis=0).squeeze()    print(preds.shape)    print(trues.shape)    return preds, trues

plt.plot(loss_history, marker='o', color='dodgerblue', linestyle='-', linewidth=2)plt.title("Training Loss Curve")plt.xlabel("Epoch")plt.ylabel("MSE Loss")plt.tight_layout()plt.savefig('output_image1.png', dpi=300, format='png')plt.show()

plt.plot(trues, label="True Values", color='limegreen')plt.plot(preds, label="Predicted Values", color='crimson')plt.title("True vs. Predicted Values")plt.xlabel("Sample Index")plt.ylabel("Trend Value")plt.legend()plt.tight_layout()plt.savefig('output_image2.png', dpi=300, format='png')plt.show()

idx = idxplt.figure(figsize=(10,6))plt.plot(range(1, pred_len+1), trues[idx], marker='o', linewidth=3, label="True", color=vivid_colors[1])plt.plot(range(1, pred_len+1), preds[idx], marker='s', linewidth=3, label="Pred", color=vivid_colors[0])plt.fill_between(range(1,pred_len+1), preds[idx]-0.05, preds[idx]+0.05, color=vivid_colors[0], alpha=0.15, label="±0.05 band")plt.title("comparison of multi-step predictions on test samples")plt.xlabel("prediction step(horizon)")plt.ylabel("numerical value")plt.legend()plt.tight_layout()plt.savefig('output_image3.png', dpi=300, format='png')plt.show()

resid = (preds - trues).reshape(-1)sns.histplot(resid, bins=40, stat="density", color=vivid_colors[3], kde=True, alpha=0.7)plt.axvline(np.mean(resid), color=vivid_colors[0], linestyle='--', linewidth=2, label=f"Mean={np.mean(resid):.3f}")plt.title("residual distribution and KDE")plt.xlabel("residual")plt.ylabel("density")plt.legend()plt.tight_layout()plt.savefig('output_image4.png', dpi=300, format='png')plt.show()

win = 50mape_series = np.abs((preds - trues)/(np.abs(trues)+1e-8)).mean(axis=1)rolling = np.array([mape_series[i:i+win].mean() for i in range(0, len(mape_series)-win+1)])plt.figure(figsize=(12,5))plt.plot(rolling, linewidth=3, color=vivid_colors[6])plt.title(f"Rolling MAPE (window={win})")plt.xlabel("start sample index of window")plt.ylabel("MAPE")plt.tight_layout()plt.savefig('output_image5.png', dpi=300, format='png')plt.show()

def evaluate_metrics(y_true, y_pred):    y_true = y_true.reshape(-1)    y_pred = y_pred.reshape(-1)    eps = 1e-8    mse = np.mean((y_true - y_pred)**2)    rmse = np.sqrt(mse + eps)    mae = np.mean(np.abs(y_true - y_pred))    mape = np.mean(np.abs((y_true - y_pred)/(np.abs(y_true)+eps)))    smape = np.mean(2*np.abs(y_true - y_pred)/(np.abs(y_true)+np.abs(y_pred)+eps))    # R2    ss_res = np.sum((y_true - y_pred)**2)    ss_tot = np.sum((y_true - np.mean(y_true))**2) + eps    r2 = 1 - ss_res/ss_tot    # Pearson    corr = np.corrcoef(y_true, y_pred)[0,1]    return dict(MSE=mse, RMSE=rmse, MAE=mae, MAPE=mape, sMAPE=smape, R2=r2, Corr=corr)