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()