from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Dict, Tuple from datetime import datetime, timezone, timedelta from zoneinfo import ZoneInfo from contextlib import asynccontextmanager import os import joblib import requests import numpy as np import pandas as pd import tensorflow as tf import ta APP_TZ = ZoneInfo(os.getenv("APP_TIMEZONE", "Asia/Jakarta")) app = FastAPI(title="Crypto Prediction API") models: Dict[str, tf.keras.Model] = {} scalers: Dict[str, Tuple] = {} BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) MODEL_BASE_DIR = os.path.join(BASE_DIR, "models") BINANCE_KLINES_ENDPOINT = "https://api.binance.us/api/v3/klines" FGI_ENDPOINT = "https://api.alternative.me/fng/" FEATURE_COLUMNS = ["Open", "High", "Low", "Close", "Volume", "RSI", "MACD", "FGI"] @asynccontextmanager async def lifespan(app: FastAPI): print("Memuat arsitektur CNN-LSTM ke dalam memori...") try: for asset in ["BTC", "ETH"]: for horizon in [1, 3, 7]: model_key = f"{asset}_{horizon}" scaler_prefix = f"{asset.lower()}_h{horizon}" model_path = os.path.join(MODEL_BASE_DIR, f"{asset.lower()}_h{horizon}.keras") if not os.path.exists(model_path): raise FileNotFoundError(f"Model tidak ditemukan: {model_path}") models[model_key] = tf.keras.models.load_model(model_path) feature_path = os.path.join(MODEL_BASE_DIR, f"{scaler_prefix}_feature_scaler.pkl") target_path = os.path.join(MODEL_BASE_DIR, f"{scaler_prefix}_target_scaler.pkl") if os.path.exists(feature_path) and os.path.exists(target_path): feature_scaler = joblib.load(feature_path) target_scaler = joblib.load(target_path) scalers[model_key] = (feature_scaler, target_scaler) print(f" -> Scaler dimuat: {scaler_prefix}") else: print(f" -> WARNING: Scaler tidak ditemukan untuk {scaler_prefix}") print(f"Model {model_key} dimuat dari {model_path}") print(f"Total model: {len(models)}, Total scaler: {len(scalers)}") print("Model dan scaler siap digunakan!") except Exception as e: print(f"Gagal memuat model: {e}") raise yield print("Shutting down prediction API...") app = FastAPI(title="Crypto Prediction API", lifespan=lifespan) class PredictionRequest(BaseModel): coin: str # "BTC" atau "ETH" days: int # 1, 3, atau 7 base_date: str | None = None # "YYYY-MM-DD" untuk backfill, None untuk live def fetch_ohlcv(coin: str, interval: str = "1h", limit: int = 200, end_time: int | None = None) -> pd.DataFrame: symbol = f"{coin.upper()}USDT" params: dict = {"symbol": symbol, "interval": interval, "limit": limit} if end_time: params["endTime"] = end_time response = requests.get( BINANCE_KLINES_ENDPOINT, params=params, timeout=15, ) if response.status_code != 200: raise HTTPException(status_code=502, detail=f"Gagal mengambil data OHLCV dari Binance: {response.text}") columns = [ "open_time", "open", "high", "low", "close", "volume", "close_time", "quote_asset_volume", "number_of_trades", "taker_buy_base_asset_volume", "taker_buy_quote_asset_volume", "ignore", ] df = pd.DataFrame(response.json(), columns=columns) df["open_time"] = pd.to_datetime(df["open_time"], unit="ms") df["close_time"] = pd.to_datetime(df["close_time"], unit="ms") df[["open", "high", "low", "close", "volume"]] = df[["open", "high", "low", "close", "volume"]].astype(float) df.rename(columns={"open": "Open", "high": "High", "low": "Low", "close": "Close", "volume": "Volume"}, inplace=True) return df def fetch_fgi() -> float: response = requests.get(FGI_ENDPOINT, timeout=15) if response.status_code != 200: raise HTTPException(status_code=502, detail=f"Gagal mengambil FGI dari Alternative.me: {response.text}") payload = response.json() data = payload.get("data", []) if not data: raise HTTPException(status_code=502, detail="Data FGI tidak tersedia.") try: return float(data[0].get("value", 0)) except (TypeError, ValueError): raise HTTPException(status_code=502, detail="Format FGI tidak valid.") def fetch_fgi_for_date(target_date: str) -> float: try: response = requests.get(FGI_ENDPOINT, params={"limit": 30}, timeout=15) if response.status_code != 200: return fetch_fgi() payload = response.json() data = payload.get("data", []) if not data: return fetch_fgi() target_ts = int(datetime.strptime(target_date, "%Y-%m-%d").timestamp()) closest = None for entry in data: ts = int(entry.get("timestamp", 0)) if closest is None or abs(ts - target_ts) < abs(int(closest.get("timestamp", 0)) - target_ts): closest = entry if closest: return float(closest.get("value", 50)) except Exception: pass return 50.0 def build_feature_matrix(df: pd.DataFrame, fgi_value: float) -> pd.DataFrame: df = df.copy() df["RSI"] = ta.momentum.RSIIndicator(df["Close"], window=14).rsi() df["MACD"] = ta.trend.MACD(df["Close"]).macd() df["FGI"] = fgi_value df[["RSI", "MACD"]] = df[["RSI", "MACD"]].bfill() df = df.dropna(subset=["RSI", "MACD"]).reset_index(drop=True) if len(df) < 14: raise HTTPException(status_code=422, detail="Data historis tidak cukup untuk membangun fitur input.") return df[FEATURE_COLUMNS] def prepare_input(model: tf.keras.Model, feature_df: pd.DataFrame, feature_scaler) -> np.ndarray: if not model.inputs: raise HTTPException(status_code=500, detail="Model tidak memiliki input yang valid.") try: input_shape = model.inputs[0].shape.as_list() except AttributeError: input_shape = list(model.inputs[0].shape) lookback = input_shape[1] or 14 num_features = input_shape[2] if input_shape[2] is not None else len(FEATURE_COLUMNS) if num_features != len(FEATURE_COLUMNS): raise HTTPException(status_code=500, detail=f"Model memerlukan {num_features} fitur, tapi preprocessing memberikan {len(FEATURE_COLUMNS)} fitur.") if len(feature_df) < lookback: raise HTTPException(status_code=422, detail=f"Data historis kurang dari {lookback} bar untuk inferensi.") window = feature_df.iloc[-lookback:].astype(float) scaled = feature_scaler.transform(window) return scaled.reshape(1, lookback, num_features) def fetch_and_preprocess(coin: str, model: tf.keras.Model, feature_scaler, base_date: str | None = None): if base_date: end_dt = datetime.strptime(base_date, "%Y-%m-%d") end_ms = int((end_dt + timedelta(days=1)).timestamp() * 1000) - 1 raw_df = fetch_ohlcv(coin, interval="1h", limit=200, end_time=end_ms) fgi_value = fetch_fgi_for_date(base_date) base_mask = raw_df["close_time"] <= end_dt + timedelta(days=1) base_rows = raw_df[base_mask] if len(base_rows) == 0: raise HTTPException(status_code=422, detail=f"Tidak ada data OHLCV untuk {base_date}") last_close = float(base_rows["Close"].iloc[-1]) prediction_date = base_date else: raw_df = fetch_ohlcv(coin, interval="1h", limit=200) fgi_value = fetch_fgi() last_close = float(raw_df["Close"].iloc[-1]) prediction_date = datetime.now(APP_TZ).strftime("%Y-%m-%d") feature_df = build_feature_matrix(raw_df, fgi_value) base_date_result = raw_df["close_time"].iloc[-1].strftime("%Y-%m-%d") return prepare_input(model, feature_df, feature_scaler), last_close, base_date_result, prediction_date @app.get("/") async def root(): return { "status": "OK", "available_models": list(models.keys()), } @app.get("/model/status") async def model_status(coin: str, days: int): model_key = f"{coin.upper()}_{days}" if model_key not in models: raise HTTPException(status_code=400, detail="Koin atau rentang waktu tidak valid.") return { "coin": coin.upper(), "target_days": days, "status": "OK", "message": f"Model {model_key} tersedia.", "model_key": model_key, "available_models": list(models.keys()), } @app.post("/predict") async def get_prediction(request: PredictionRequest): model_key = f"{request.coin.upper()}_{request.days}" if model_key not in models: raise HTTPException(status_code=400, detail="Koin atau rentang waktu tidak valid.") if model_key not in scalers: raise HTTPException(status_code=500, detail=f"Scaler tidak tersedia untuk model {model_key}. Lakukan re-training.") selected_model = models[model_key] feature_scaler, target_scaler = scalers[model_key] processed_matrix, last_close, base_date, prediction_date = fetch_and_preprocess( request.coin, selected_model, feature_scaler, request.base_date ) try: prediction = selected_model.predict(processed_matrix, verbose=0) except Exception as exc: raise HTTPException(status_code=500, detail=f"Gagal melakukan inferensi model: {exc}") if prediction.size == 0: raise HTTPException(status_code=500, detail="Model mengembalikan prediksi kosong.") predicted_diff_scaled = float(np.asarray(prediction).reshape(-1)[0]) predicted_diff = target_scaler.inverse_transform([[predicted_diff_scaled]])[0][0] predicted_price = float(last_close + predicted_diff) return { "coin": request.coin.upper(), "target_days": request.days, "status": "Success", "message": f"Inferensi menggunakan model {model_key} berhasil.", "predicted_price": predicted_price, "prediction_date": prediction_date, "base_date": base_date, "last_close": last_close, "predicted_diff": float(predicted_diff), "model_status": {"model_key": model_key, "available": True}, }