Bagian 2: Membangun Model Sekuens

Supervised learning membutuhkan pasangan (X, y), sedangkan deret waktu hanyalah satu garis nilai. Bagian ini menyelesaikan transformasinya lewat sliding window — jendela 20 nilai menjadi masukan dan nilai ke-21 menjadi target — kemudian merakit LSTM yang membaca jendela tersebut langkah demi langkah. Dua keputusan desain diambil sekaligus di tahap ini: pembagian dilakukan pada deret mentah secara temporal sebelum windowing, dan normalisasi memakai statistik segmen latih saja. Keduanya ditanam di modul dataset.py agar keseluruhan proyek tinggal memanggil satu fungsi.

Dari Deret ke Dataset Supervised #

Modul dataset.py menyediakan fungsi kecil yang masing-masing dapat diuji terpisah:

Fungsi	Tugas
`muat_deret`	baca `data/deret.csv` menjadi array 400 nilai
`bagi_deret`	potong deret mentah menjadi latih 280 / validasi 60 / uji 60
`statistik` + `normalisasi`	hitung mu/sigma dari latih; terapkan ke semua segmen
`buat_window`	potong satu segmen menjadi X `(N, W, 1)` dan y `(N,)`
`siapkan_dataset`	rangkai semuanya, kembalikan dictionary lengkap

flowchart LR A["deret.csv 400 titik"] --> B["bagi temporal 280 / 60 / 60"] B --> C["normalisasi statistik latih"] C --> D["window W=20 per segmen"] D --> E["X (N, 20, 1) y (N,)"]

Pembagian dilakukan sebelum windowing dengan alasan yang jarang dijelaskan: jika windowing dilakukan pada deret utuh lalu jendelanya dibagi, jendela pertama segmen validasi memuat 19 nilai terakhir segmen latih — masukan uji yang sebagian berasal dari masa “sekarang” model. Dengan membagi deret mentah lebih dulu, setiap jendela sepenuhnya lahir dari segmennya sendiri.

python
Modul dataset (src/dataset.py)
"""Dataset supervised dari deret waktu: windowing, pembagian temporal,
dan normalisasi.

Tiga fungsi utama:
1. muat_deret      -- baca data/deret.csv menjadi array NumPy.
2. bagi_deret      -- potong deret mentah menjadi latih/validasi/uji
                      SECARA TEMPORAL (tanpa pengacakan).
3. buat_window     -- potong satu segmen deret menjadi pasangan
                      (X, y): jendela W langkah -> nilai berikutnya.

Pembagian dilakukan pada deret mentah sebelum windowing sehingga tidak
ada jendela yang menjembatani dua segmen: seluruh nilai masukan sebuah
sampel uji benar-benar berasal dari segmen uji.
"""

import csv
from pathlib import Path

import numpy as np

BERKAS_DATA = Path(__file__).resolve().parent.parent / "data" / "deret.csv"
PANJANG_LATIH = 280   # t = 0..279
PANJANG_VALIDASI = 60  # t = 280..339
SISA_UJI = 60          # t = 340..399


def muat_deret(berkas=BERKAS_DATA):
    """Baca CSV dua kolom (waktu, nilai) menjadi array nilai."""
    nilai = []
    with open(berkas, newline="", encoding="utf-8") as f:
        pembaca = csv.DictReader(f)
        for baris in pembaca:
            nilai.append(float(baris["nilai"]))
    return np.array(nilai)


def bagi_deret(deret, panjang_latih=PANJANG_LATIH,
               panjang_validasi=PANJANG_VALIDASI):
    """Bagi deret mentah menjadi tiga segmen berurutan tanpa diacak."""
    latih = deret[:panjang_latih]
    validasi = deret[panjang_latih:panjang_latih + panjang_validasi]
    uji = deret[panjang_latih + panjang_validasi:]
    return latih, validasi, uji


def statistik(deret):
    """Kembalikan (rata-rata, deviasi baku) satu deret."""
    return float(deret.mean()), float(deret.std())


def normalisasi(deret, mu, sigma):
    """Ubah deret ke skala nol-rata-rata satuan-deviasi."""
    return (deret - mu) / sigma


def denormalisasi(nilai, mu, sigma):
    """Kembalikan nilai ter-normalisasi ke satuan asli."""
    return nilai * sigma + mu


def buat_window(segmen, panjang_window):
    """Potong satu segmen deret menjadi dataset supervised (X, y).

    X berbentuk (N, W, 1): N jendela, tiap jendela W langkah, satu
    fitur per langkah. y berbentuk (N,): nilai tepat sesudah jendela.
    """
    X, y = [], []
    for i in range(len(segmen) - panjang_window):
        X.append(segmen[i:i + panjang_window])
        y.append(segmen[i + panjang_window])
    X = np.array(X).reshape(-1, panjang_window, 1)
    y = np.array(y)
    return X, y


