Bertindaklah sebagai Data Scientist senior yang ah...

Saya sudah cek file Anda dan membuat skrip XGBoost yang siap dijalankan di Google Colab. Dataset terdeteksi berisi 2.169 baris dan kolomnya sesuai: timestamp, raw_number, d1-d4, sum_digits, p1-p4, m1-m4, is_double, lag features, dan cyclic features.

File script lengkap bisa diunduh di sini:

Download script Colab XGBoost

Saya juga sudah test dengan dataset Anda. Hasil prediksi baris berikutnya dari script saat ini adalah:

text
Prediksi 4 digit next: 4555

Catatan penting: untuk data angka yang sangat acak, model ML tidak bisa menjamin prediksi benar. Di test set dataset Anda, akurasi pembulatan per digit berada sekitar 9%–11%, dan exact 4-digit match masih 0%, jadi metrik evaluasi harus dijadikan acuan utama.

Berikut versi script lengkap yang bisa langsung Anda copy ke 1 cell Google Colab:

python
# ============================================================
# XGBoost Time-Series Forecasting untuk Prediksi 4 Digit Berikutnya
# Dataset: PRIZE_LEVEL_4_ELITE.csv
# Cocok dijalankan di Google Colab sebagai 1 cell
# ============================================================

import os
import sys
import subprocess
from pathlib import Path


def ensure_package(package_name: str, import_name: str | None = None) -> None:
    """Install package jika belum tersedia. Aman untuk Google Colab."""
    import_name = import_name or package_name
    try:
        __import__(import_name)
    except ImportError:
        subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", package_name])


ensure_package("xgboost")
ensure_package("scikit-learn", "sklearn")
ensure_package("pandas")
ensure_package("numpy")
ensure_package("matplotlib")

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from xgboost import XGBRegressor
from sklearn.metrics import mean_absolute_error, accuracy_score


# -----------------------------
# 1) Konfigurasi umum
# -----------------------------
RANDOM_STATE = 42
TEST_SIZE_RATIO = 0.20

# Jika True, lag dan rolling dihitung ulang dari sum_digits secara causal.
# Ini membantu memastikan tidak ada leakage dari masa depan.
RECOMPUTE_SAFE_LAG_FEATURES = True

# Jika True, h_sin dan h_cos dihitung ulang dari timestamp.
RECOMPUTE_CYCLICAL_FEATURES = True

# Default False agar d1-d4 saat ini tidak dipakai sebagai fitur utama.
# Ubah ke True jika Anda memang ingin menggunakan digit hasil saat ini
# untuk memprediksi baris berikutnya.
INCLUDE_CURRENT_DIGITS_AS_FEATURES = False

TARGET_COLS = ["d1", "d2", "d3", "d4"]

BASE_FEATURE_COLS = [
    "sum_digits",
    "is_double",
    "lag_1_sum",
    "lag_2_sum",
    "rolling_mean_sum_3",
    "h_sin",
    "h_cos",
    "p1", "p2", "p3", "p4",
    "m1", "m2", "m3", "m4",
]

OPTIONAL_CURRENT_DIGIT_FEATURES = ["d1", "d2", "d3", "d4"]


# -----------------------------
# 2) Load file CSV di Google Colab
# -----------------------------
def resolve_csv_path(filename: str = "PRIZE_LEVEL_4_ELITE.csv") -> Path:
    """Cari file CSV. Jika di Colab belum ada, minta user upload file."""
    candidate_paths = [
        Path(filename),
        Path("/content") / filename,
        Path("/mnt/data") / filename,
    ]

    for path in candidate_paths:
        if path.exists():
            return path

    try:
        from google.colab import files
        print(f"File '{filename}' belum ditemukan. Silakan upload file CSV Anda.")
        uploaded = files.upload()

        if filename in uploaded:
            return Path(filename)

        if uploaded:
            first_file = next(iter(uploaded.keys()))
            print(f"Menggunakan file yang diupload: {first_file}")
            return Path(first_file)

    except Exception as exc:
        raise FileNotFoundError(
            f"File '{filename}' tidak ditemukan. Upload file ke Colab "
            f"atau ubah path secara manual. Detail: {exc}"
        )

    raise FileNotFoundError(f"File '{filename}' tidak ditemukan.")


csv_path = resolve_csv_path("PRIZE_LEVEL_4_ELITE.csv")
print(f"Membaca dataset dari: {csv_path}")

raw_df = pd.read_csv(csv_path)
print(f"Shape awal: {raw_df.shape}")
print("Kolom dataset:", list(raw_df.columns))


# -----------------------------
# 3) Preprocessing timestamp, sorting, dan validasi kolom
# -----------------------------
required_cols = ["timestamp", "raw_number"] + TARGET_COLS + BASE_FEATURE_COLS
missing_cols = [col for col in required_cols if col not in raw_df.columns]

