TIF_E41253395/Api/main.py

from __future__ import annotations

import json
import os
import sqlite3
import threading
import time
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any, Literal

import numpy as np
import pandas as pd
import torch
import torch.nn as nn
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from sklearn.preprocessing import MinMaxScaler
from torch.utils.data import DataLoader, TensorDataset
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait


ROOT = Path(__file__).resolve().parents[1]
DB_PATH = Path(os.getenv("NBEATS_DB_PATH", ROOT / "data deploy hasil.sqlite"))
MODEL_PATH = Path(os.getenv("NBEATS_MODEL_PATH", ROOT / "best_nbeats_model.pth"))
HYPERPARAMETERS_PATH = Path(
	os.getenv("NBEATS_HYPERPARAMETERS_PATH", ROOT / "best_nbeats_hyperparameters.json")
)
TABLE_NAME = "data_deploy_hasil"
VALUE_COLUMNS = ("Harga Medium 1", "Harga Medium 2", "rata2")
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
MODEL_LOCK = threading.RLock()
UPDATE_LOCK = threading.Lock()
SCRAPING_URL = "https://www.bi.go.id/hargapangan/TabelHarga/PasarTradisionalDaerah"
SCRAPING_COMMODITIES = ("Beras Kualitas Medium I", "Beras Kualitas Medium II")
SCRAPING_PROVINCE = "Jawa Timur"
SCRAPING_REGENCY = "Kab. Banyuwangi"
FORECAST_HISTORY_DAYS = 90


def trend_basis(theta: torch.Tensor, size: int) -> torch.Tensor:
	time = torch.linspace(0, 1, steps=size, device=theta.device)
	basis = torch.stack([time**i for i in range(theta.shape[1])], dim=0)
	return theta @ basis


def seasonality_basis(theta: torch.Tensor, size: int) -> torch.Tensor:
	harmonics = theta.shape[1] // 2
	time = torch.linspace(0, 2 * np.pi, steps=size, device=theta.device)
	sine = torch.stack([torch.sin((i + 1) * time) for i in range(harmonics)], dim=0)
	cosine = torch.stack([torch.cos((i + 1) * time) for i in range(harmonics)], dim=0)
	return theta @ torch.cat([sine, cosine], dim=0)