def siapkan_dataset(panjang_window=20):
    """Rangkai seluruh pipeline: muat, bagi, normalisasi, window.

    Normalisasi memakai statistik segmen latih saja. Kembalikan
    dictionary berisi tensor masukan dan target ketiga segmen beserta
    statistik normalisasi.
    """
    deret = muat_deret()
    latih, validasi, uji = bagi_deret(deret)
    mu, sigma = statistik(latih)

    z_latih = normalisasi(latih, mu, sigma)
    z_validasi = normalisasi(validasi, mu, sigma)
    z_uji = normalisasi(uji, mu, sigma)

    X_latih, y_latih = buat_window(z_latih, panjang_window)
    X_validasi, y_validasi = buat_window(z_validasi, panjang_window)
    X_uji, y_uji = buat_window(z_uji, panjang_window)

    return {
        "X_latih": X_latih, "y_latih": y_latih,
        "X_validasi": X_validasi, "y_validasi": y_validasi,
        "X_uji": X_uji, "y_uji": y_uji,
        "mu": mu, "sigma": sigma,
        "panjang_window": panjang_window,
    }


if __name__ == "__main__":
    deret = muat_deret()
    latih, validasi, uji = bagi_deret(deret)
    mu, sigma = statistik(latih)

    print(f"panjang deret : {len(deret)}")
    print(f"segmen latih  : {len(latih)} | validasi {len(validasi)} "
          f"| uji {len(uji)}")
    print(f"statistik latih: mu {mu:.2f} | sigma {sigma:.2f}")

    X, y = buat_window(latih[:40], panjang_window=20)
    print(f"\nwindow dari 40 titik: X {X.shape} | y {y.shape}")
    print("X[0,:,0]:", [round(float(v), 2) for v in X[0, :, 0]])
    print("y[0]    :", round(float(y[0]), 2))
    print("X[1,0,0]:", round(float(X[1, 0, 0]), 2),
          "(= X[0,1,0], jendela bergeser satu langkah)")

    data = siapkan_dataset()
    print(f"\ndataset penuh (W={data['panjang_window']}):")
    print(f"  latih     : X {data['X_latih'].shape} | y {data['y_latih'].shape}")
    print(f"  validasi  : X {data['X_validasi'].shape} | y {data['y_validasi'].shape}")
    print(f"  uji       : X {data['X_uji'].shape} | y {data['y_uji'].shape}")

Output yang Dihasilkan:

text
panjang deret : 400
segmen latih  : 280 | validasi 60 | uji 60
statistik latih: mu 128.31 | sigma 10.33

window dari 40 titik: X (20, 20, 1) | y (20,)
X[0,:,0]: [120.61, 124.63, 133.43, 136.52, 130.69, 129.7, 129.09, 124.74, 122.74, 119.6, 122.36, 121.45, 118.55, 118.08, 113.3, 107.18, 107.35, 104.64, 111.03, 114.39]
y[0]    : 120.83
X[1,0,0]: 124.63 (= X[0,1,0], jendela bergeser satu langkah)

dataset penuh (W=20):
  latih     : X (260, 20, 1) | y (260,)
  validasi  : X (40, 20, 1) | y (40,)
  uji       : X (40, 20, 1) | y (40,)

Mari kita telaah bagian-bagian pentingnya:

Baris 37-44 (bagi_deret): tiga potongan diambil dengan slicing berurutan tanpa pengacakan; segmen validasi dimulai tepat setelah segmen latih berakhir.
Baris 61-73 (buat_window): perulangan i memotong segmen[i:i+W] sebagai jendela dan segmen[i+W] sebagai target; reshape(-1, W, 1) menegakkan konvensi batch-first (N, T, D) dari chapter Data Sekuens .
Baris 76-101 (siapkan_dataset): statistik dihitung dari latih lalu diterapkan ke ketiga segmen — mu 128,31 dan sigma 10,33 hanya boleh mengetahui masa lalu. Hasil akhirnya 260 jendela latih, 40 validasi, 40 uji.
Baris 104-125 (demo __main__): keluaran memperlihatkan sifat sliding window — X[1,0,0] sama dengan X[0,1,0] karena jendela hanya bergeser satu langkah, dan y[0] menjadi elemen terakhir jendela berikutnya.

Model LSTM #

Arsitektur ModelDeret sengaja dibuat minimal: satu layer nn.LSTM dengan 64 unit tersembunyi diikuti satu nn.Linear yang mengubah hidden state terakhir menjadi satu angka prediksi.

