Spaces:
Paused
Paused
File size: 5,859 Bytes
96e0cc2 |
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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 |
import requests
import time
import hashlib
import hmac
import uuid
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("BYBIT_API_KEY")
secret_key = os.getenv("BYBIT_API_SECRET")
httpClient = requests.Session()
recv_window = str(5000)
url = "https://api.bybit.com"
def HTTP_Request(endPoint, method, payload, Info):
global time_stamp
time_stamp = str(int(time.time() * 10 ** 3))
signature = genSignature(payload)
headers = {
'X-BAPI-API-KEY': api_key,
'X-BAPI-SIGN': signature,
'X-BAPI-SIGN-TYPE': '2',
'X-BAPI-TIMESTAMP': time_stamp,
'X-BAPI-RECV-WINDOW': recv_window,
'Content-Type': 'application/json'
}
if(method == "POST"):
response = httpClient.request(method, url + endPoint, headers=headers, data=payload)
else:
response = httpClient.request(method, url + endPoint + "?" + payload, headers=headers)
print(f"\n{Info}:")
print(f"Status Code: {response.status_code}")
print(f"Response: {response.text}")
print(f"Elapsed Time: {response.elapsed}")
return response
def genSignature(payload):
param_str = str(time_stamp) + api_key + recv_window + payload
hash = hmac.new(bytes(secret_key, "utf-8"), param_str.encode("utf-8"), hashlib.sha256)
signature = hash.hexdigest()
return signature
def test_wallet_balance():
print("Testing Wallet Balance...")
endpoint = "/v5/account/wallet-balance"
method = "GET"
params = 'accountType=UNIFIED'
response = HTTP_Request(endpoint, method, params, "Wallet Balance")
# Parse response to check for specific errors
try:
import json
resp_data = json.loads(response.text) if response.text else {}
ret_code = resp_data.get('retCode', -1)
ret_msg = resp_data.get('retMsg', 'Unknown error')
if ret_code == 0:
return True
else:
print(f" API Error: {ret_code} - {ret_msg}")
return False
except:
return response.status_code == 200
def test_positions():
print("Testing Positions...")
endpoint = "/v5/position/list"
method = "GET"
params = 'category=linear&settleCoin=USDT'
response = HTTP_Request(endpoint, method, params, "Positions")
return parse_api_response(response)
def test_create_order():
print("Testing Order Creation...")
print("Note: Order creation may fail due to symbol whitelisting or insufficient balance")
print("This is normal for testing - the bot will handle this gracefully")
endpoint = "/v5/order/create"
method = "POST"
orderLinkId = uuid.uuid4().hex
# Try market order with minimum viable amount
params = f'{{"category":"linear","symbol":"BTCUSDT","side":"Buy","orderType":"Limit","qty":"0.001","price":"84100","timeInForce":"GTC","orderLinkId":"{orderLinkId}"}}'
response = HTTP_Request(endpoint, method, params, "Create Order (Market)")
success = parse_api_response(response)
if not success:
print(" Market order failed - this is expected if symbol not whitelisted or insufficient balance")
print(" The bot can still function for monitoring and analysis")
return success, orderLinkId
def test_get_orders():
print("Testing Get Orders...")
endpoint = "/v5/order/realtime"
method = "GET"
params = 'category=linear&settleCoin=USDT'
response = HTTP_Request(endpoint, method, params, "Get Orders")
return parse_api_response(response)
def parse_api_response(response):
"""Parse API response and return success status"""
try:
import json
resp_data = json.loads(response.text) if response.text else {}
ret_code = resp_data.get('retCode', -1)
ret_msg = resp_data.get('retMsg', 'Unknown error')
if ret_code == 0:
return True
else:
print(f" API Error: {ret_code} - {ret_msg}")
return False
except Exception as e:
print(f" Parse Error: {e}")
return response.status_code == 200
def main():
print("π Bybit API Raw HTTP Test")
print("=" * 50)
if not api_key or not secret_key:
print("β API credentials not found!")
return
print(f"API Key: {api_key[:10]}...")
print(f"Secret Key: {secret_key[:10]}...")
print(f"Testnet URL: {url}")
tests = [
("Wallet Balance", test_wallet_balance),
("Positions", test_positions),
("Get Orders", test_get_orders),
]
results = []
for test_name, test_func in tests:
try:
success = test_func()
results.append((test_name, success))
print(f"β
{test_name}: {'PASS' if success else 'FAIL'}")
except Exception as e:
print(f"β {test_name}: ERROR - {e}")
results.append((test_name, False))
# Test order creation (optional)
try:
print("\nTesting Order Creation (will fail safely if no balance)...")
order_success, order_id = test_create_order()
results.append(("Create Order", order_success))
except Exception as e:
print(f"β Create Order: ERROR - {e}")
results.append(("Create Order", False))
print("\n" + "=" * 50)
print("π Test Results:")
passed = sum(1 for _, success in results if success)
total = len(results)
for test_name, success in results:
status = "β
PASS" if success else "β FAIL"
print(f" {test_name}: {status}")
print(f"\nπ― Overall: {passed}/{total} tests passed")
if passed == total:
print("π All API tests passed! Ready for trading.")
elif passed > 0:
print("β οΈ Some tests passed. Check permissions and account setup.")
else:
print("β All tests failed. Check API credentials and permissions.")
if __name__ == "__main__":
main()
|