File size: 2,749 Bytes
73b58b3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
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()
|