Skip to content

Instantly share code, notes, and snippets.

@jongan69
Last active October 30, 2024 01:07
Show Gist options
  • Save jongan69/92c430703cf261ae0701f4fed99d7212 to your computer and use it in GitHub Desktop.
Save jongan69/92c430703cf261ae0701f4fed99d7212 to your computer and use it in GitHub Desktop.
Lockin Bot??? RSI??? someone plz shoot me
OX_API_KEY=
OX_API_SECRET=
import os
from dotenv import load_dotenv
import requests
import numpy as np
import time
import hmac
import base64
import hashlib
import datetime
import json
import logging
# Load environment variables
load_dotenv()
# API credentials using https://ox.fun:
# https://ox.fun/register?shareAccountId=5MU57aDG
api_key = os.getenv("OX_API_KEY")
api_secret = os.getenv("OX_API_SECRET")
rest_url = 'https://api.ox.fun'
rest_path = 'api.ox.fun'
# Trading parameters
asset = 'LOCKIN-USD-SWAP-LIN'
rsi_period = 14
overbought = 70
oversold = 30
quantity = 10
timeframe = '60'
limit = 200
# Configure logging
logging.basicConfig(level=logging.INFO)
def generate_signature(api_secret, ts, nonce, verb, method, body):
msg_string = '{}\n{}\n{}\n{}\n{}\n{}'.format(ts, nonce, verb, rest_path, method, body)
sig = base64.b64encode(hmac.new(api_secret.encode('utf-8'), msg_string.encode('utf-8'), hashlib.sha256).digest()).decode('utf-8')
return sig
def get_headers(api_key, api_secret, verb, method, body):
ts = datetime.datetime.now(datetime.timezone.utc).isoformat()
nonce = int(time.time() * 1000)
sig = generate_signature(api_secret, ts, nonce, verb, method, body)
headers = {
'Content-Type': 'application/json',
'AccessKey': api_key,
'Timestamp': ts,
'Signature': sig,
'Nonce': str(nonce)
}
return headers
def fetch_historical_data():
method = f'v3/candles?marketCode={asset}&timeframe={timeframe}&limit={limit}'
try:
response = requests.get(f'{rest_url}/{method}')
response.raise_for_status()
data = response.json()
if 'data' in data and isinstance(data['data'], list):
close_prices = [float(candle['close']) for candle in data['data']]
return close_prices
else:
logging.error("Error: 'data' key not found in the response or is not a list")
logging.error(data)
return []
except requests.exceptions.RequestException as e:
logging.error(f"Request error: {e}")
return []
def calculate_rsi(prices, period):
deltas = np.diff(prices)
seed = deltas[:period+1]
up = seed[seed >= 0].sum()/period
down = -seed[seed < 0].sum()/period
rs = up/down
rsi = np.zeros_like(prices)
rsi[:period] = 100. - 100./(1. + rs)
for i in range(period, len(prices)):
delta = deltas[i-1]
if delta > 0:
upval = delta
downval = 0.
else:
upval = 0.
downval = -delta
up = (up*(period-1) + upval)/period
down = (down*(period-1) + downval)/period
rs = up/down
rsi[i] = 100. - 100./(1. + rs)
return rsi
def place_order(side, price):
method = '/v3/orders/place'
ts = datetime.datetime.now(datetime.timezone.utc).isoformat()
nonce = int(time.time() * 1000)
body = json.dumps({
"recvWindow": 20000,
"responseType": "FULL",
"timestamp": int(datetime.datetime.now().timestamp() * 1000),
"orders": [
{
"clientOrderId": str(nonce),
"marketCode": asset,
"side": side,
"quantity": str(quantity),
"orderType": "MARKET",
"price": str(price)
}
]
})
headers = get_headers(api_key, api_secret, "POST", method, body)
try:
# if body:
# path = method + '?' + body
# else:
# path = method
response = requests.post(rest_url + method, data=body, headers=headers)
response.raise_for_status()
logging.info(response.json())
except requests.exceptions.RequestException as e:
logging.error(f"Order placement error: {e}")
def main():
# Place an initial buy order for testing purposes
try:
close_prices = fetch_historical_data()
if close_prices:
logging.info("Placing initial BUY order for testing purposes")
place_order("BUY", close_prices[-1])
except Exception as e:
logging.error(f"Error placing initial order: {e}")
while True:
try:
close_prices = fetch_historical_data()
if len(close_prices) > rsi_period:
rsi = calculate_rsi(close_prices, rsi_period)[-1]
logging.info(f"Current RSI: {rsi}")
if rsi > overbought:
logging.info("RSI indicates overbought, placing SELL order")
place_order("SELL", close_prices[-1])
print("Market is bearish")
elif rsi < oversold:
logging.info("RSI indicates oversold, placing BUY order")
place_order("BUY", close_prices[-1])
print("Market is bullish")
else:
print("Market is neutral")
time.sleep(30) # Sleep for 30 seconds before the next fetch
except Exception as e:
logging.error(f"Error: {e}")
time.sleep(5)
if __name__ == "__main__":
main()
@jongan69
Copy link
Author

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment