import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import matplotlib.pyplot as plt

# 生成示例時(shí)間序列數(shù)據(jù)
def generate_sine_wave(seq_length, num_samples):
    x = np.linspace(0, 4 * np.pi, seq_length)
    data = np.array([np.sin(x + np.random.uniform(0, 2 * np.pi)) for _ in range(num_samples)])
    return data

# 數(shù)據(jù)準(zhǔn)備
seq_length = 50  # 輸入序列長度
num_samples = 1000  # 樣本數(shù)
prediction_length = 1  # 預(yù)測(cè)的未來步數(shù)

data = generate_sine_wave(seq_length + prediction_length, num_samples)

# 劃分?jǐn)?shù)據(jù)：輸入和目標(biāo)
X = data[:, :-prediction_length]  # 輸入序列
y = data[:, -prediction_length:]  # 目標(biāo)值

# 轉(zhuǎn)為 PyTorch 張量
X = torch.tensor(X, dtype=torch.float32).unsqueeze(1)  # 添加通道維度 (batch, channel, seq_length)
y = torch.tensor(y, dtype=torch.float32)

# 劃分訓(xùn)練和測(cè)試集
train_size = int(0.8 * len(X))
X_train, X_test = X[:train_size], X[train_size:]
y_train, y_test = y[:train_size], y[train_size:]

# 檢查數(shù)據(jù)形狀
print("訓(xùn)練集輸入形狀:", X_train.shape)  # (batch_size, channels, seq_length)
print("訓(xùn)練集目標(biāo)形狀:", y_train.shape)  # (batch_size, prediction_length)

# 定義 1D 卷積預(yù)測(cè)模型
class Conv1DPredictor(nn.Module):
    def __init__(self, input_channels, output_size, kernel_size=3):
        super(Conv1DPredictor, self).__init__()
        self.conv1 = nn.Conv1d(in_channels=input_channels, out_channels=16, kernel_size=kernel_size, padding=1)
        self.conv2 = nn.Conv1d(in_channels=16, out_channels=32, kernel_size=kernel_size, padding=1)
        self.fc = nn.Linear(32 * seq_length, output_size)  # 全連接層，用于生成預(yù)測(cè)結(jié)果

    def forward(self, x):
        x = torch.relu(self.conv1(x))  # 第一層卷積 + 激活
        x = torch.relu(self.conv2(x))  # 第二層卷積 + 激活
        x = x.view(x.size(0), -1)  # 展平
        x = self.fc(x)  # 全連接層輸出
        return x

# 初始化模型
model = Conv1DPredictor(input_channels=1, output_size=prediction_length)
criterion = nn.MSELoss()  # 損失函數(shù)
optimizer = optim.Adam(model.parameters(), lr=0.001)  # 優(yōu)化器

# 訓(xùn)練模型
num_epochs = 50
train_losses = []

for epoch in range(num_epochs):
    model.train()
    optimizer.zero_grad()
    output = model(X_train)  # 前向傳播
    loss = criterion(output, y_train)  # 計(jì)算損失
    loss.backward()  # 反向傳播
    optimizer.step()  # 更新權(quán)重
    train_losses.append(loss.item())
    if (epoch + 1) % 10 == 0:
        print(f"Epoch [{epoch + 1}/{num_epochs}], Loss: {loss.item():.4f}")

# 測(cè)試模型
model.eval()
with torch.no_grad():
    y_pred = model(X_test)
    test_loss = criterion(y_pred, y_test)
    print(f"測(cè)試集損失: {test_loss.item():.4f}")

# 可視化結(jié)果
plt.figure(figsize=(10, 5))
plt.plot(y_test.numpy()[:50], label="True Values")
plt.plot(y_pred.numpy()[:50], label="Predictions")
plt.legend()
plt.title("Time Series Prediction")
plt.show()

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv1D, Dense, Flatten
from tensorflow.keras.optimizers import Adam
import matplotlib.pyplot as plt

# 生成示例時(shí)間序列數(shù)據(jù)
def generate_sine_wave(seq_length, num_samples):
    x = np.linspace(0, 4 * np.pi, seq_length)
    data = np.array([np.sin(x + np.random.uniform(0, 2 * np.pi)) for _ in range(num_samples)])
    return data

# 數(shù)據(jù)準(zhǔn)備
seq_length = 50  # 輸入序列長度
num_samples = 1000  # 樣本數(shù)
prediction_length = 1  # 預(yù)測(cè)未來步數(shù)

data = generate_sine_wave(seq_length + prediction_length, num_samples)

# 劃分?jǐn)?shù)據(jù)：輸入和目標(biāo)
X = data[:, :-prediction_length]  # 輸入序列
y = data[:, -prediction_length:]  # 目標(biāo)值

# 劃分訓(xùn)練集和測(cè)試集
train_size = int(0.8 * len(X))
X_train, X_test = X[:train_size], X[train_size:]
y_train, y_test = y[:train_size], y[train_size:]

# 擴(kuò)展維度以適配 Conv1D 輸入格式
X_train = X_train[..., np.newaxis]  # 轉(zhuǎn)換為 (batch_size, seq_length, channels)
X_test = X_test[..., np.newaxis]

# 檢查數(shù)據(jù)形狀
print("訓(xùn)練集輸入形狀:", X_train.shape)  # (batch_size, seq_length, channels)
print("訓(xùn)練集目標(biāo)形狀:", y_train.shape)  # (batch_size, prediction_length)

# 構(gòu)建 1D 卷積預(yù)測(cè)模型
model = Sequential([
    Conv1D(filters=16, kernel_size=3, activatinotallow='relu', padding='same', input_shape=(seq_length, 1)),
    Conv1D(filters=32, kernel_size=3, activatinotallow='relu', padding='same'),
    Flatten(),
    Dense(10, activatinotallow='relu'),
    Dense(prediction_length)  # 輸出層
])

# 編譯模型
model.compile(optimizer=Adam(learning_rate=0.001), loss='mse')

# 訓(xùn)練模型
history = model.fit(X_train, y_train, epochs=50, batch_size=32, validation_split=0.2, verbose=1)

# 測(cè)試模型
test_loss = model.evaluate(X_test, y_test, verbose=0)
print(f"測(cè)試集損失: {test_loss:.4f}")

# 預(yù)測(cè)并可視化
y_pred = model.predict(X_test)

plt.figure(figsize=(10, 5))
plt.plot(y_test[:50], label="True Values")
plt.plot(y_pred[:50], label="Predictions")
plt.legend()
plt.title("Time Series Prediction with 1D Convolution")
plt.show()