test bybit opening/closing orders
This commit is contained in:
Dobromir Popov
220
test_bybit_public_api.py
Normal file
220
test_bybit_public_api.py
Normal file
@ -0,0 +1,220 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test Bybit public API functionality (no authentication required)
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import logging
|
||||
|
||||
# Add the project root to the path
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
from NN.exchanges.bybit_rest_client import BybitRestClient
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def test_public_api():
|
||||
"""Test public API endpoints"""
|
||||
print("=" * 60)
|
||||
print("BYBIT PUBLIC API TEST")
|
||||
print("=" * 60)
|
||||
|
||||
# Test both testnet and live for public endpoints
|
||||
for testnet in [True, False]:
|
||||
env_name = "TESTNET" if testnet else "LIVE"
|
||||
print(f"\n🔄 Testing {env_name} environment...")
|
||||
|
||||
client = BybitRestClient(
|
||||
api_key="dummy",
|
||||
api_secret="dummy",
|
||||
testnet=testnet
|
||||
)
|
||||
|
||||
# Test 1: Server time
|
||||
try:
|
||||
server_time = client.get_server_time()
|
||||
time_second = server_time.get('result', {}).get('timeSecond')
|
||||
print(f"✅ Server time: {time_second}")
|
||||
except Exception as e:
|
||||
print(f"❌ Server time failed: {e}")
|
||||
continue
|
||||
|
||||
# Test 2: Get ticker for ETHUSDT
|
||||
try:
|
||||
ticker = client.get_ticker('ETHUSDT', 'linear')
|
||||
ticker_data = ticker.get('result', {}).get('list', [])
|
||||
if ticker_data:
|
||||
data = ticker_data[0]
|
||||
print(f"✅ ETH/USDT ticker:")
|
||||
print(f" Last Price: ${float(data.get('lastPrice', 0)):.2f}")
|
||||
print(f" 24h Volume: {float(data.get('volume24h', 0)):.2f}")
|
||||
print(f" 24h Change: {float(data.get('price24hPcnt', 0)) * 100:.2f}%")
|
||||
else:
|
||||
print("❌ No ticker data received")
|
||||
except Exception as e:
|
||||
print(f"❌ Ticker failed: {e}")
|
||||
|
||||
# Test 3: Get instruments info
|
||||
try:
|
||||
instruments = client.get_instruments_info('linear')
|
||||
instruments_list = instruments.get('result', {}).get('list', [])
|
||||
eth_instruments = [i for i in instruments_list if 'ETH' in i.get('symbol', '')]
|
||||
print(f"✅ Found {len(eth_instruments)} ETH instruments")
|
||||
for instr in eth_instruments[:3]: # Show first 3
|
||||
print(f" {instr.get('symbol')} - Status: {instr.get('status')}")
|
||||
except Exception as e:
|
||||
print(f"❌ Instruments failed: {e}")
|
||||
|
||||
# Test 4: Get orderbook
|
||||
try:
|
||||
orderbook = client.get_orderbook('ETHUSDT', 'linear', 5)
|
||||
ob_data = orderbook.get('result', {})
|
||||
bids = ob_data.get('b', [])
|
||||
asks = ob_data.get('a', [])
|
||||
|
||||
if bids and asks:
|
||||
print(f"✅ Orderbook (top 3):")
|
||||
print(f" Best bid: ${float(bids[0][0]):.2f} (qty: {float(bids[0][1]):.4f})")
|
||||
print(f" Best ask: ${float(asks[0][0]):.2f} (qty: {float(asks[0][1]):.4f})")
|
||||
spread = float(asks[0][0]) - float(bids[0][0])
|
||||
print(f" Spread: ${spread:.2f}")
|
||||
else:
|
||||
print("❌ No orderbook data received")
|
||||
except Exception as e:
|
||||
print(f"❌ Orderbook failed: {e}")
|
||||
|
||||
print(f"📊 {env_name} environment test completed")
|
||||
|
||||
def test_live_authentication():
|
||||
"""Test live authentication (if user wants to test with live credentials)"""
|
||||
print("\n" + "=" * 60)
|
||||
print("BYBIT LIVE AUTHENTICATION TEST")
|
||||
print("=" * 60)
|
||||
print("⚠️ This will test with LIVE credentials (not testnet)")
|
||||
|
||||
# Load environment variables
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
load_dotenv()
|
||||
except ImportError:
|
||||
# If dotenv is not available, try to load .env manually
|
||||
if os.path.exists('.env'):
|
||||
with open('.env', 'r') as f:
|
||||
for line in f:
|
||||
if line.strip() and not line.startswith('#'):
|
||||
key, value = line.strip().split('=', 1)
|
||||
os.environ[key] = value
|
||||
|
||||
api_key = os.getenv('BYBIT_API_KEY')
|
||||
api_secret = os.getenv('BYBIT_API_SECRET')
|
||||
|
||||
if not api_key or not api_secret:
|
||||
print("❌ No API credentials found in environment")
|
||||
return
|
||||
|
||||
print(f"🔑 Using API key: {api_key[:8]}...")
|
||||
|
||||
# Test with live environment (testnet=False)
|
||||
client = BybitRestClient(
|
||||
api_key=api_key,
|
||||
api_secret=api_secret,
|
||||
testnet=False # Use live environment
|
||||
)
|
||||
|
||||
# Test connectivity
|
||||
try:
|
||||
if client.test_connectivity():
|
||||
print("✅ Basic connectivity OK")
|
||||
else:
|
||||
print("❌ Basic connectivity failed")
|
||||
return
|
||||
except Exception as e:
|
||||
print(f"❌ Connectivity error: {e}")
|
||||
return
|
||||
|
||||
# Test authentication
|
||||
try:
|
||||
if client.test_authentication():
|
||||
print("✅ Authentication successful!")
|
||||
|
||||
# Get account info
|
||||
account_info = client.get_account_info()
|
||||
accounts = account_info.get('result', {}).get('list', [])
|
||||
|
||||
if accounts:
|
||||
print("📊 Account information:")
|
||||
for account in accounts:
|
||||
account_type = account.get('accountType', 'Unknown')
|
||||
print(f" Account Type: {account_type}")
|
||||
|
||||
coins = account.get('coin', [])
|
||||
usdt_balance = None
|
||||
for coin in coins:
|
||||
if coin.get('coin') == 'USDT':
|
||||
usdt_balance = float(coin.get('walletBalance', 0))
|
||||
break
|
||||
|
||||
if usdt_balance:
|
||||
print(f" USDT Balance: ${usdt_balance:.2f}")
|
||||
|
||||
# Show positions if any
|
||||
try:
|
||||
positions = client.get_positions('linear')
|
||||
pos_list = positions.get('result', {}).get('list', [])
|
||||
active_positions = [p for p in pos_list if float(p.get('size', 0)) != 0]
|
||||
|
||||
if active_positions:
|
||||
print(f" Active Positions: {len(active_positions)}")
|
||||
for pos in active_positions:
|
||||
symbol = pos.get('symbol')
|
||||
side = pos.get('side')
|
||||
size = float(pos.get('size', 0))
|
||||
pnl = float(pos.get('unrealisedPnl', 0))
|
||||
print(f" {symbol}: {side} {size} (PnL: ${pnl:.2f})")
|
||||
else:
|
||||
print(" No active positions")
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Could not get positions: {e}")
|
||||
|
||||
return True
|
||||
else:
|
||||
print("❌ Authentication failed")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Authentication error: {e}")
|
||||
return False
|
||||
|
||||
def main():
|
||||
"""Main function"""
|
||||
print("🚀 Starting Bybit API Tests...")
|
||||
|
||||
# Test public API
|
||||
test_public_api()
|
||||
|
||||
# Ask user if they want to test live authentication
|
||||
print("\n" + "=" * 60)
|
||||
response = input("Do you want to test live authentication? (y/N): ").lower()
|
||||
|
||||
if response == 'y' or response == 'yes':
|
||||
success = test_live_authentication()
|
||||
if success:
|
||||
print("\n✅ Live authentication test passed!")
|
||||
print("🎯 Your Bybit integration is working!")
|
||||
else:
|
||||
print("\n❌ Live authentication test failed")
|
||||
else:
|
||||
print("\n📋 Skipping live authentication test")
|
||||
|
||||
print("\n🎉 Public API tests completed successfully!")
|
||||
print("📈 Bybit integration is functional for market data")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user