Spaces:
Sleeping
Sleeping
File size: 3,009 Bytes
b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 a90bc0e b8086d5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
import numpy as np
import torch
# Menggunakan ChronosPipeline untuk pemuatan dan inferensi yang efisien
from chronos import ChronosPipeline
class ModelHandler:
def __init__(self):
# Mengganti model lama dengan Chronos-2 yang lebih canggih
self.model_name = "amazon/chronos-2"
self.pipeline = None
# Penentuan device: "cuda" jika ada GPU, jika tidak "cpu"
self.device = "cuda" if torch.cuda.is_available() else "cpu"
self.load_model()
def load_model(self):
"""Load Chronos-2 model optimized for CPU/GPU"""
try:
print(f"Loading {self.model_name} on {self.device}...")
# ChronosPipeline menangani semua proses tokenisasi dan pemuatan arsitektur
self.pipeline = ChronosPipeline.from_pretrained(
self.model_name,
device_map=self.device,
)
print("Chronos-2 pipeline loaded successfully.")
except Exception as e:
print(f"Error loading Chronos-2 model: {e}")
print("Using fallback prediction method")
self.pipeline = None
def predict(self, data, horizon=10):
"""Generate predictions using Chronos-2 or fallback"""
try:
# Menggunakan data['original'] yang merupakan harga aktual riil
if data is None or len(data['original']) < 20:
return np.array([0] * horizon)
if self.pipeline is None:
# --- Fallback Logic ---
# Logic ekstrapolasi tren lama tetap dipertahankan jika model Deep Learning gagal dimuat
values = data['original']
recent_trend = np.polyfit(range(len(values[-20:])), values[-20:], 1)[0]
predictions = []
last_value = values[-1]
for i in range(horizon):
next_value = last_value + recent_trend * (i + 1)
noise = np.random.normal(0, data['std'] * 0.1)
predictions.append(next_value + noise)
return np.array(predictions)
# --- Chronos-2 Inference ---
# Input: numpy array dari harga Close historis yang riil
predictions_samples = self.pipeline.predict(
data['original'],
prediction_length=horizon,
# Mengambil 20 sampel prediksi untuk mendapatkan prediksi probablistik
num_samples=20
)
# Untuk chart (garis tunggal), ambil nilai rata-rata (mean) dari semua sampel.
mean_predictions = np.mean(predictions_samples, axis=0)
return mean_predictions
except Exception as e:
print(f"Prediction error: {e}")
# Mengembalikan array nol jika ada error saat inferensi Chronos
return np.array([0] * horizon) |