# ml_pipeline.py — Full ML pipeline: cleaning → training → forecasting → labeling import os, json, pickle, warnings, logging import pandas as pd import numpy as np from datetime import date from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score from sklearn.model_selection import TimeSeriesSplit import lightgbm as lgb import xgboost as xgb from statsmodels.tsa.arima.model import ARIMA from statsmodels.tsa.statespace.sarimax import SARIMAX import optuna optuna.logging.set_verbosity(optuna.logging.WARNING) warnings.filterwarnings('ignore') logging.getLogger('cmdstanpy').setLevel(logging.WARNING) logging.getLogger('prophet').setLevel(logging.WARNING) MODEL_DIR = os.getenv('MODEL_DIR', 'models') os.makedirs(MODEL_DIR, exist_ok=True) # ══════════════════════════════════════════════════════════════════════════════ # 1. DATA QUALITY & CLEANING # ══════════════════════════════════════════════════════════════════════════════ def analyze_data_quality(df, col): total, missing = len(df), df[col].isna().sum() gap_lengths, cur = [], 0 for v in df[col].isna(): if v: cur += 1 else: if cur > 0: gap_lengths.append(cur) cur = 0 max_gap = max(gap_lengths) if gap_lengths else 0 outliers = (df[col].pct_change().abs() > 0.5).sum() mean, std = df[col].mean(), df[col].std() cv = (std / mean * 100) if mean > 0 else 0 return { 'total_records': total, 'missing' : int(missing), 'missing_pct' : round(missing / total * 100, 2), 'max_gap_days' : max_gap, 'outliers' : int(outliers), 'cv' : round(cv, 2), 'volatility' : 'rendah' if cv < 2 else ('sedang' if cv < 5 else 'tinggi'), 'mean_price' : round(mean, 2), 'std_price' : round(std, 2), } def clean_price_data(df, col, quality_info): df = df.copy() df[f'{col}_was_missing'] = df[col].isna().astype(int) temp = df[col].copy() mask = temp.isna() gap_id = (mask != mask.shift()).cumsum() gap_sizes = mask.groupby(gap_id).transform('sum') temp = temp.fillna(method='ffill', limit=3) med_mask = mask & (gap_sizes > 3) & (gap_sizes <= 7) ti = temp.interpolate(method='linear', limit=7) temp[med_mask] = ti[med_mask] long_mask = mask & (gap_sizes > 7) if long_mask.any(): rolling_med = temp.rolling(30, min_periods=3, center=True).median() ti2 = temp.interpolate(method='linear') temp[long_mask] = ti2.clip(lower=rolling_med*0.85, upper=rolling_med*1.15)[long_mask] df[col] = temp outlier_mask = df[col].pct_change().abs() > 0.4 if outlier_mask.any(): df.loc[outlier_mask, col] = np.nan df[col] = df[col].interpolate(method='linear', limit=5) return df def recommend_models(quality_info): models = ['lightgbm', 'xgboost'] if quality_info['missing_pct'] < 30 and quality_info['max_gap_days'] < 14: models.append('prophet') if quality_info['cv'] < 5 and quality_info['missing_pct'] < 20: models.append('arima') if quality_info['total_records'] >= 90 and quality_info['missing_pct'] < 25: models.append('sarima') return models # ══════════════════════════════════════════════════════════════════════════════ # 2. FEATURE ENGINEERING # ══════════════════════════════════════════════════════════════════════════════ def create_features(df, target_col, other_cols=None): df = df.copy() df['year'] = df['date'].dt.year df['month'] = df['date'].dt.month df['day'] = df['date'].dt.day df['day_of_week'] = df['date'].dt.dayofweek df['day_of_year'] = df['date'].dt.dayofyear df['week_of_year'] = df['date'].dt.isocalendar().week.astype(int) df['quarter'] = df['date'].dt.quarter df['is_weekend'] = (df['day_of_week'] >= 5).astype(int) df['is_month_start']= df['date'].dt.is_month_start.astype(int) df['is_month_end'] = df['date'].dt.is_month_end.astype(int) for lag in [1, 2, 3, 7, 14, 30]: df[f'price_lag_{lag}'] = df[target_col].shift(lag) for w in [7, 14, 30]: df[f'price_rolling_mean_{w}'] = df[target_col].rolling(w).mean() df[f'price_rolling_std_{w}'] = df[target_col].rolling(w).std() df[f'price_rolling_min_{w}'] = df[target_col].rolling(w).min() df[f'price_rolling_max_{w}'] = df[target_col].rolling(w).max() df['price_diff_1'] = df[target_col].diff(1) df['price_diff_7'] = df[target_col].diff(7) df['price_pct_change_1'] = df[target_col].pct_change(1) * 100 df['price_pct_change_7'] = df[target_col].pct_change(7) * 100 df['sma_7'] = df[target_col].rolling(7).mean() df['sma_30'] = df[target_col].rolling(30).mean() df['sma_diff'] = df['sma_7'] - df['sma_30'] df['volatility_7'] = df[target_col].rolling(7).std() df['volatility_30'] = df[target_col].rolling(30).std() if other_cols: for i, oc in enumerate(other_cols): if oc in df.columns: df[f'other_price_{i}'] = df[oc] df[f'price_spread_{i}'] = df[target_col] - df[oc] df[f'price_ratio_{i}'] = df[target_col] / (df[oc] + 1e-6) return df # ══════════════════════════════════════════════════════════════════════════════ # 3. TRAINING # ══════════════════════════════════════════════════════════════════════════════ def evaluate_model(y_true, y_pred, model_name=""): rmse = float(np.sqrt(mean_squared_error(y_true, y_pred))) return { 'model' : model_name, 'rmse' : rmse, 'mae' : float(mean_absolute_error(y_true, y_pred)), 'r2' : float(r2_score(y_true, y_pred)), 'mape' : float(np.mean(np.abs((y_true - y_pred) / (np.abs(y_true)+1e-6))) * 100), 'median_ae': float(np.median(np.abs(y_true - y_pred))), } def train_lightgbm(X_train, y_train, X_test, y_test): m = lgb.LGBMRegressor( objective='regression', metric='rmse', num_leaves=31, max_depth=7, learning_rate=0.05, n_estimators=300, min_child_samples=20, subsample=0.8, colsample_bytree=0.8, reg_alpha=0.1, reg_lambda=0.1, verbose=-1, random_state=42, force_col_wise=True, ) m.fit(X_train, y_train, eval_set=[(X_test, y_test)], eval_metric='rmse', callbacks=[lgb.early_stopping(50, verbose=False)]) return m, evaluate_model(y_test, m.predict(X_test), 'LightGBM') def train_xgboost(X_train, y_train, X_test, y_test): m = xgb.XGBRegressor( objective='reg:squarederror', n_estimators=300, max_depth=6, learning_rate=0.05, subsample=0.8, colsample_bytree=0.8, reg_alpha=0.1, reg_lambda=0.1, random_state=42, verbosity=0, ) m.fit(X_train, y_train, eval_set=[(X_test, y_test)], verbose=False) return m, evaluate_model(y_test, m.predict(X_test), 'XGBoost') def train_prophet(dates_train, y_train, dates_test, y_test): """Train Prophet with robust error handling and data validation.""" try: from prophet import Prophet except ImportError: raise RuntimeError("Prophet not installed") # Ensure clean datetime + float data ds_train = pd.to_datetime(dates_train).reset_index(drop=True) y_tr = y_train.reset_index(drop=True).astype(float) ds_test = pd.to_datetime(dates_test).reset_index(drop=True) y_te = y_test.reset_index(drop=True).astype(float) # Remove NaN rows mask = y_tr.notna() ds_train = ds_train[mask] y_tr = y_tr[mask] if len(y_tr) < 30: raise ValueError("Prophet: data training terlalu sedikit setelah cleaning") df_train = pd.DataFrame({'ds': ds_train.values, 'y': y_tr.values}) df_test = pd.DataFrame({'ds': ds_test.values}) m = Prophet( yearly_seasonality=True, weekly_seasonality=True, daily_seasonality=False, changepoint_prior_scale=0.05, ) # Matikan log dengan cara yang lebih aman logging.getLogger('prophet').setLevel(logging.ERROR) logging.getLogger('cmdstanpy').setLevel(logging.ERROR) m.fit(df_train) fc = m.predict(df_test) pred = np.clip(fc['yhat'].values, 0, None) return m, evaluate_model(y_te.values, pred, 'Prophet') def train_arima(y_train, y_test): m = ARIMA(y_train.values, order=(5, 1, 0)).fit() fc = np.clip(m.forecast(steps=len(y_test)), 0, None) return m, evaluate_model(y_test.values, fc, 'ARIMA') def train_sarima(y_train, y_test): """Train SARIMA with seasonal_order for weekly patterns.""" m = SARIMAX( y_train.values, order=(1, 1, 1), seasonal_order=(1, 1, 1, 7), enforce_stationarity=False, enforce_invertibility=False, ).fit(disp=False, maxiter=200) fc = np.clip(m.forecast(steps=len(y_test)), 0, None) return m, evaluate_model(y_test.values, fc, 'SARIMA') def forecast_sarima(model, n_days=7): """Forecast n_days ahead using fitted SARIMA model.""" fc = np.clip(model.forecast(steps=n_days), 0, None) last_date = pd.Timestamp.now().normalize() return [ {'tanggal': str((last_date + pd.Timedelta(days=i+1)).date()), 'harga_prediksi': round(float(fc[i]), 2)} for i in range(n_days) ] def tune_with_optuna(model_name, X_train, y_train, n_trials=50): def objective(trial): if model_name == 'LightGBM': params = { 'objective':'regression','metric':'rmse','verbose':-1, 'random_state':42,'force_col_wise':True, 'num_leaves' : trial.suggest_int('num_leaves', 20, 100), 'max_depth' : trial.suggest_int('max_depth', 4, 12), 'learning_rate' : trial.suggest_float('learning_rate', 0.01, 0.3), 'n_estimators' : trial.suggest_int('n_estimators', 100, 500), 'min_child_samples': trial.suggest_int('min_child_samples', 5, 50), 'subsample' : trial.suggest_float('subsample', 0.5, 1.0), 'colsample_bytree' : trial.suggest_float('colsample_bytree', 0.5, 1.0), 'reg_alpha' : trial.suggest_float('reg_alpha', 0.0, 1.0), 'reg_lambda' : trial.suggest_float('reg_lambda', 0.0, 1.0), } Cls = lgb.LGBMRegressor else: params = { 'objective':'reg:squarederror','random_state':42,'verbosity':0, 'n_estimators' : trial.suggest_int('n_estimators', 100, 500), 'max_depth' : trial.suggest_int('max_depth', 3, 10), 'learning_rate' : trial.suggest_float('learning_rate', 0.01, 0.3), 'subsample' : trial.suggest_float('subsample', 0.5, 1.0), 'colsample_bytree': trial.suggest_float('colsample_bytree', 0.5, 1.0), 'reg_alpha' : trial.suggest_float('reg_alpha', 0.0, 1.0), 'reg_lambda' : trial.suggest_float('reg_lambda', 0.0, 1.0), } Cls = xgb.XGBRegressor scores = [] for tr, val in TimeSeriesSplit(n_splits=5).split(X_train): m = Cls(**params) m.fit(X_train.iloc[tr], y_train.iloc[tr]) scores.append(np.sqrt(mean_squared_error(y_train.iloc[val], m.predict(X_train.iloc[val])))) return np.mean(scores) study = optuna.create_study(direction='minimize') study.optimize(objective, n_trials=n_trials) return study.best_params # ══════════════════════════════════════════════════════════════════════════════ # 4. FORECASTING # ══════════════════════════════════════════════════════════════════════════════ def forecast_tree(model, df_hist, target_col, other_cols, feature_cols, n_days=7): df_temp = df_hist.copy() preds = [] last_date = df_temp['date'].max() for i in range(1, n_days + 1): next_date = last_date + pd.Timedelta(days=i) new_row = {'date': next_date, target_col: np.nan} for oc in (other_cols or []): if oc in df_temp.columns: lo = df_temp[oc].dropna() new_row[oc] = float(lo.iloc[-1]) if len(lo) else np.nan df_temp = pd.concat([df_temp, pd.DataFrame([new_row])], ignore_index=True) df_feat = create_features(df_temp, target_col, other_cols) last_row = df_feat.iloc[[-1]].copy() for col in feature_cols: if col not in last_row.columns: last_row[col] = 0.0 pred = max(0, float(model.predict(last_row[feature_cols])[0])) preds.append({'tanggal': str(next_date.date()), 'harga_prediksi': round(pred, 2)}) df_temp.loc[df_temp['date'] == next_date, target_col] = pred return preds def forecast_arima(model, n_days=7): """Forecast n_days ahead using fitted ARIMA model.""" fc = np.clip(model.forecast(steps=n_days), 0, None) last_date = pd.Timestamp.now().normalize() return [ {'tanggal': str((last_date + pd.Timedelta(days=i+1)).date()), 'harga_prediksi': round(float(fc[i]), 2)} for i in range(n_days) ] def forecast_prophet(model, last_date, n_days=7): future = pd.DataFrame({'ds': pd.date_range(start=last_date + pd.Timedelta(days=1), periods=n_days)}) fc = model.predict(future) return [{'tanggal': str(r.ds.date()), 'harga_prediksi': round(max(0, r.yhat), 2)} for _, r in fc.iterrows()] # ══════════════════════════════════════════════════════════════════════════════ # 5. LABEL GENERATOR (untuk frontend) # ══════════════════════════════════════════════════════════════════════════════ def get_tren(preds): pct = (preds[-1]['harga_prediksi'] - preds[0]['harga_prediksi']) / preds[0]['harga_prediksi'] * 100 if pct > 2: return 'naik_tajam', pct elif pct > 0.5: return 'naik_terkendali', pct elif pct < -2: return 'turun_tajam', pct elif pct < -0.5: return 'turun_terkendali', pct else: return 'stabil', pct def get_labels(mape, rmse, mean_price, cv, preds, komoditas_nama): harga_vals = [p['harga_prediksi'] for p in preds] tren, pct = get_tren(preds) confidence = round(max(0, min(100, 100 - mape * 2)), 2) stabilitas = ('optimal' if mape < 2 else 'baik' if mape < 5 else 'cukup' if mape < 10 else 'perlu_retrain') mape_label = ('Sangat Rendah' if mape < 5 else 'Rendah' if mape < 10 else 'Sedang' if mape < 15 else 'Tinggi' if mape < 25 else 'Sangat Tinggi') rmse_pct = rmse / mean_price * 100 rmse_label = ('Presisi Sangat Tinggi' if rmse_pct < 1 else 'Presisi Tinggi' if rmse_pct < 2 else 'Presisi Sedang' if rmse_pct < 5 else 'Presisi Rendah') cl_label = ('Sangat Tinggi' if confidence >= 90 else 'Tinggi' if confidence >= 80 else 'Sedang' if confidence >= 65 else 'Rendah') vol_label = ('Rendah' if cv < 2 else 'Sedang' if cv < 5 else 'Tinggi') if mape < 10: status = {'judul': 'Akurasi Terverifikasi', 'deskripsi': f'Prediksi divalidasi, deviasi rata-rata di bawah {mape:.1f}%.'} elif mape < 20: status = {'judul': 'Akurasi Cukup', 'deskripsi': f'Deviasi rata-rata {mape:.1f}%. Gunakan sebagai referensi.'} else: status = {'judul': 'Perlu Perhatian', 'deskripsi': f'Deviasi {mape:.1f}%. Disarankan retraining.'} tren_map = { 'stabil' : ('positif', 'check-circle', f'Harga {komoditas_nama} diprediksi stabil 7 hari ke depan.'), 'naik_terkendali' : ('negatif', 'trending-up', f'Harga {komoditas_nama} diprediksi naik {abs(pct):.1f}%.'), 'naik_tajam' : ('negatif', 'alert-triangle', f'Harga {komoditas_nama} diprediksi naik tajam {abs(pct):.1f}%.'), 'turun_terkendali': ('positif', 'trending-down', f'Harga {komoditas_nama} diprediksi turun {abs(pct):.1f}%.'), 'turun_tajam' : ('netral', 'alert-triangle', f'Harga {komoditas_nama} diprediksi turun tajam {abs(pct):.1f}%.'), } t1, ikon1, k1 = tren_map.get(tren, ('netral', 'info', f'Tren: {tren}')) peak = max(preds, key=lambda x: x['harga_prediksi']) trough = min(preds, key=lambda x: x['harga_prediksi']) insights = [ {'tipe': t1, 'ikon': ikon1, 'konten': k1, 'urutan': 1}, { 'tipe' : 'positif' if cv < 2 else ('netral' if cv < 5 else 'negatif'), 'ikon' : 'check-circle' if cv < 2 else 'info', 'konten': ('Pasokan terpantau mencukupi.' if cv < 2 else f'Volatilitas sedang (CV={cv:.1f}%).' if cv < 5 else f'Harga sangat fluktuatif (CV={cv:.1f}%).'), 'urutan': 2, }, { 'tipe' : 'netral', 'ikon' : 'calendar', 'konten': (f"Harga tertinggi Rp {peak['harga_prediksi']:,.0f}, " f"terendah Rp {trough['harga_prediksi']:,.0f} dalam 7 hari."), 'urutan': 3, }, ] return { 'tren' : tren, 'pct_change' : round(pct, 2), 'confidence_level': confidence, 'confidence_label': cl_label, 'stabilitas' : stabilitas, 'mape_label' : mape_label, 'rmse_label' : rmse_label, 'volatilitas_label': vol_label, 'status_analisis' : status, 'insights' : insights, 'harga_min' : min(harga_vals), 'harga_max' : max(harga_vals), } # ══════════════════════════════════════════════════════════════════════════════ # 6. MAIN PIPELINE — dipanggil oleh API # ══════════════════════════════════════════════════════════════════════════════ def run_pipeline(komoditas_id: int, komoditas_nama: str, df: pd.DataFrame, target_market: str, pasar_id: int, n_trials: int = 50, forecast_days: int = 7, log_cb=None): """ Full pipeline: cleaning → training → tuning → forecasting → labeling → save. log_cb: callback function(msg: str) untuk streaming log ke frontend. """ def log(msg): if log_cb: log_cb(msg) other_cols = [c for c in df.columns if c not in ['date', target_market] and not c.endswith('_was_missing')] # ── Cleaning ── log("🧹 Cleaning data...") all_cols = [target_market] + other_cols quality_reports = {} for col in all_cols: if col in df.columns: q = analyze_data_quality(df, col) quality_reports[col] = q df = clean_price_data(df, col, q) models_to_run = recommend_models(quality_reports.get(target_market, {})) log(f"🤖 Model yang akan dijalankan: {models_to_run}") # ── Prepare dataset ── exclude = ['date'] + [c for c in df.columns if c.endswith('_was_missing')] df_feat = create_features(df, target_market, other_cols) df_clean = df_feat.dropna(subset=[target_market]).reset_index(drop=True) feature_cols = [c for c in df_clean.columns if c not in exclude + all_cols] X = df_clean[feature_cols] y = df_clean[target_market] dates = df_clean['date'] split = int(len(df_clean) * 0.7) X_train, X_test = X.iloc[:split], X.iloc[split:] y_train, y_test = y.iloc[:split], y.iloc[split:] mean_price = float(y.mean()) # ── Train ── all_results = {} lgb_model = xgb_model = prophet_model = None if 'lightgbm' in models_to_run: log("Training LightGBM...") lgb_model, res = train_lightgbm(X_train, y_train, X_test, y_test) all_results['LightGBM'] = {**res, 'obj': lgb_model, 'type': 'tree'} if 'xgboost' in models_to_run: log("Training XGBoost...") xgb_model, res = train_xgboost(X_train, y_train, X_test, y_test) all_results['XGBoost'] = {**res, 'obj': xgb_model, 'type': 'tree'} if 'prophet' in models_to_run: log("Training Prophet...") try: prophet_model, res = train_prophet(dates.iloc[:split], y_train, dates.iloc[split:], y_test) all_results['Prophet'] = {**res, 'obj': prophet_model, 'type': 'prophet'} except Exception as e: log(f"⚠️ Prophet gagal: {e}") if 'arima' in models_to_run: log("Training ARIMA...") try: arima_model, res = train_arima(y_train, y_test) all_results['ARIMA'] = {**res, 'obj': arima_model, 'type': 'arima'} except Exception as e: log(f"⚠️ ARIMA gagal: {e}") if 'sarima' in models_to_run: log("Training SARIMA...") try: sarima_model, res = train_sarima(y_train, y_test) all_results['SARIMA'] = {**res, 'obj': sarima_model, 'type': 'sarima'} except Exception as e: log(f"⚠️ SARIMA gagal: {e}") best_name = min(all_results, key=lambda k: all_results[k]['rmse']) best = all_results[best_name] log(f"🏆 Best model: {best_name} (RMSE={best['rmse']:.2f})") # ── Tuning ── final_model = best['obj'] final_result = best final_name = best_name if best['type'] == 'tree': log(f"🔧 Tuning {best_name} dengan Optuna ({n_trials} trials)...") best_params = tune_with_optuna(best_name, X_train, y_train, n_trials) if best_name == 'LightGBM': tuned = lgb.LGBMRegressor(**best_params, objective='regression', verbose=-1, random_state=42, force_col_wise=True) else: tuned = xgb.XGBRegressor(**best_params, objective='reg:squarederror', random_state=42, verbosity=0) tuned.fit(X_train, y_train) tuned_res = evaluate_model(y_test, tuned.predict(X_test), f"{best_name}_Tuned") if tuned_res['rmse'] < best['rmse']: final_model = tuned final_result = tuned_res final_name = f"{best_name}_Tuned" log(f"✅ Tuned lebih baik, improvement={(best['rmse']-tuned_res['rmse'])/best['rmse']*100:.1f}%") # ── Forecast ── n_fc = forecast_days log(f"📅 Forecasting {n_fc} hari...") df_hist = df[['date', target_market] + [c for c in other_cols if c in df.columns]].copy() f_type = final_result.get('type', '') if f_type == 'prophet' or final_name == 'Prophet': predictions = forecast_prophet(final_model, df_hist['date'].max(), n_fc) elif f_type == 'arima': predictions = forecast_arima(final_model, n_fc) elif f_type == 'sarima': predictions = forecast_sarima(final_model, n_fc) else: predictions = forecast_tree(final_model, df_hist, target_market, other_cols, feature_cols, n_fc) # ── Labels ── cv = quality_reports.get(target_market, {}).get('cv', 0) labels = get_labels( final_result['mape'], final_result['rmse'], mean_price, cv, predictions, komoditas_nama ) # ── Simpan model ── prefix = os.path.join(MODEL_DIR, f"model_{komoditas_id}_{pasar_id}") model_path= f"{prefix}.pkl" with open(model_path, 'wb') as f: pickle.dump(final_model, f) with open(f"{prefix}_features.json", 'w') as f: json.dump(feature_cols, f) metadata = { 'komoditas_id' : komoditas_id, 'komoditas_nama' : komoditas_nama, 'pasar_id' : pasar_id, 'target_market' : target_market, 'other_markets' : other_cols, 'nama_model' : final_name, 'versi' : '1.0', 'file_path' : model_path, 'mape' : round(final_result['mape'], 4), 'rmse' : round(final_result['rmse'], 4), 'mae' : round(final_result['mae'], 4), 'r2_score' : round(final_result['r2'], 4), 'confidence_level': labels['confidence_level'], 'stabilitas' : labels['stabilitas'], 'status_validasi' : 'terverifikasi', 'catatan_validasi': labels['status_analisis']['deskripsi'], 'tanggal_training': date.today().isoformat(), 'tanggal_evaluasi': date.today().isoformat(), 'deskripsi' : f"Model {final_name} untuk prediksi harga {komoditas_nama}.", 'models_tried' : list(all_results.keys()), 'data_quality' : quality_reports.get(target_market, {}), } with open(f"{prefix}_metadata.json", 'w') as f: json.dump(metadata, f, indent=2) log(f"✅ Pipeline selesai! MAPE={final_result['mape']:.2f}%") return { 'predictions' : predictions, 'labels' : labels, 'metadata' : metadata, 'model_comparison': {k: { 'rmse': round(v['rmse'], 4), 'mape': round(v['mape'], 4), 'mae': round(v['mae'], 4), 'r2': round(v['r2'], 4) } for k, v in all_results.items()}, }