|
import requests |
|
import hmac |
|
import hashlib |
|
import time |
|
import pandas as pd |
|
import xgboost as xgb |
|
from sklearn.model_selection import train_test_split |
|
|
|
|
|
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']: |
|
|
|
order = place_order('buy', 0.001, df.iloc[-1]['close']) |
|
print(f'Placed buy order: {order}') |
|
else: |
|
|
|
order = place_order('sell', 0.001, df.iloc[-1]['close']) |
|
print(f'Placed sell order: {order}') |
|
|
|
|
|
time.sleep(60) |
|
|
|
if __name__ == '__main__': |
|
main() |
|
|
