Create app.py

app.py

@@ -0,0 +1,154 @@
+import pickle
+import gradio as gr
+import matplotlib.pyplot as plt
+import ccxt
+import pandas as pd
+import numpy as np
+from sklearn.preprocessing import MinMaxScaler
+from sklearn.impute import SimpleImputer
+from scipy import stats
+
+# Fetch data
+def fetch_binance_data(symbol, timeframe, limit=2000):
+    binance = ccxt.binance()
+    ohlcv = binance.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
+
+# Rolling Window Normalizer
+class RollingWindowNormalizer:
+    def __init__(self, window=24):
+        self.window = window
+        self.stats_ = {}
+    def fit(self, X, columns):
+        for column in columns:
+            rolling_mean = X[column].rolling(window=self.window).mean()
+            rolling_std = X[column].rolling(window=self.window).std()
+            self.stats_[column] = {'rolling_mean': rolling_mean, 'rolling_std': rolling_std}
+        return self
+    def transform(self, X, columns):
+        for column in columns:
+            rolling_mean = self.stats_[column]['rolling_mean']
+            rolling_std = self.stats_[column]['rolling_std']
+            X[column] = (X[column] - rolling_mean) / rolling_std
+        X.dropna(inplace=True)
+        return X
+    def fit_transform(self, X, columns):
+        return self.fit(X, columns).transform(X, columns)
+
+def normalize(X, columns=['open','high','low','close']):
+    X_copy = X.copy()
+    Rm = RollingWindowNormalizer()
+    Rm.fit(X_copy, columns)
+    Y = Rm.transform(X_copy, columns)
+    return Y, Rm
+
+# Outlier removal
+def remove_outliers(x, epsilon):
+    z_score = stats.zscore(x['low'])
+    return x[z_score.abs() < epsilon]
+
+# Advanced features
+def calculate_rsi(d, window=14):
+    delta = d.diff()
+    gain = np.where(delta > 0, delta, 0)
+    loss = np.where(delta < 0, -delta, 0)
+    avg_gain = pd.Series(gain).rolling(window=window).mean()
+    avg_loss = pd.Series(loss).rolling(window=window).mean()
+    rs = avg_gain / avg_loss
+    return 100 - (100 / (1 + rs))
+
+def generate_advanced_features(d, other_data=None):
+    d = d.copy()
+    d['ma_7'] = d['close'].rolling(window=7).mean()
+    d['ma_21'] = d['close'].rolling(window=21).mean()
+    d['rsi'] = calculate_rsi(d['close'])
+    d['ma_ratio'] = d['ma_7'] / d['ma_21']
+    for k in ['close','high']:
+        for i in range(1,5):
+            d[f'lag_{k}{i}'] = d[k].shift(i)
+    d['std_last_10'] = d['close'].rolling(window=10).std()
+    if other_data is not None:
+        d['relative_strength'] = d['close'] / other_data['close']
+        d['relative_strength_1'] = d['close'].shift(2) / other_data['close'].shift(2)
+    return d.iloc[:,1:].values
+
+def create_features_and_labels_with_advanced_features(btc, eth):
+    btc_copy = btc.copy()
+    eth_copy = eth.copy()
+    btc_features = generate_advanced_features(btc_copy, eth_copy)
+    eth_features = generate_advanced_features(eth_copy, btc_copy)
+    df = btc.copy()
+    df['future'] = df['close'].rolling(window=5).mean().shift(-1)
+    df['trend'] = (df['future'] > df['close']).astype(int)
+    labels = df['trend'].dropna().values
+    features = np.vstack((btc_features, eth_features))
+    return features, labels
+
+def get_data_predict(btc_ori, eth_ori, symbol='ETH/USDT', timeframe='4h', epsilon=2, normalized=False, limit=50):
+    btc_data_ = fetch_binance_data('BTC/USDT', timeframe, limit=limit)
+    eth_data_ = fetch_binance_data(symbol, timeframe, limit=limit)
+    btc_data_ = remove_outliers(btc_data_, epsilon)
+    eth_data_ = remove_outliers(eth_data_, epsilon)
+    if normalized:
+        btc_data_all = pd.concat([btc_ori, btc_data_]).drop_duplicates(subset='timestamp').reset_index(drop=True)
+        eth_data_all = pd.concat([eth_ori, eth_data_]).drop_duplicates(subset='timestamp').reset_index(drop=True)
+        btc_data_, _ = normalize(btc_data_all)
+        eth_data_, _ = normalize(eth_data_all)
+        label = btc_data_.copy()[['timestamp','close']].shift(-1)
+    return btc_data_, eth_data_, label
+
+def predictions(model, X1, X2, name, n_steps):
+    features_, labels_ = create_features_and_labels_with_advanced_features(X1, X2)
+    imputer = SimpleImputer(strategy='mean')
+    features_imputed = imputer.fit_transform(features_)
+    y = model.predict_proba(features_imputed)[:,1]
+    return y
+
+def plot(y, label, timeframe='1h', ma=5, n_steps=None):
+    if n_steps is None:
+        n_steps = len(y)
+    plt.figure(figsize=(12,6))
+    if ma:
+        df_plot = pd.DataFrame({'date': label['timestamp'].values[-n_steps:], 'prediction':5*(y[-n_steps:]-0.5), 'real': label['close'].values[-n_steps:]})
+        plt.plot(df_plot['date'], df_plot['prediction'].rolling(window=ma).mean(), label='updown')
+        plt.plot(df_plot['date'], df_plot['real'].rolling(window=ma).mean(), label='real')
+        plt.plot(df_plot['date'], (df_plot['real']-df_plot['prediction']).rolling(window=ma).mean(), label='difference')
+    else:
+        plt.plot(label['timestamp'].values[-n_steps:], 5*(y[-n_steps:]-0.5), label='updown')
+        plt.plot(label['timestamp'].values[-n_steps:], label['close'].values[-n_steps:], label='real')
+    plt.axhline(0, linestyle='--')
+    plt.title(f"BTC timeframe {timeframe}")
+    plt.xlabel('Timestamp')
+    plt.ylabel('Values')
+    plt.legend()
+    return plt.gcf()
+
+# Load pre-trained models
+with open('model_n1d_cat.pkl','rb') as f:
+    model_n1d_cat = pickle.load(f)
+with open('model_n4h_cat.pkl','rb') as f:
+    model_n4h_cat = pickle.load(f)
+
+def predict_and_plot(timeframe, limit, epsilon, n_steps, ma):
+    btc_ori = yf.download('BTC-USD', period=f'{limit}d', interval=timeframe)
+    eth_ori = yf.download('ETH-USD', period=f'{limit}d', interval=timeframe)
+    btc_data, eth_data, label = get_data_predict(btc_ori, eth_ori, symbol='ETH/USDT', timeframe=timeframe, epsilon=epsilon, normalized=True, limit=limit)
+    model = model_n1d_cat if timeframe=='1d' else model_n4h_cat
+    preds = predictions(model, btc_data, eth_data, name=timeframe, n_steps=n_steps)
+    fig = plot(preds, label=btc_data, timeframe=timeframe, ma=ma, n_steps=n_steps)
+    return fig
+
+interface = gr.Interface(fn=predict_and_plot,
+                       inputs=[gr.Dropdown(['1d','4h'], label='Timeframe', value='1d'),
+                               gr.Slider(50,500,step=50,value=100,label='Data Limit'),
+                               gr.Slider(0.1,5.0,step=0.1,value=2.0,label='Epsilon'),
+                               gr.Slider(50,500,step=50,value=200,label='N_steps'),
+                               gr.Slider(1,20,step=1,value=5,label='Moving Average Window (ma)')],
+                       outputs=gr.Plot(),
+                       title='BTC Price Movement Prediction',
+                       description='Predict BTC price movements using pre-trained LightGBM models.')
+
+if __name__=='__main__':
+    interface.launch()