if missing_cols:
    raise ValueError(f"Kolom wajib tidak ditemukan di dataset: {missing_cols}")

df = raw_df.copy()

# Simpan urutan asli agar timestamp duplikat tetap mengikuti urutan file.
df["__original_order"] = np.arange(len(df))

# Konversi timestamp ke datetime.
df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce")

invalid_ts_count = df["timestamp"].isna().sum()
if invalid_ts_count > 0:
    print(f"Menghapus {invalid_ts_count} baris dengan timestamp tidak valid.")
    df = df.dropna(subset=["timestamp"]).copy()

# Urutkan berdasarkan waktu, lalu urutan asli.
df = df.sort_values(
    ["timestamp", "__original_order"],
    kind="mergesort"
).reset_index(drop=True)

# Pastikan kolom numerik valid.
numeric_cols = list(set(TARGET_COLS + BASE_FEATURE_COLS + ["raw_number"]))

for col in numeric_cols:
    df[col] = pd.to_numeric(df[col], errors="coerce")

important_numeric_cols = TARGET_COLS + BASE_FEATURE_COLS
before_drop = len(df)

df = df.dropna(subset=important_numeric_cols).reset_index(drop=True)

print(f"Baris dihapus karena nilai numerik kosong/tidak valid: {before_drop - len(df)}")


# -----------------------------
# 4) Rekayasa fitur aman dari leakage
# -----------------------------
if RECOMPUTE_SAFE_LAG_FEATURES:
    # Untuk baris t:
    # lag_1_sum = sum_digits baris t-1
    # lag_2_sum = sum_digits baris t-2
    df["lag_1_sum"] = df["sum_digits"].shift(1)
    df["lag_2_sum"] = df["sum_digits"].shift(2)

    # rolling_mean_sum_3 memakai 3 nilai sum_digits sebelumnya saja.
    # shift(1) memastikan sum_digits baris saat ini tidak masuk rolling window.
    df["rolling_mean_sum_3"] = (
        df["sum_digits"]
        .shift(1)
        .rolling(window=3, min_periods=1)
        .mean()
    )

    # Baris awal belum punya history, isi 0.
    df[["lag_1_sum", "lag_2_sum", "rolling_mean_sum_3"]] = df[
        ["lag_1_sum", "lag_2_sum", "rolling_mean_sum_3"]
    ].fillna(0)

if RECOMPUTE_CYCLICAL_FEATURES:
    # Encode waktu harian menjadi fitur siklikal.
    # 00:00 dan 23:59 akan berada dekat dalam ruang sin-cos.
    minutes_in_day = df["timestamp"].dt.hour * 60 + df["timestamp"].dt.minute
    angle = 2 * np.pi * minutes_in_day / (24 * 60)

    df["h_sin"] = np.sin(angle)
    df["h_cos"] = np.cos(angle)


# -----------------------------
# 5) Buat target next-row: d1/d2/d3/d4 baris berikutnya
# -----------------------------
future_target_cols = [f"next_{col}" for col in TARGET_COLS]

for col, next_col in zip(TARGET_COLS, future_target_cols):
    df[next_col] = df[col].shift(-1)

# Baris terakhir tidak punya target aktual berikutnya,
# jadi tidak dipakai training/evaluasi.
# Namun baris terakhir tetap dipakai nanti sebagai input prediksi masa depan.
model_df = df.dropna(subset=future_target_cols).copy().reset_index(drop=True)

for col in future_target_cols:
    model_df[col] = model_df[col].astype(int)

feature_cols = BASE_FEATURE_COLS.copy()

if INCLUDE_CURRENT_DIGITS_AS_FEATURES:
    feature_cols += OPTIONAL_CURRENT_DIGIT_FEATURES

feature_cols = [col for col in feature_cols if col in model_df.columns]

X = model_df[feature_cols].copy()
y = model_df[future_target_cols].copy()

print("\nFitur yang digunakan:")
print(feature_cols)
print(f"Jumlah sampel modeling: {len(model_df)}")

if len(model_df) < 20:
    raise ValueError("Data terlalu sedikit untuk train/test split yang stabil.")


# -----------------------------
# 6) Time-based train/test split
# -----------------------------
split_idx = int(len(model_df) * (1 - TEST_SIZE_RATIO))

X_train = X.iloc[:split_idx]
X_test = X.iloc[split_idx:]

y_train = y.iloc[:split_idx]
y_test = y.iloc[split_idx:]

print("\nRentang data train/test:")
print(f"Train: index 0 s.d. {split_idx - 1} | rows = {len(X_train)}")
print(f"Test : index {split_idx} s.d. {len(model_df) - 1} | rows = {len(X_test)}")
print(f"Timestamp train akhir: {model_df.loc[split_idx - 1, 'timestamp']}")
print(f"Timestamp test awal   : {model_df.loc[split_idx, 'timestamp']}")