python
Modul model LSTM (src/model.py)
"""Model prediksi deret waktu: LSTM kecil PyTorch.

Arsitektur ModelDeret:
    jendela (B, W, 1)
      -> nn.LSTM(input 1, hidden 32, batch_first=True)
      -> ambil hidden state terakhir (B, 32)
      -> nn.Linear(32, 1)  -> prediksi nilai berikutnya (B, 1)
"""

import torch
import torch.nn as nn


class ModelDeret(nn.Module):
    """LSTM satu lapis yang memprediksi satu langkah ke depan."""

    def __init__(self, dim_input=1, dim_tersembunyi=64):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size=dim_input,
            hidden_size=dim_tersembunyi,
            batch_first=True,
        )
        self.kepala = nn.Linear(dim_tersembunyi, 1)

    def forward(self, jendela):
        keluaran, (h_n, c_n) = self.lstm(jendela)   # (B, W, H), (1, B, H)
        fitur = keluaran[:, -1, :]                  # hidden state terakhir
        return self.kepala(fitur)                   # (B, 1)


def hitung_parameter(model):
    """Hitung total parameter yang dapat dipelajari."""
    return sum(p.numel() for p in model.parameters())


if __name__ == "__main__":
    torch.manual_seed(42)
    model = ModelDeret()

    jendela = torch.randn(4, 12, 1)  # batch 4, jendela 12 langkah
    with torch.no_grad():
        prediksi = model(jendela)

    print(f"shape jendela   : {tuple(jendela.shape)}")
    print(f"shape prediksi  : {tuple(prediksi.shape)}")
    print(f"total parameter : {hitung_parameter(model)}")
    print("prediksi (4 nilai):", [round(float(v), 3) for v in prediksi])

    keluaran, (h_n, c_n) = model.lstm(jendela)
    print(f"shape keluaran LSTM : {tuple(keluaran.shape)}")
    print(f"shape hidden state  : {tuple(h_n.shape)}")
    print(f"shape cell state    : {tuple(c_n.shape)}")

Skrip di atas menghasilkan keluaran berikut saat dijalankan di mesin uji (CPU, PyTorch 2.14):

Output yang Dihasilkan:

text
shape jendela   : (4, 12, 1)
shape prediksi  : (4, 1)
total parameter : 17217
prediksi (4 nilai): [0.075, 0.064, 0.062, 0.054]
shape keluaran LSTM : (4, 12, 64)
shape hidden state  : (1, 4, 64)
shape cell state    : (1, 4, 64)

Mari kita telaah kode di atas baris per baris:

Baris 17-24 (__init__): batch_first=True membuat LSTM menerima (B, W, 1) langsung tanpa perlu transposisi — kebiasaan yang diingatkan chapter Data Sekuens karena bawaan layer rekuren sebenarnya (T, B, D).
Baris 26-30 (forward): keluaran[:, -1, :] mengambil representasi langkah waktu terakhir dari seluruh batch; kepala nn.Linear(64, 1) menerjemahkannya menjadi prediksi satu nilai berikutnya.
Baris 37-53 (demo __main__): keluaran menunjukkan 17.217 parameter — LSTM menyumbang 17.152 (empat gerbang × bobot masukan, rekuren, dan bias) dan Linear menyumbang 65. Shape (1, 4, 64) pada hidden state menegaskan satu lapis satu arah.

Uji Kedua Modul #

Dua berkas test menjaga kontrak modul ini: bentuk window, urutan temporal, roundtrip normalisasi, shape internal LSTM, sampai determinisme inisialisasi:

python
Test dataset (tests/test_dataset.py)
"""Test untuk modul dataset (windowing dan pembagian temporal).

Jalankan: python3 test_dataset.py
"""

import sys
from pathlib import Path

import numpy as np

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))

from dataset import (  # noqa: E402
    bagi_deret,
    buat_window,
    denormalisasi,
    muat_deret,
    normalisasi,
    siapkan_dataset,
    statistik,
)


def test_muat_deret():
    deret = muat_deret()
    assert deret.shape == (400,)
    assert deret.dtype == np.float64
    assert 100.0 < deret.min() < deret.max() < 200.0


def test_bentuk_window():
    deret = np.arange(30.0)
    X, y = buat_window(deret, panjang_window=8)
    assert X.shape == (22, 8, 1)      # 30 - 8 jendela
    assert y.shape == (22,)
    assert X[0, :, 0].tolist() == [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0]
    assert y[0] == 8.0                # nilai tepat sesudah jendela
    assert X[1, 0, 0] == 1.0          # jendela bergeser satu langkah
    assert y[1] == X[1, -1, 0] + 1.0  # target = nilai berikutnya


