-
-
Save jongan69/92c430703cf261ae0701f4fed99d7212 to your computer and use it in GitHub Desktop.
Lockin Bot??? RSI??? someone plz shoot me
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
OX_API_KEY= | |
OX_API_SECRET= |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
https://ox.fun/register?shareAccountId=5MU57aDG