Delta Exchange API Base URL
BASE_URL = “https://api.india.delta.exchange”
Function to fetch historical OHLC data from Delta Exchange
def fetch_historical_prices(symbol=“BTCUSD”, resolution=“1m”, limit=200):
url = f"{BASE_URL}/v2/history/candles"
end_time = int(time.time()) # Current time in UNIX timestamp
start_time = end_time - (limit * 60) # Fetch last 'limit' number of 1-minute candles
params = {
"symbol": symbol,
"resolution": resolution,
"start": start_time,
"end": end_time
}
headers = {
'api-key': api_key # API Key for authentication
}
try:
response = requests.get(url, params=params, headers=headers)
data = response.json()
if response.status_code == 200 and 'result' in data and isinstance(data['result'], list):
df = pd.DataFrame(data['result'])
if not df.empty:
df = df.sort_values(by="time") # Sort by timestamp
df[['open', 'high', 'low', 'close']] = df[['open', 'high', 'low', 'close']].astype(float)
return df
else:
print(f"API Error: {data}")
return None
except requests.exceptions.RequestException as e:
print(f"API Request Failed: {e}")
return None
Function to calculate RSI using TA-Lib
def calculate_rsi(prices, period=14):
if len(prices) < period:
return None # Not enough data to calculate RSI
prices = np.array(prices, dtype=np.float64) # Ensure correct data type
rsi = talib.RSI(prices, timeperiod=period)
if np.isnan(rsi[-1]): # Ensure valid RSI value
return None
return rsi[-1]
Generate API signature
def generate_signature(method, endpoint, payload):
timestamp = str(int(time.time()))
signature_data = method + timestamp + endpoint + payload
message = bytes(signature_data, ‘utf-8’)
secret = bytes(api_secret, ‘utf-8’)
hash = hmac.new(secret, message, hashlib.sha256)
return hash.hexdigest(), timestamp
Trading Logic: Wait until RSI crosses above 60
while True:
df = fetch_historical_prices()
if df is not None and len(df) > 14: # Ensure enough data for RSI calculation
rsi_value = calculate_rsi(df['close'].tolist())
latest_candle = df.iloc[-1] # Get the latest candle
print(f"\nLatest Candle: Time: {latest_candle['time']}, Open: {latest_candle['open']}, High: {latest_candle['high']}, Low: {latest_candle['low']}, Close: {latest_candle['close']}")
print(f"Current RSI: {rsi_value}")
if rsi_value is not None and rsi_value > 60:
print("\n✅ RSI is above 60, placing a BUY order...")
# Prepare order data
order_data = {
'product_id': 27, # BTCUSD product ID (change if needed)
'size': 1,
'order_type': 'market_order',
'side': 'buy'
}
body = json.dumps(order_data, separators=(',', ':'))
method = 'POST'
endpoint = '/v2/orders'
signature, timestamp = generate_signature(method, endpoint, body)
# API headers for order placement
headers = {
'api-key': api_key,
'signature': signature,
'timestamp': timestamp,
'Content-Type': 'application/json'
}
# Send buy order
try:
response = requests.post(f"{BASE_URL}/v2/orders", headers=headers, json=order_data)
order_response = response.json()
print("\n🚀 Order Placed Successfully:", order_response)
break # Exit loop after successful buy order
except requests.exceptions.RequestException as e:
print(f"⚠️ Order API Request Failed: {e}")
else:
print("⏳ RSI is below 60. Retrying in 60 seconds...")
time.sleep(30) # Wait before checking again
else:
print("⚠️ Failed to fetch historical BTCUSD price data. Retrying in 60 seconds...")
time.sleep(30) # Wait before retrying