import os import sys # Inject virtualenv site-packages path to bypass CloudLinux / Passenger application registration issues. # This allows running the script using direct python while loading packages from the virtualenv. for path in [ "/home/gaskutec/virtualenv/public_html/LPG_python/3.9/lib/python3.9/site-packages", "/home/gaskutec/virtualenv/public_html/LPG_python/3.9/lib64/python3.9/site-packages", "/home/gaskutec/virtualenv/public_html/LPG/3.9/lib/python3.9/site-packages", "/home/gaskutec/virtualenv/public_html/LPG/3.9/lib64/python3.9/site-packages", ]: if os.path.exists(path) and path not in sys.path: sys.path.insert(0, path) # Limit thread usage for scientific libraries to prevent OpenBLAS thread initialization errors (RLIMIT_NPROC / Resource temporarily unavailable) os.environ["MKL_NUM_THREADS"] = "1" os.environ["NUMEXPR_NUM_THREADS"] = "1" os.environ["OMP_NUM_THREADS"] = "1" os.environ["OPENBLAS_NUM_THREADS"] = "1" os.environ["VECLIB_MAXIMUM_THREADS"] = "1" import json import argparse import warnings warnings.filterwarnings("ignore") import pandas as pd import numpy as np from sqlalchemy import create_engine from statsmodels.tsa.statespace.sarimax import SARIMAX # Global variable for output mode JSON_MODE = False def my_print(*args, **kwargs): if not JSON_MODE: print(*args, **kwargs) # ========================================== # 1. KONEKSI DATABASE & PENGAMBILAN DATA # ========================================== def fetch_lpg_data(host='127.0.0.1', port='3306', db='LPG', user='root', password='', json_mode=False): def log_msg(msg): if not json_mode: print(msg) try: import urllib.parse safe_password = urllib.parse.quote_plus(password) engine = create_engine(f"mysql+pymysql://{user}:{safe_password}@{host}:{port}/{db}") query = """ SELECT DATE(tanggal_transaksi) as tanggal, SUM(qty_beli_isi) as transaksi_harian FROM transaksi GROUP BY DATE(tanggal_transaksi) ORDER BY tanggal ASC """ df = pd.read_sql(query, engine) if df.empty or len(df) < 14: log_msg("\n[PERINGATAN] Data riil dari MySQL belum mencukupi (Minimal 14 hari).") return df log_msg("[BERHASIL] Menggunakan data riil dari tabel database.") df['tanggal'] = pd.to_datetime(df['tanggal']) df['transaksi_harian'] = pd.to_numeric(df['transaksi_harian']) df.set_index('tanggal', inplace=True) # Re-index apabila ada hari kosong yang tidak ada transaksi idx = pd.date_range(df.index.min(), df.index.max()) df = df.reindex(idx, fill_value=0) return df except Exception as err: print(f"Error Database: {err}") return None # ========================================== # 2. FUNGSI UTAMA (MAIN ROUTINE) # ========================================== def main(): global JSON_MODE parser = argparse.ArgumentParser(description='LPG Forecasting using SARIMA') parser.add_argument('--json', action='store_true', help='Output hasil dalam format JSON') parser.add_argument('--host', type=str, default='127.0.0.1', help='Database host') parser.add_argument('--port', type=str, default='3306', help='Database port') parser.add_argument('--db', type=str, default='LPG', help='Database name') parser.add_argument('--user', type=str, default='root', help='Database user') parser.add_argument('--password', type=str, default='', help='Database password') args = parser.parse_args() JSON_MODE = args.json # Suppress output jika json mode import io if JSON_MODE: old_stdout = sys.stdout sys.stdout = io.StringIO() my_print("1. Mengambil data transaksi dari MySQL...") df = fetch_lpg_data( host=args.host, port=args.port, db=args.db, user=args.user, password=args.password, json_mode=JSON_MODE ) if df is None or df.empty or len(df) < 14: msg = "Data kosong atau belum cukup untuk modeling SARIMA (Butuh minimal 14 hari data historis)." if JSON_MODE: sys.stdout = old_stdout print("\nJSON_DATA:" + json.dumps({"status": "error", "error": msg})) else: print(msg) return my_print(f"2. Memulai fitting model SARIMA pada {len(df)} data harian...") try: # Menentukan order SARIMA (Seasonal ARIMAX) # s=7 karena data harian memiliki siklus mingguan (weekly seasonality) model = SARIMAX( df['transaksi_harian'], order=(1, 1, 1), seasonal_order=(1, 1, 1, 7), enforce_stationarity=False, enforce_invertibility=False ) model_fit = model.fit(disp=False) my_print("3. Membuat prediksi 14 hari kedepan...") forecast = model_fit.forecast(steps=14) # Tanggal prediksi last_date = df.index[-1] future_dates = pd.date_range(start=last_date + pd.Timedelta(days=1), periods=14) # Nilai prediksi (pastikan tidak negatif) predictions = np.clip(forecast.values, 0, None) my_print("4. Mengevaluasi akurasi model...") # Hitung fitted values di historical data untuk menghitung akurasi (min 7 data terakhir) eval_start = max(7, len(df) - 30) in_sample_pred = model_fit.predict(start=eval_start, end=len(df)-1) actual = df['transaksi_harian'].iloc[eval_start:] errors = actual - in_sample_pred rmse = float(np.sqrt(np.mean(errors**2))) sum_abs_error = float(np.sum(np.abs(errors))) sum_actual = float(np.sum(actual)) wmape = (sum_abs_error / sum_actual) * 100 if sum_actual > 0 else 0.0 accuracy = max(0.0, 100.0 - wmape) my_print("=== HASIL PERAMALAN SARIMA ===") my_print(f"Akurasi Model (WMAPE): {accuracy:.2f}%") my_print(f"RMSE: {rmse:.2f}") past_days = min(30, len(df)) if JSON_MODE: sys.stdout = old_stdout result = { "status": "success", "algorithm": "SARIMA (Seasonal ARIMA)", "future_dates": [d.strftime('%Y-%m-%d') for d in future_dates], "predictions": [int(round(p)) for p in predictions], "past_dates": [d.strftime('%Y-%m-%d') for d in df.index[-past_days:]], "past_values": [int(v) for v in df['transaksi_harian'].values[-past_days:]], "accuracy": round(accuracy, 2), "rmse": round(rmse, 2), "plot_url": None } print("\nJSON_DATA:" + json.dumps(result)) except Exception as e: msg = f"Gagal melatih model SARIMA: {str(e)}" if JSON_MODE: sys.stdout = old_stdout print("\nJSON_DATA:" + json.dumps({"status": "error", "error": msg})) else: print(msg) if __name__ == "__main__": main()