def test_bagi_deret_temporal():
    deret = muat_deret()
    latih, validasi, uji = bagi_deret(deret)
    assert len(latih) == 280 and len(validasi) == 60 and len(uji) == 60
    # urutan waktu terjaga: tidak ada pengacakan
    assert latih[-1] == deret[279]
    assert validasi[0] == deret[280]
    assert uji[0] == deret[340]
    assert uji[-1] == deret[399]


def test_normalisasi_roundtrip():
    deret = muat_deret()
    mu, sigma = statistik(deret[:280])
    z = normalisasi(deret, mu, sigma)
    assert abs(z.mean() - (deret.mean() - mu) / sigma) < 1e-12
    kembali = denormalisasi(z, mu, sigma)
    assert np.allclose(kembali, deret)


def test_siapkan_dataset():
    data = siapkan_dataset(panjang_window=20)
    assert data["X_latih"].shape == (260, 20, 1)
    assert data["X_validasi"].shape == (40, 20, 1)
    assert data["X_uji"].shape == (40, 20, 1)
    assert data["y_latih"].shape == (260,)
    # jendela uji tidak menjembatani segmen validasi
    assert data["X_uji"][0, 0, 0] != data["X_validasi"][-1, -1, 0]


if __name__ == "__main__":
    test_muat_deret()
    test_bentuk_window()
    test_bagi_deret_temporal()
    test_normalisasi_roundtrip()
    test_siapkan_dataset()
    print("SEMUA TEST LULUS")

Output yang Dihasilkan:

text

SEMUA TEST LULUS

python
Test model (tests/test_model.py)
"""Test untuk modul model LSTM (bentuk keluaran, parameter, gradien).

Jalankan: python3 test_model.py
"""

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))

import torch  # noqa: E402

from model import ModelDeret, hitung_parameter  # noqa: E402


def buat_model():
    torch.manual_seed(42)
    return ModelDeret(dim_input=1, dim_tersembunyi=64)


def test_bentuk_prediksi():
    model = buat_model().eval()
    jendela = torch.randn(6, 20, 1)  # batch 6, jendela 20 langkah
    with torch.no_grad():
        prediksi = model(jendela)
    assert prediksi.shape == (6, 1)
    assert torch.isfinite(prediksi).all()


def test_bentuk_internal_lstm():
    model = buat_model().eval()
    jendela = torch.randn(3, 20, 1)
    keluaran, (h_n, c_n) = model.lstm(jendela)
    assert keluaran.shape == (3, 20, 64)   # (B, T, H) karena batch_first
    assert h_n.shape == (1, 3, 64)         # satu lapis, satu arah
    assert c_n.shape == (1, 3, 64)
    # hidden state akhir tiap sampel sama dengan keluaran langkah terakhir
    assert torch.allclose(h_n[0], keluaran[:, -1, :])


def test_jumlah_parameter():
    model = ModelDeret(dim_input=1, dim_tersembunyi=64)
    # LSTM: 4 * ((1*64) + (64*64) + 64 + 64); Linear: 64*1 + 1
    lstm = 4 * (1 * 64 + 64 * 64 + 64 + 64)
    linear = 64 * 1 + 1
    assert hitung_parameter(model) == lstm + linear


def test_gradien_mengalir():
    model = buat_model()
    jendela = torch.randn(2, 20, 1)
    target = torch.zeros(2, 1)
    loss = torch.nn.functional.mse_loss(model(jendela), target)
    loss.backward()
    assert model.lstm.weight_ih_l0.grad is not None
    assert model.kepala.weight.grad is not None
    assert model.lstm.weight_ih_l0.grad.abs().sum().item() > 0


def test_deterministik_seed():
    torch.manual_seed(42)
    model_a = ModelDeret(dim_tersembunyi=64).eval()
    torch.manual_seed(42)
    model_b = ModelDeret(dim_tersembunyi=64).eval()
    jendela = torch.randn(4, 20, 1)
    with torch.no_grad():
        assert torch.allclose(model_a(jendela), model_b(jendela))


if __name__ == "__main__":
    test_bentuk_prediksi()
    test_bentuk_internal_lstm()
    test_jumlah_parameter()
    test_gradien_mengalir()
    test_deterministik_seed()
    print("SEMUA TEST LULUS")

Output yang Dihasilkan:

text

SEMUA TEST LULUS

Data dan model kini bertemu dalam bentuk yang kompatibel: 260 jendela (20, 1) menantang 17.217 parameter. Bagian berikutnya melatih keduanya dengan pembagian temporal yang jujur dan membuktikan lewat eksperimen mengapa shuffle dilarang.