class NBeatsBlock(nn.Module):
	def __init__(self, input_size: int, output_size: int, hidden_size: int,
				 theta_size: int, basis_type: str) -> None:
		super().__init__()
		self.input_size = input_size
		self.output_size = output_size
		self.basis_type = basis_type
		self.fc = nn.Sequential(
			nn.Linear(input_size, hidden_size), nn.GELU(),
			nn.Linear(hidden_size, hidden_size), nn.GELU(),
			nn.Linear(hidden_size, hidden_size), nn.GELU(),
			nn.Linear(hidden_size, hidden_size), nn.GELU(),
		)
		self.theta_back = nn.Linear(hidden_size, theta_size)
		self.theta_fore = nn.Linear(hidden_size, theta_size)

	def forward(self, values: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
		hidden = self.fc(values)
		theta_back = self.theta_back(hidden)
		theta_fore = self.theta_fore(hidden)
		if self.basis_type == "trend":
			return trend_basis(theta_back, self.input_size), trend_basis(theta_fore, self.output_size)
		return seasonality_basis(theta_back, self.input_size), seasonality_basis(theta_fore, self.output_size)


class NBeatsStack(nn.Module):
	def __init__(self, input_size: int, output_size: int, hidden_size: int,
				 theta_size: int, blocks: int, basis_type: str) -> None:
		super().__init__()
		self.blocks = nn.ModuleList(
			NBeatsBlock(input_size, output_size, hidden_size, theta_size, basis_type)
			for _ in range(blocks)
		)

	def forward(self, values: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor, list[torch.Tensor], list[torch.Tensor]]:
		residual = values
		forecast = torch.zeros((values.size(0), self.blocks[0].output_size), device=values.device)
		backcasts, inputs = [], []
		for block in self.blocks:
			backcast, block_forecast = block(residual)
			inputs.append(residual.detach())
			backcasts.append(backcast)
			residual = residual - backcast
			forecast = forecast + block_forecast
		return residual, forecast, backcasts, inputs


class NBeats(nn.Module):
	def __init__(self, input_size: int, output_size: int, hidden_size: int,
				 n_stacks_trend: int, n_stacks_seasonal: int,
				 blocks_per_stack: int, theta_size_trend: int,
				 theta_size_season: int) -> None:
		super().__init__()
		self.output_size = output_size
		self.trend_stacks = nn.ModuleList(
			NBeatsStack(input_size, output_size, hidden_size, theta_size_trend,
						blocks_per_stack, "trend") for _ in range(n_stacks_trend)
		)
		self.season_stacks = nn.ModuleList(
			NBeatsStack(input_size, output_size, hidden_size, theta_size_season * 2,
						blocks_per_stack, "seasonality") for _ in range(n_stacks_seasonal)
		)

	def forward(self, values: torch.Tensor) -> tuple[torch.Tensor, list[torch.Tensor], list[torch.Tensor]]:
		residual = values
		forecast = torch.zeros((values.size(0), self.output_size), device=values.device)
		backcasts, inputs = [], []
		for stack in [*self.trend_stacks, *self.season_stacks]:
			residual, stack_forecast, stack_backcasts, stack_inputs = stack(residual)
			forecast = forecast + stack_forecast
			backcasts.extend(stack_backcasts)
			inputs.extend(stack_inputs)
		return forecast, backcasts, inputs


def load_hyperparameters() -> dict[str, Any]:
	# Load tuned parameters when available; otherwise use the built-in defaults.
	if HYPERPARAMETERS_PATH.exists():
		return json.loads(HYPERPARAMETERS_PATH.read_text(encoding="utf-8"))
	return {
		"input_size": 7, "output_size": 1, "hidden_size": 512,
		"n_stacks_trend": 3, "n_stacks_seasonal": 4,
		"blocks_per_stack": 3, "theta_size_trend": 2, "theta_size_season": 4,
	}


def build_model(hyperparameters: dict[str, Any]) -> NBeats:
	# Construct the model architecture from the selected hyperparameters.
	model = NBeats(
		input_size=int(hyperparameters["input_size"]),
		output_size=int(hyperparameters["output_size"]),
		hidden_size=int(hyperparameters["hidden_size"]),
		n_stacks_trend=int(hyperparameters["n_stacks_trend"]),
		n_stacks_seasonal=int(hyperparameters["n_stacks_seasonal"]),
		blocks_per_stack=int(hyperparameters["blocks_per_stack"]),
		theta_size_trend=int(hyperparameters["theta_size_trend"]),
		theta_size_season=int(hyperparameters["theta_size_season"]),
	).to(DEVICE)
	return model


def clean_data(data: pd.DataFrame, value_column: str) -> pd.DataFrame:
	# Normalize dates and values before sorting and removing duplicate dates.
	data["Tanggal"] = pd.to_datetime(data["Tanggal"], errors="coerce")
	data[value_column] = pd.to_numeric(data[value_column], errors="coerce")
	data = (
		data
		.dropna(subset=["Tanggal"])
		.sort_values("Tanggal")
		.drop_duplicates(subset=["Tanggal"], keep="last")
		.reset_index(drop=True)
	)

	if data.empty:
		return data

	window = 3
	previous_mean = (
		data[value_column]
		.shift(1)
		.rolling(window=window, min_periods=1)
		.mean()
	)
	next_mean = (
		data[value_column]
		.shift(-1)
		.rolling(window=window, min_periods=1)
		.mean()
	)
	data[value_column] = (
		# Fill missing values using neighboring trends, then interpolation as fallback.
		data[value_column]
		.fillna((previous_mean + next_mean) / 2)
		.fillna(previous_mean)
		.fillna(next_mean)
		.interpolate(method="linear")
		.bfill()
		.ffill()
	)

	q1 = data[value_column].quantile(0.25)
	q3 = data[value_column].quantile(0.75)
	iqr = q3 - q1
	lower_bound = q1 - 1.5 * iqr
	upper_bound = q3 + 1.5 * iqr
	outlier_mask = (
		(data[value_column] < lower_bound)
		| (data[value_column] > upper_bound)
	)
	# Replace IQR outliers and interpolate them using surrounding observations.
	data.loc[outlier_mask, value_column] = np.nan
	data[value_column] = (
		data[value_column]
		.interpolate(method="linear")
		.bfill()
		.ffill()
	)
	return data


def read_data(value_column: str) -> pd.DataFrame:
	# Read one selected series for forecasting or retraining.
	if not DB_PATH.exists():
		raise HTTPException(status_code=404, detail=f"Database tidak ditemukan: {DB_PATH}")
	try:
		with sqlite3.connect(DB_PATH) as connection:
			data = pd.read_sql_query(
				f'SELECT "Tanggal", "{value_column}" FROM "{TABLE_NAME}" ORDER BY "Tanggal"',
				connection,
			)
	except (sqlite3.Error, pd.errors.DatabaseError) as error:
		raise HTTPException(status_code=400, detail=f"Gagal membaca database: {error}") from error
	data = clean_data(data, value_column)
	if data.empty:
		raise HTTPException(status_code=422, detail="Tidak ada data numerik yang dapat digunakan.")
	if data[value_column].isna().any():
		raise HTTPException(status_code=422, detail="Data masih memiliki nilai kosong setelah preprocessing.")
	return data


def read_all_data() -> pd.DataFrame:
	"""Read every column and row from SQLite without preprocessing."""
	# Keep the raw database response unchanged for GET /data.
	if not DB_PATH.exists():
		raise HTTPException(status_code=404, detail=f"Database tidak ditemukan: {DB_PATH}")
	try:
		with sqlite3.connect(DB_PATH) as connection:
			return pd.read_sql_query(
				f'SELECT * FROM "{TABLE_NAME}"',
				connection,
			)
	except (sqlite3.Error, pd.errors.DatabaseError) as error:
		raise HTTPException(status_code=400, detail=f"Gagal membaca database: {error}") from error


def preprocess_all_data(data: pd.DataFrame) -> pd.DataFrame:
	"""Clean dates and numeric columns, then add a sequential index."""
	# Validate the complete schema before applying transformations.
	required_columns = {"Tanggal", *VALUE_COLUMNS}
	missing_columns = required_columns.difference(data.columns)
	if missing_columns:
		raise HTTPException(
			status_code=422,
			detail=f"Kolom wajib tidak ditemukan: {', '.join(sorted(missing_columns))}",
		)

	data = data.copy()
	data["Tanggal"] = pd.to_datetime(data["Tanggal"], errors="coerce")
	for column in VALUE_COLUMNS:
		data[column] = pd.to_numeric(data[column], errors="coerce")
	data = (
		# Keep the latest record for duplicate dates after chronological sorting.
		data
		.dropna(subset=["Tanggal"])
		.sort_values("Tanggal")
		.drop_duplicates(subset=["Tanggal"], keep="last")
		.reset_index(drop=True)
	)
	if data.empty:
		raise HTTPException(status_code=422, detail="Tidak ada data dengan tanggal valid.")

	for column in VALUE_COLUMNS:
		# Fill missing values first so IQR bounds are calculated from usable data.
		data[column] = (
			data[column]
			.interpolate(method="linear", limit_direction="both")
			.bfill()
			.ffill()
		)
		q1 = data[column].quantile(0.25)
		q3 = data[column].quantile(0.75)
		iqr = q3 - q1
		lower_bound = q1 - 1.5 * iqr
		upper_bound = q3 + 1.5 * iqr
		data.loc[
			(data[column] < lower_bound) | (data[column] > upper_bound),
			column,
		] = np.nan
		# Fill values identified as outliers and verify that no missing values remain.
		data[column] = data[column].interpolate(
			method="linear", limit_direction="both"
		).bfill().ffill()
		if data[column].isna().any():
			raise HTTPException(
				status_code=422,
				detail=f"Kolom {column} tidak memiliki nilai valid untuk mengisi missing value.",
			)

	data["Tanggal"] = data["Tanggal"].dt.date.astype(str)
	data.insert(0, "index", range(len(data)))
	return data


def records_from_dataframe(data: pd.DataFrame) -> list[dict[str, Any]]:
	"""Convert a dataframe to JSON-safe records, including null values."""
	records = data.astype(object).where(pd.notna(data), None).to_dict(orient="records")
	for record in records:
		for key, value in record.items():
			if isinstance(value, (np.integer, np.floating)):
				record[key] = value.item()
	return records


def _click_scraping_tree_item(driver: webdriver.Chrome, wait: WebDriverWait,
							  tree_id: str, item_name: str) -> None:
	def find_checkbox(browser: webdriver.Chrome):
		rows = browser.find_elements(By.CSS_SELECTOR, f"#{tree_id} tr.dx-data-row")
		for row in rows:
			if row.text.replace("\u2003", " ").strip() == item_name:
				return row.find_element(By.CSS_SELECTOR, ".dx-select-checkbox")
		return False

	checkbox = wait.until(find_checkbox)
	if checkbox.get_attribute("aria-checked") != "true":
		driver.execute_script(
			"arguments[0].scrollIntoView({block: 'center'});", checkbox
		)
		checkbox.click()
		wait.until(lambda browser: checkbox.get_attribute("aria-checked") == "true")


def _set_scraping_date(driver: webdriver.Chrome, date_box_id: str, value: date) -> None:
	driver.execute_script(
		"window.jQuery(arguments[0]).dxDateBox('instance').option('value', arguments[1]);",
		f"#{date_box_id}",
		value.isoformat(),
	)


def _read_scraped_prices(driver: webdriver.Chrome) -> pd.DataFrame:
	header_row = next(
		row for row in driver.find_elements(
			By.CSS_SELECTOR, ".dx-datagrid-headers tr.dx-header-row"
		)
		if any(
			"/" in (cell.get_attribute("aria-label") or "")
			for cell in row.find_elements(By.CSS_SELECTOR, 'td[role="columnheader"]')
		)
	)
	headers = [
		(cell.get_attribute("aria-label") or "").replace("Column ", "").strip()
		for cell in header_row.find_elements(By.CSS_SELECTOR, 'td[role="columnheader"]')
		if (cell.get_attribute("aria-label") or "").startswith("Column ")
	]
	label_rows: list[list[str]] = []
	value_rows: list[list[str]] = []
	for table in driver.find_elements(By.CSS_SELECTOR, ".dx-datagrid-rowsview table"):
		rows = table.find_elements(By.CSS_SELECTOR, "tbody tr.dx-data-row")
		parsed_rows = [
			[cell.text.replace("\u2003", " ").strip() for cell in row.find_elements(By.CSS_SELECTOR, "td")]
			for row in rows
		]
		if any(
			any(commodity in " ".join(row) for commodity in SCRAPING_COMMODITIES)
			for row in parsed_rows
		):
			label_rows = parsed_rows
		elif parsed_rows:
			value_rows = parsed_rows

	if not headers or not label_rows or not value_rows:
		raise RuntimeError("Tabel harga belum menampilkan data.")

	records: list[dict[str, str]] = []
	for label_row, value_row in zip(label_rows, value_rows):
		if len(label_row) < 2:
			continue
		commodity = label_row[1]
		commodity = commodity.split(" ", 1)[1] if commodity[:1].isdigit() else commodity
		if commodity not in SCRAPING_COMMODITIES:
			continue
		for header, value in zip(headers[2:], value_row[2:]):
			if value and value != "-":
				records.append({"komoditas": commodity, "tanggal": header, "harga": value})
	return pd.DataFrame(records)


def scrape_missing_prices(start_date: date, end_date: date) -> pd.DataFrame:
	options = Options()
	options.add_argument("--headless=new")
	options.add_argument("--window-size=1920,1080")
	driver = webdriver.Chrome(options=options)
	wait = WebDriverWait(driver, 30)
	try:
		driver.get(SCRAPING_URL)
		wait.until(lambda browser: browser.find_element(By.TAG_NAME, "body"))
		for commodity in SCRAPING_COMMODITIES:
			_click_scraping_tree_item(driver, wait, "CommodityTree", commodity)
		_click_scraping_tree_item(driver, wait, "cboProvince", SCRAPING_PROVINCE)
		wait.until(
			lambda browser: browser.find_elements(
				By.CSS_SELECTOR, "#cboRegency tr.dx-data-row"
			)
		)
		_click_scraping_tree_item(driver, wait, "cboRegency", SCRAPING_REGENCY)
		_set_scraping_date(driver, "dboDateMulai", start_date)
		_set_scraping_date(driver, "dboDateSelesai", end_date)
		driver.find_element(By.ID, "btnReport").click()
		wait.until(
			lambda browser: browser.find_elements(
				By.CSS_SELECTOR,
				".dx-datagrid-rowsview table tbody tr.dx-data-row",
			)
		)
		time.sleep(1)
		return _read_scraped_prices(driver)
	finally:
		driver.quit()


def merge_scraped_prices(existing: pd.DataFrame, scraped: pd.DataFrame) -> pd.DataFrame:
	expected_columns = ["Tanggal", "Harga Medium 1", "Harga Medium 2", "rata2"]
	current = existing[expected_columns].copy()
	current["Tanggal"] = pd.to_datetime(current["Tanggal"], errors="coerce")
	current = current.dropna(subset=["Tanggal"]).set_index("Tanggal")
	if not scraped.empty:
		incoming = scraped.copy()
		incoming["Tanggal"] = pd.to_datetime(
			incoming["tanggal"].astype(str).str.replace(" ", "", regex=False),
			format="%d/%m/%Y",
			errors="coerce",
		)
		incoming["harga"] = pd.to_numeric(
			incoming["harga"].astype(str)
			.str.replace(",", "", regex=False)
			.str.replace("Rp", "", regex=False)
			.str.strip(),
			errors="coerce",
		)
		incoming = incoming.dropna(subset=["Tanggal", "harga"])
		new_values = incoming.pivot_table(
			index="Tanggal",
			columns="komoditas",
			values="harga",
			aggfunc="last",
		).rename(columns={
			"Beras Kualitas Medium I": "Harga Medium 1",
			"Beras Kualitas Medium II": "Harga Medium 2",
		})
		new_values = new_values.reindex(columns=["Harga Medium 1", "Harga Medium 2"])
		current = pd.concat([current.drop(index=new_values.index, errors="ignore"), new_values])

	if current.empty:
		raise HTTPException(status_code=422, detail="Tidak ada data tanggal yang valid.")
	current = current.sort_index()
	full_dates = pd.date_range(current.index.min(), current.index.max(), freq="D")
	current = current.reindex(full_dates)
	current.index.name = "Tanggal"
	price_columns = ["Harga Medium 1", "Harga Medium 2"]
	current[price_columns] = current[price_columns].interpolate(
		method="linear", limit_direction="both"
	)
	current["rata2"] = current[price_columns].mean(axis=1)
	return current.reset_index()[expected_columns]


def update_database_from_scraping() -> dict[str, Any]:
	existing = read_all_data()
	if existing.empty:
		raise HTTPException(status_code=422, detail="Database tidak memiliki data awal.")
	existing["Tanggal"] = pd.to_datetime(existing["Tanggal"], errors="coerce")
	existing = existing.dropna(subset=["Tanggal"])
	last_date = existing["Tanggal"].max().date()
	today = date.today()
	allowed_date = today if datetime.now().hour >= 13 else today - timedelta(days=1)
	if last_date >= allowed_date:
		return {
			"status": "up_to_date",
			"message": "Data sudah diperbarui sampai batas waktu yang diizinkan.",
			"last_date": last_date.isoformat(),
			"allowed_date": allowed_date.isoformat(),
			"updated_rows": 0,
		}

	start_date = last_date + timedelta(days=1)
	scraped = scrape_missing_prices(start_date, allowed_date)
	if scraped.empty:
		raise HTTPException(status_code=502, detail="Situs sumber tidak mengembalikan data.")
	merged = merge_scraped_prices(existing, scraped)
	with sqlite3.connect(DB_PATH) as connection:
		merged.to_sql(TABLE_NAME, connection, if_exists="replace", index=False)
	return {
		"status": "updated",
		"message": "Data berhasil diperbarui dari situs sumber.",
		"start_date": start_date.isoformat(),
		"last_date": last_date.isoformat(),
		"allowed_date": allowed_date.isoformat(),
		"updated_rows": len(scraped),
		"total_rows": len(merged),
	}


def dataset(values: np.ndarray, input_size: int) -> tuple[np.ndarray, np.ndarray]:
	# Convert the time series into sliding windows and next-step targets.
	if len(values) <= input_size:
		raise ValueError(f"Minimal {input_size + 1} data diperlukan.")
	inputs = np.array([values[i:i + input_size] for i in range(len(values) - input_size)])
	targets = np.array([values[i + input_size] for i in range(len(values) - input_size)])[:, None]
	return inputs, targets


class ForecastRequest(BaseModel):
	horizon: int = Field(default=7, ge=1, le=90)
	value_column: Literal["rata2", "Harga Medium 1", "Harga Medium 2"] = "rata2"


class RetrainRequest(BaseModel):
	train_ratio: float = Field(default=0.8, gt=0.5, lt=1.0)
	epochs: int = Field(default=20, ge=1, le=1000)
	batch_size: int = Field(default=32, ge=1, le=1024)
	learning_rate: float = Field(default=0.0001, gt=0, le=1)
	value_column: Literal["rata2", "Harga Medium 1", "Harga Medium 2"] = "rata2"


app = FastAPI(title="N-BEATS Forecast API", version="1.0.0")
HYPERPARAMETERS = load_hyperparameters()
MODEL = build_model(HYPERPARAMETERS)
CHECKPOINT: dict[str, Any] | None = None


def load_saved_model() -> None:
	global CHECKPOINT
	if not MODEL_PATH.exists():
		return
	# Restore model weights and preprocessing scaler from the saved checkpoint.
	checkpoint = torch.load(MODEL_PATH, map_location=DEVICE, weights_only=False)
	MODEL.load_state_dict(checkpoint["model_state_dict"])
	MODEL.eval()
	CHECKPOINT = checkpoint


try:
	load_saved_model()
except Exception as error:
	raise RuntimeError(f"Checkpoint tidak kompatibel dengan arsitektur model: {error}") from error


def scaler_from_checkpoint(data: pd.DataFrame, value_column: str) -> MinMaxScaler:
	# Reuse the training scaler when it matches the requested value column.
	scaler = MinMaxScaler()
	saved = CHECKPOINT.get("scaler") if CHECKPOINT else None
	if saved and saved.get("value_column") == value_column:
		scaler.min_ = np.asarray(saved["min_"], dtype=float)
		scaler.scale_ = np.asarray(saved["scale_"], dtype=float)
		scaler.data_min_ = np.asarray(saved["data_min_"], dtype=float)
		scaler.data_max_ = np.asarray(saved["data_max_"], dtype=float)
		scaler.data_range_ = np.asarray(saved["data_range_"], dtype=float)
		scaler.n_features_in_ = 1
		scaler.n_samples_seen_ = int(saved.get("n_samples_seen_", len(data)))
		return scaler
	scaler.fit(data[[value_column]].to_numpy())
	return scaler


# Endpoint GET /data: mengambil seluruh data mentah dari SQLite.
@app.get("/data")
def all_data() -> dict[str, Any]:
	"""Return all rows and columns exactly as stored in SQLite."""
	# Serialize the raw dataframe into JSON-safe records for the website.
	data = read_all_data()
	return {"total_rows": len(data), "columns": list(data.columns), "rows": records_from_dataframe(data)}


# Endpoint GET /data/preprocessed: mengambil seluruh data setelah preprocessing.
@app.get("/data/preprocessed")
def preprocess_data() -> dict[str, Any]:
	"""Return the complete dataset after date, missing-value, and outlier handling."""
	# Apply the complete cleaning pipeline before returning website data.
	data = preprocess_all_data(read_all_data())
	return {
		"total_rows": len(data),
		"columns": list(data.columns),
		"rows": records_from_dataframe(data),
	}


# Endpoint POST /data/update: mengambil data terbaru dan memperbarui SQLite.
@app.post("/data/update")
def update_data() -> dict[str, Any]:
	# Serialize updates so concurrent requests cannot overwrite each other.
	if not DB_PATH.exists():
		raise HTTPException(status_code=404, detail=f"Database tidak ditemukan: {DB_PATH}")
	if not UPDATE_LOCK.acquire(blocking=False):
		raise HTTPException(status_code=409, detail="Proses update data sedang berjalan.")
	try:
		try:
			return update_database_from_scraping()
		except HTTPException:
			raise
		except Exception as error:
			raise HTTPException(
				status_code=502,
				detail=f"Gagal mengambil atau menyimpan data scraping: {error}",
			) from error
	finally:
		UPDATE_LOCK.release()


# Endpoint POST /forecast: menghasilkan prediksi harga menggunakan model N-BEATS.
@app.post("/forecast")
def forecast(request: ForecastRequest) -> dict[str, Any]:
	# Normalize the latest 90 calendar days, predict one step repeatedly, and inverse-transform results.
	data = read_data(request.value_column)
	last_available_date = data["Tanggal"].iloc[-1]
	first_forecast_date = last_available_date - timedelta(days=FORECAST_HISTORY_DAYS - 1)
	data = data[data["Tanggal"] >= first_forecast_date].reset_index(drop=True)
	input_size = int(HYPERPARAMETERS["input_size"])
	if len(data) < input_size:
		raise HTTPException(
			status_code=422,
			detail=(
				f"Minimal {input_size} data valid dari {FORECAST_HISTORY_DAYS} hari terakhir "
				"diperlukan."
			),
		)
	scaler = scaler_from_checkpoint(data, request.value_column)
	history = scaler.transform(data[[request.value_column]].to_numpy()).ravel().tolist()
	predictions: list[float] = []
	with MODEL_LOCK, torch.no_grad():
		MODEL.eval()
		for _ in range(request.horizon):
			# Use the latest input window so each prediction feeds the next one.
			values = torch.tensor([history[-input_size:]], dtype=torch.float32, device=DEVICE)
			predicted = float(MODEL(values)[0][0, 0].cpu())
			history.append(predicted)
			predictions.append(float(scaler.inverse_transform([[predicted]])[0, 0]))
	last_date = data["Tanggal"].iloc[-1].date()
	return {"value_column": request.value_column, "horizon": request.horizon,
			"forecast": [{"date": (last_date + timedelta(days=i)).isoformat(), "value": value}
						 for i, value in enumerate(predictions, start=1)]}


# Endpoint POST /retrain: melatih ulang model dan menyimpan checkpoint terbaru.
@app.post("/retrain")
def retrain(request: RetrainRequest) -> dict[str, Any]:
	# Prepare sliding-window training and validation data from the selected series.
	data = read_data(request.value_column)
	input_size = int(HYPERPARAMETERS["input_size"])
	scaler = MinMaxScaler().fit(data[[request.value_column]].iloc[: int(len(data) * request.train_ratio)])
	scaled = scaler.transform(data[[request.value_column]]).ravel()
	inputs, targets = dataset(scaled, input_size)
	split_index = max(1, int(len(inputs) * request.train_ratio))
	if split_index >= len(inputs):
		raise HTTPException(status_code=422, detail="Split menghasilkan data validasi kosong.")
	train_inputs, val_inputs = inputs[:split_index], inputs[split_index:]
	train_targets, val_targets = targets[:split_index], targets[split_index:]
	train_loader = DataLoader(TensorDataset(torch.tensor(train_inputs, dtype=torch.float32),
											torch.tensor(train_targets, dtype=torch.float32)),
							  batch_size=request.batch_size, shuffle=False)
	model = build_model(HYPERPARAMETERS)
	optimizer = torch.optim.AdamW(model.parameters(), lr=request.learning_rate, weight_decay=1e-4)
	loss_fn = nn.MSELoss()
	history: list[dict[str, float]] = []
	for epoch in range(request.epochs):
		# Optimize model weights on training batches, then measure validation loss.
		model.train()
		train_loss = 0.0
		for batch_inputs, batch_targets in train_loader:
			batch_inputs, batch_targets = batch_inputs.to(DEVICE), batch_targets.to(DEVICE)
			prediction, backcasts, block_inputs = model(batch_inputs)
			loss = loss_fn(prediction, batch_targets)
			loss = loss + float(HYPERPARAMETERS.get("alpha", 0.1)) * sum(
				loss_fn(backcast, block_input) for backcast, block_input in zip(backcasts, block_inputs)
			) / len(backcasts)
			optimizer.zero_grad()
			loss.backward()
			optimizer.step()
			train_loss += loss.item() * len(batch_inputs)
		model.eval()
		with torch.no_grad():
			validation_prediction = model(torch.tensor(val_inputs, dtype=torch.float32, device=DEVICE))[0]
			validation_loss = loss_fn(validation_prediction, torch.tensor(val_targets, dtype=torch.float32, device=DEVICE)).item()
		history.append({"epoch": epoch + 1, "train_loss": train_loss / len(train_inputs), "validation_loss": validation_loss})
	checkpoint = {"model_state_dict": model.state_dict(), "optimizer_state_dict": optimizer.state_dict(),
				  "epoch": request.epochs, "best_val_loss": history[-1]["validation_loss"],
				  "hyperparameters": HYPERPARAMETERS, "scaler": {"value_column": request.value_column,
				  **{key: getattr(scaler, key).tolist() for key in ("min_", "scale_", "data_min_", "data_max_", "data_range_")},
				  "n_samples_seen_": int(scaler.n_samples_seen_)}}
	temporary_path = MODEL_PATH.with_suffix(".tmp")
	with MODEL_LOCK:
		# Atomically replace the checkpoint and activate the newly trained model.
		torch.save(checkpoint, temporary_path)
		temporary_path.replace(MODEL_PATH)
		MODEL.load_state_dict(model.state_dict())
		MODEL.eval()
		globals()["CHECKPOINT"] = checkpoint
	return {"message": "Model berhasil diretrain.", "train_rows": len(train_inputs),
			"validation_rows": len(val_inputs), "train_ratio": request.train_ratio,
			"epochs": request.epochs, "final_validation_loss": history[-1]["validation_loss"]}


if __name__ == "__main__":
	import uvicorn

	uvicorn.run(app, host="0.0.0.0", port=8000, reload=False)