Update app.py

app.py

@@ -1,4 +1,5 @@
 import gradio as gr
+from fastapi import FastAPI, HTTPException
 import ccxt
 import pandas as pd
 import pandas_ta as ta
@@ -7,205 +8,124 @@ from sklearn.ensemble import RandomForestClassifier
 from sklearn.preprocessing import MinMaxScaler
 import joblib
 import os
-import json
-from datetime import datetime
+from pydantic import BaseModel
 
-# Konfigurasi
+app = FastAPI(title="Crypto ML Strategy (Paper 2105.06827v3)")
+
+# 🔧 Konfigurasi
 SYMBOL = "ETH/USDT"
 TIMEFRAME = "4h"
-FEE = 0.0015
-WINDOW = 60
+FEE = 0.0015        # 0.15% sesuai paper
+WINDOW = 60         # 60 candle * 4h = 240h lookback
 MODEL_PATH = "rf_model.pkl"
 SCALER_PATH = "scaler.pkl"
 
+# 📥 1. Ambil Data Historis
 def fetch_data(symbol: str, timeframe: str, limit: int = 2000) -> pd.DataFrame:
-    """Ambil data dari Binance"""
-    try:
-        exchange = ccxt.binance({'enableRateLimit': True})
-        ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
-        df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
-        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
-        return df
-    except Exception as e:
-        return pd.DataFrame()
+    exchange = ccxt.binance()
+    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
+    df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
+    df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
+    return df
 
+# 🧮 2. Tambah Indikator & Siapkan Fitur/Label
 def prepare_dataset(df: pd.DataFrame):
-    """Siapkan dataset dengan indikator teknikal"""
-    if df.empty:
-        return None, None, None, None, None
-        
     df = df.copy()
-    
-    try:
-        df.ta.rsi(length=14, append=True)
-        df.ta.macd(fast=12, slow=26, signal=9, append=True)
-        df.ta.bbands(length=20, append=True)
-        df.ta.cci(length=14, append=True)
-    except Exception as e:
-        return None, None, None, None, None
-    
+    # Indikator teknikal (sesuai Appendix A paper)
+    df.ta.rsi(append=True)
+    df.ta.macd(append=True)
+    df.ta.bbands(append=True)
+    df.ta.cci(append=True)
     df.dropna(inplace=True)
-    
+
+    # Label: 1 jika pergerakan 4h ke depan > fee (0.15%), else 0
     df["next_close"] = df["close"].shift(-1)
     df["label"] = (df["next_close"] > df["close"] * (1 + FEE)).astype(int)
     df.dropna(inplace=True)
-    
+
+    # Buat sliding window (60 candle per sampel)
     X_list, y_list, idx_list = [], [], []
-    feature_cols = [col for col in df.columns if col not in ["timestamp", "next_close", "label"]]
-    
     for i in range(WINDOW, len(df)):
-        window = df.iloc[i-WINDOW:i][feature_cols]
+        window = df.iloc[i-WINDOW:i].drop(columns=["timestamp", "next_close", "label"])
         X_list.append(window.values.flatten())
         y_list.append(df.iloc[i]["label"])
         idx_list.append(i)
-    
+
     X = np.array(X_list)
     y = np.array(y_list)
     scaler = MinMaxScaler()
     X_scaled = scaler.fit_transform(X)
-    
     return X_scaled, y, idx_list, df, scaler
 
-def train_model():
-    """Training model dan backtest"""
+# 📊 3. FastAPI Endpoints
+@app.get("/train")
+def train_and_backtest():
     if os.path.exists(MODEL_PATH):
-        return "Model sudah ada. Hapus file model untuk retrain."
-    
+        return {"message": "Model sudah ada. Hapus model.pkl & scaler.pkl untuk retrain."}
+
     df = fetch_data(SYMBOL, TIMEFRAME)
-    if df.empty:
-        return "❌ Gagal mengambil data"
-    
     X, y, idx, df_full, scaler = prepare_dataset(df)
-    if X is None:
-        return "❌ Gagal menyiapkan dataset"
-    
+
+    # Split 95% train / 5% test (sesuai paper)
     split = int(len(X) * 0.95)
     X_train, X_test = X[:split], X[split:]
     y_train, y_test = y[:split], y[split:]
-    
-    model = RandomForestClassifier(n_estimators=500, max_depth=None, random_state=42, n_jobs=-1)
+
+    # Training
+    model = RandomForestClassifier(n_estimators=500, max_depth=None, random_state=42)
     model.fit(X_train, y_train)
     acc = float(model.score(X_test, y_test))
-    
+
     joblib.dump(model, MODEL_PATH)
     joblib.dump(scaler, SCALER_PATH)
-    
-    # Backtest
+
+    # 🧪 Backtest pada data unseen (5% terakhir)
     test_indices = idx[split:]
     df_test = df_full.loc[test_indices].copy()
     df_test["pred"] = model.predict(X_test)
-    
+
     pnl_list = []
     for i in range(len(df_test) - 1):
         if df_test.iloc[i]["pred"] == 1:
             entry = df_test.iloc[i]["close"]
             exit_ = df_test.iloc[i + 1]["close"]
             ret = (exit_ - entry) / entry
-            pnl = ret - FEE
+            pnl = ret - FEE  # Kurangi fee exchange
             pnl_list.append(pnl)
