221 lines
8.0 KiB
Python
221 lines
8.0 KiB
Python
#!/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 core.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()
|