Create Sosis.py

Sosis.py

@@ -0,0 +1,83 @@
+import requests
+import hmac
+import hashlib
+import time
+import pandas as pd
+import xgboost as xgb
+from sklearn.model_selection import train_test_split
+
+# Replace with your own API key and secret
+api_key = 'your_api_key'
+api_secret = 'your_api_secret'
+
+def create_signature(query_string, secret):
+    return hmac.new(secret.encode('utf-8'), query_string.encode('utf-8'), hashlib.sha256).digest()
+
+def place_order(side, amount, price):
+    query = f'access_id={api_key}&amount={amount}&price={price}&side={side}&type=limit'
+    signature = create_signature(query, api_secret)
+
+    headers = {
+        'Content-Type': 'application/json',
+        'Authorization': f'{api_key}:{signature.hex()}'
+    }
+
+    response = requests.post('https://api.coinex.com/v1/order/place', data=query, headers=headers)
+    return response.json()
+
+def get_data():
+    response = requests.get('https://api.coinex.com/v1/market/kline/BTCUSDT?interval=1h&limit=60')
+    data = response.json()
+    df = pd.DataFrame(data['data'])
+    df['time'] = pd.to_datetime(df['time'], unit='ms')
+    df.set_index('time', inplace=True)
+    df.drop(['vol', 'amount'], axis=1, inplace=True)
+    return df
+
+def train_model(df):
+    X = df.drop('close', axis=1)
+    y = df['close']
+    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False)
+    dtrain = xgb.DMatrix(X_train, label=y_train)
+    dtest = xgb.DMatrix(X_test, label=y_test)
+    params = {
+        'objective': 'reg:squarederror',
+        'eval_metric': 'rmse',
+        'max_depth': 3,
+        'eta': 0.1
+    }
+    model = xgb.train(params, dtrain)
+    predictions = model.predict(dtest)
+    accuracy = 1 - (sum((predictions - y_test) ** 2) / sum(y_test ** 2))
+    return model, accuracy
+
+def predict_next_hour(model, df):
+    last_row = df.iloc[-1]
+    X = pd.DataFrame([[last_row['open'], last_row['high'], last_row['low']]], index=[last_row.name + pd.Timedelta(hours=1)])
+    dtest = xgb.DMatrix(X)
+    prediction = model.predict(dtest)[0]
+    return prediction
+
+def main():
+    while True:
+        df = get_data()
+        model, accuracy = train_model(df)
+        print(f'Model accuracy: {accuracy}')
+
+        if accuracy > 0.9:
+            prediction = predict_next_hour(model, df)
+
+            if prediction > df.iloc[-1]['close']:
+                # Place buy order
+                order = place_order('buy', 0.001, df.iloc[-1]['close'])
+                print(f'Placed buy order: {order}')
+            else:
+                # Place sell order
+                order = place_order('sell', 0.001, df.iloc[-1]['close'])
+                print(f'Placed sell order: {order}')
+
+        # Wait for some time before checking the price again
+        time.sleep(60)
+
+if __name__ == '__main__':
+    main()