Spaces:
Paused
Paused
| 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() | |