-    
+
     wins = sum(1 for p in pnl_list if p > 0)
     losses = len(pnl_list) - wins
     gross_profit = sum(p for p in pnl_list if p > 0)
     gross_loss = sum(abs(p) for p in pnl_list if p <= 0)
     profit_factor = gross_profit / gross_loss if gross_loss > 0 else float("inf")
-    
-    result = f"""
-    ✅ **Training Selesai!**
-    
-    📊 **Hasil Backtest (66 hari):**
-    - Accuracy: {acc:.2%}
-    - Profit Factor: {profit_factor:.3f}
-    - Net Profit: {sum(pnl_list)*100:.2f}%
-    - Win Rate: {wins/len(pnl_list)*100:.2f}%
-    - Total Trades: {len(pnl_list)}
-    - Wins: {wins} | Losses: {losses}
-    """
-    return result
 
+    return {
+        "accuracy_test": round(acc, 4),
+        "profit_factor": round(profit_factor, 3),
+        "net_profit_pct": round(sum(pnl_list) * 100, 3),
+        "wins": wins,
+        "losses": losses,
+        "win_rate_pct": round(wins / len(pnl_list) * 100, 2) if pnl_list else 0,
+        "message": "Model trained & backtested on unseen data"
+    }
+
+@app.get("/predict")
 def predict_next_move():
-    """Prediksi pergerakan harga berikutnya"""
     if not os.path.exists(MODEL_PATH):
-        return "❌ Model belum dilatih. Klik tombol 'Train Model' terlebih dahulu."
+        raise HTTPException(400, "Model belum dilatih. Panggil /train terlebih dahulu.")
     
     df = fetch_data(SYMBOL, TIMEFRAME, limit=WINDOW + 10)
     _, _, _, df_full, scaler = prepare_dataset(df)
     
-    if df_full is None or len(df_full) < WINDOW:
-        return "❌ Data tidak cukup untuk prediksi"
-    
-    feature_cols = [col for col in df_full.columns if col not in ["timestamp", "next_close", "label"]]
-    latest_window = df_full.iloc[-WINDOW:][feature_cols].values.flatten().reshape(1, -1)
+    latest_window = df_full.iloc[-WINDOW:].drop(columns=["timestamp", "next_close", "label"]).values.flatten().reshape(1, -1)
     latest_scaled = scaler.transform(latest_window)
     
     model = joblib.load(MODEL_PATH)
     pred = int(model.predict(latest_scaled)[0])
-    proba = model.predict_proba(latest_scaled)[0]
-    
-    signal = "🟢 **BUY** (Hold 4 jam)" if pred == 1 else "🔴 **HOLD/SELL**"
-    
-    result = f"""
-    📈 **Prediksi {SYMBOL} - {TIMEFRAME}**
-    
-    Signal: {signal}
-    
-    **Probabilitas:**
-    - Naik (>0.15%): {proba[1]*100:.2f}%
-    - Turun/Stagnan: {proba[0]*100:.2f}%
-    - Confidence: {max(proba)*100:.2f}%
-    
-    **Waktu Prediksi:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
-    """
-    return result
-
-def get_price_chart():
-    """Generate price chart data"""
-    df = fetch_data(SYMBOL, TIMEFRAME, limit=200)
-    if df.empty:
-        return None
-    
-    # Buat simple chart data
-    chart_data = df[['timestamp', 'close']].tail(100).copy()
-    return chart_data
-
-# Gradio Interface
-with gr.Blocks(title="Crypto ML Trading Strategy", theme=gr.themes.Soft()) as demo:
-    gr.Markdown("""
-    #  Crypto ML Trading Strategy
-    **Paper Reference:** [2105.06827v3](https://arxiv.org/abs/2105.06827)
-    
-    Prediksi arah harga cryptocurrency menggunakan Machine Learning (Random Forest) dengan indikator teknikal.
-    """)
-    
-    with gr.Row():
-        with gr.Column():
-            gr.Markdown("### 📊 Kontrol")
-            train_btn = gr.Button("🎯 Train Model", variant="primary")
-            predict_btn = gr.Button("🔮 Prediksi Sekarang")
-            chart_btn = gr.Button("📈 Lihat Chart")
-        
-        with gr.Column():
-            gr.Markdown("### 💡 Informasi")
-            gr.Markdown("""
-            - **Asset:** ETH/USDT
-            - **Timeframe:** 4 Jam
-            - **Model:** Random Forest
-            - **Threshold:** 0.15% (fee Binance)
-            - **Data:** 2017-2021
-            """)
-    
-    output = gr.Textbox(label="Hasil", lines=15)
-    chart_output = gr.Dataframe(label="Price Data")
-    
-    train_btn.click(fn=train_model, inputs=None, outputs=output)
-    predict_btn.click(fn=predict_next_move, inputs=None, outputs=output)
-    chart_btn.click(fn=get_price_chart, inputs=None, outputs=chart_output)
     
-    gr.Markdown("""
-    ---
-    **Disclaimer:** Ini adalah proof-of-concept akademis. Bukan saran finansial. 
-    Trading cryptocurrency berisiko tinggi.
-    """)
-
-if __name__ == "__main__":
-    demo.launch()
+    return {
+        "symbol": SYMBOL,
+        "timeframe": TIMEFRAME,
+        "prediction": "BUY (Hold 4h)" if pred == 1 else "HOLD/SELL",
+        "logic": "Prediksi >0.15% movement dalam 4 jam ke depan"
+    }