# -----------------------------
# 7) Training 4 model XGBRegressor terpisah
# -----------------------------
models = {}
test_pred_float = np.zeros((len(X_test), len(future_target_cols)))

for idx, target_col in enumerate(future_target_cols):
    model = XGBRegressor(
        objective="reg:squarederror",
        n_estimators=400,
        learning_rate=0.03,
        max_depth=3,
        subsample=0.90,
        colsample_bytree=0.90,
        reg_lambda=1.0,
        min_child_weight=2,
        n_jobs=1,
        random_state=RANDOM_STATE + idx,
        tree_method="hist",
        eval_metric="mae",
    )

    model.fit(X_train, y_train[target_col])

    models[target_col] = model
    test_pred_float[:, idx] = model.predict(X_test)

# Output regresi berupa float, jadi dibulatkan dan dipaksa ke range digit 0-9.
test_pred_rounded = np.rint(test_pred_float).clip(0, 9).astype(int)


# -----------------------------
# 8) Evaluasi MAE dan akurasi digit
# -----------------------------
evaluation_rows = []

for idx, target_col in enumerate(future_target_cols):
    mae = mean_absolute_error(y_test[target_col], test_pred_float[:, idx])

    rounded_acc = accuracy_score(
        y_test[target_col].astype(int),
        test_pred_rounded[:, idx]
    )

    evaluation_rows.append({
        "target": target_col,
        "MAE": mae,
        "rounded_digit_accuracy": rounded_acc,
    })

eval_df = pd.DataFrame(evaluation_rows)

exact_4digit_accuracy = np.mean(
    (y_test.to_numpy(dtype=int) == test_pred_rounded).all(axis=1)
)

print("\nEvaluasi per digit pada test set:")
print(eval_df.to_string(index=False))

print(f"\nExact 4-digit match accuracy: {exact_4digit_accuracy:.4f}")


# -----------------------------
# 9) Preview beberapa hasil prediksi test
# -----------------------------
preview = pd.DataFrame({
    "timestamp_input": model_df.iloc[split_idx:]["timestamp"].values,
    "actual_next_4d": [
        "".join(map(str, row))
        for row in y_test.to_numpy(dtype=int)
    ],
    "pred_next_4d": [
        "".join(map(str, row))
        for row in test_pred_rounded
    ],
})

print("\nPreview prediksi test terakhir:")
print(preview.tail(10).to_string(index=False))


# -----------------------------
# 10) Visualisasi MAE per digit
# -----------------------------
plt.figure(figsize=(7, 4))
plt.bar(eval_df["target"], eval_df["MAE"])
plt.title("MAE per Digit - XGBoost Test Set")
plt.xlabel("Target")
plt.ylabel("Mean Absolute Error")
plt.xticks(rotation=0)
plt.show()


# -----------------------------
# 11) Prediksi hasil selanjutnya setelah baris terakhir dataset
# -----------------------------
last_row = df.iloc[[-1]].copy()
X_next = last_row[feature_cols].copy()

next_pred_float = []

for target_col in future_target_cols:
    pred_value = models[target_col].predict(X_next)[0]
    next_pred_float.append(pred_value)

next_pred_digits = np.rint(next_pred_float).clip(0, 9).astype(int)
predicted_4digit = "".join(map(str, next_pred_digits))

last_timestamp = last_row["timestamp"].iloc[0]
last_raw_number = str(int(last_row["raw_number"].iloc[0])).zfill(4)

print("\n" + "=" * 60)
print("PREDIKSI BARIS BERIKUTNYA SETELAH DATA TERAKHIR")
print("=" * 60)
print(f"Timestamp baris terakhir : {last_timestamp}")
print(f"Raw number terakhir      : {last_raw_number}")
print(f"Prediksi float per digit : {np.round(next_pred_float, 4).tolist()}")
print(f"Prediksi digit bulat     : {next_pred_digits.tolist()}")
print(f"Prediksi 4 digit next    : {predicted_4digit}")
print("=" * 60)


# -----------------------------
# 12) Feature importance opsional
# -----------------------------
importance_tables = []

for target_col, model in models.items():
    imp = pd.DataFrame({
        "feature": feature_cols,
        "importance": model.feature_importances_,
        "target": target_col,
    }).sort_values("importance", ascending=False)

    importance_tables.append(imp)

feature_importance_df = pd.concat(importance_tables, ignore_index=True)

print("\nTop 10 feature importance per target:")

for target_col in future_target_cols:
    print(f"\n{target_col}")
    print(
        feature_importance_df[
            feature_importance_df["target"] == target_col
        ]
        .head(10)[["feature", "importance"]]
        .to_string(index=False)
    )

Untuk menjalankannya di Colab: upload PRIZE_LEVEL_4_ELITE.csv, paste script di atas ke satu cell, lalu run.