349 lines
12 KiB
Python
349 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for Bybit ETH futures position opening/closing
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
# Add the project root to the path
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
# Load environment variables from .env file
|
|
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
|
|
|
|
from core.exchanges.bybit_interface import BybitInterface
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class BybitEthFuturesTest:
|
|
"""Test class for Bybit ETH futures trading"""
|
|
|
|
def __init__(self, test_mode=True):
|
|
self.test_mode = test_mode
|
|
self.bybit = BybitInterface(test_mode=test_mode)
|
|
self.test_symbol = 'ETHUSDT'
|
|
self.test_quantity = 0.01 # Small test amount
|
|
|
|
def run_tests(self):
|
|
"""Run all tests"""
|
|
print("=" * 60)
|
|
print("BYBIT ETH FUTURES POSITION TESTING")
|
|
print("=" * 60)
|
|
print(f"Test mode: {'TESTNET' if self.test_mode else 'LIVE'}")
|
|
print(f"Symbol: {self.test_symbol}")
|
|
print(f"Test quantity: {self.test_quantity} ETH")
|
|
print("=" * 60)
|
|
|
|
# Test 1: Connection
|
|
if not self.test_connection():
|
|
print("❌ Connection failed - stopping tests")
|
|
return False
|
|
|
|
# Test 2: Check balance
|
|
if not self.test_balance():
|
|
print("❌ Balance check failed - stopping tests")
|
|
return False
|
|
|
|
# Test 3: Check current positions
|
|
self.test_current_positions()
|
|
|
|
# Test 4: Get ticker
|
|
if not self.test_ticker():
|
|
print("❌ Ticker test failed - stopping tests")
|
|
return False
|
|
|
|
# Test 5: Open a long position
|
|
long_order = self.test_open_long_position()
|
|
if not long_order:
|
|
print("❌ Open long position failed")
|
|
return False
|
|
|
|
# Test 6: Check position after opening
|
|
time.sleep(2) # Wait for position to be reflected
|
|
if not self.test_position_after_open():
|
|
print("❌ Position check after opening failed")
|
|
return False
|
|
|
|
# Test 7: Close the position
|
|
if not self.test_close_position():
|
|
print("❌ Close position failed")
|
|
return False
|
|
|
|
# Test 8: Check position after closing
|
|
time.sleep(2) # Wait for position to be reflected
|
|
self.test_position_after_close()
|
|
|
|
print("\n" + "=" * 60)
|
|
print("✅ ALL TESTS COMPLETED SUCCESSFULLY")
|
|
print("=" * 60)
|
|
return True
|
|
|
|
def test_connection(self):
|
|
"""Test connection to Bybit"""
|
|
print("\n📡 Testing connection to Bybit...")
|
|
|
|
# First test simple connectivity without auth
|
|
print("Testing basic API connectivity...")
|
|
try:
|
|
from core.exchanges.bybit_rest_client import BybitRestClient
|
|
client = BybitRestClient(
|
|
api_key="dummy",
|
|
api_secret="dummy",
|
|
testnet=True
|
|
)
|
|
|
|
# Test public endpoint (server time)
|
|
server_time = client.get_server_time()
|
|
print(f"✅ Public API working - Server time: {server_time.get('result', {}).get('timeSecond')}")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Public API failed: {e}")
|
|
return False
|
|
|
|
# Now test with actual credentials
|
|
print("Testing with API credentials...")
|
|
try:
|
|
connected = self.bybit.connect()
|
|
if connected:
|
|
print("✅ Successfully connected to Bybit with credentials")
|
|
return True
|
|
else:
|
|
print("❌ Failed to connect to Bybit with credentials")
|
|
print("This might be due to:")
|
|
print("- Invalid API credentials")
|
|
print("- Credentials not enabled for testnet")
|
|
print("- Missing required permissions")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ Connection error: {e}")
|
|
return False
|
|
|
|
def test_balance(self):
|
|
"""Test getting account balance"""
|
|
print("\n💰 Testing account balance...")
|
|
|
|
try:
|
|
# Get USDT balance (for margin)
|
|
usdt_balance = self.bybit.get_balance('USDT')
|
|
print(f"USDT Balance: {usdt_balance}")
|
|
|
|
# Get all balances
|
|
all_balances = self.bybit.get_all_balances()
|
|
print("All balances:")
|
|
for asset, balance in all_balances.items():
|
|
if balance['total'] > 0:
|
|
print(f" {asset}: Free={balance['free']}, Locked={balance['locked']}, Total={balance['total']}")
|
|
|
|
if usdt_balance > 10: # Need at least $10 for testing
|
|
print("✅ Sufficient balance for testing")
|
|
return True
|
|
else:
|
|
print("❌ Insufficient USDT balance for testing (need at least $10)")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"❌ Balance check error: {e}")
|
|
return False
|
|
|
|
def test_current_positions(self):
|
|
"""Test getting current positions"""
|
|
print("\n📊 Checking current positions...")
|
|
|
|
try:
|
|
positions = self.bybit.get_positions()
|
|
if positions:
|
|
print(f"Found {len(positions)} open positions:")
|
|
for pos in positions:
|
|
print(f" {pos['symbol']}: {pos['side']} {pos['size']} @ ${pos['entry_price']:.2f}")
|
|
print(f" PnL: ${pos['unrealized_pnl']:.2f} ({pos['percentage']:.2f}%)")
|
|
else:
|
|
print("No open positions found")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Position check error: {e}")
|
|
|
|
def test_ticker(self):
|
|
"""Test getting ticker information"""
|
|
print(f"\n📈 Testing ticker for {self.test_symbol}...")
|
|
|
|
try:
|
|
ticker = self.bybit.get_ticker(self.test_symbol)
|
|
if ticker:
|
|
print(f"✅ Ticker data received:")
|
|
print(f" Last Price: ${ticker['last_price']:.2f}")
|
|
print(f" Bid: ${ticker['bid_price']:.2f}")
|
|
print(f" Ask: ${ticker['ask_price']:.2f}")
|
|
print(f" 24h Volume: {ticker['volume_24h']:.2f}")
|
|
print(f" 24h Change: {ticker['change_24h']:.4f}%")
|
|
return True
|
|
else:
|
|
print("❌ Failed to get ticker data")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"❌ Ticker error: {e}")
|
|
return False
|
|
|
|
def test_open_long_position(self):
|
|
"""Test opening a long position"""
|
|
print(f"\n🚀 Opening long position for {self.test_quantity} {self.test_symbol}...")
|
|
|
|
try:
|
|
# Place market buy order
|
|
order = self.bybit.place_order(
|
|
symbol=self.test_symbol,
|
|
side='buy',
|
|
order_type='market',
|
|
quantity=self.test_quantity
|
|
)
|
|
|
|
if 'error' in order:
|
|
print(f"❌ Order failed: {order['error']}")
|
|
return None
|
|
|
|
print("✅ Long position opened successfully:")
|
|
print(f" Order ID: {order['order_id']}")
|
|
print(f" Symbol: {order['symbol']}")
|
|
print(f" Side: {order['side']}")
|
|
print(f" Quantity: {order['quantity']}")
|
|
print(f" Status: {order['status']}")
|
|
|
|
return order
|
|
|
|
except Exception as e:
|
|
print(f"❌ Open position error: {e}")
|
|
return None
|
|
|
|
def test_position_after_open(self):
|
|
"""Test checking position after opening"""
|
|
print(f"\n📊 Checking position after opening...")
|
|
|
|
try:
|
|
positions = self.bybit.get_positions(self.test_symbol)
|
|
if positions:
|
|
position = positions[0]
|
|
print("✅ Position found:")
|
|
print(f" Symbol: {position['symbol']}")
|
|
print(f" Side: {position['side']}")
|
|
print(f" Size: {position['size']}")
|
|
print(f" Entry Price: ${position['entry_price']:.2f}")
|
|
print(f" Mark Price: ${position['mark_price']:.2f}")
|
|
print(f" Unrealized PnL: ${position['unrealized_pnl']:.2f}")
|
|
print(f" Percentage: {position['percentage']:.2f}%")
|
|
print(f" Leverage: {position['leverage']}x")
|
|
return True
|
|
else:
|
|
print("❌ No position found after opening")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"❌ Position check error: {e}")
|
|
return False
|
|
|
|
def test_close_position(self):
|
|
"""Test closing the position"""
|
|
print(f"\n🔄 Closing position for {self.test_symbol}...")
|
|
|
|
try:
|
|
# Close the position
|
|
close_order = self.bybit.close_position(self.test_symbol)
|
|
|
|
if 'error' in close_order:
|
|
print(f"❌ Close order failed: {close_order['error']}")
|
|
return False
|
|
|
|
print("✅ Position closed successfully:")
|
|
print(f" Order ID: {close_order['order_id']}")
|
|
print(f" Symbol: {close_order['symbol']}")
|
|
print(f" Side: {close_order['side']}")
|
|
print(f" Quantity: {close_order['quantity']}")
|
|
print(f" Status: {close_order['status']}")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Close position error: {e}")
|
|
return False
|
|
|
|
def test_position_after_close(self):
|
|
"""Test checking position after closing"""
|
|
print(f"\n📊 Checking position after closing...")
|
|
|
|
try:
|
|
positions = self.bybit.get_positions(self.test_symbol)
|
|
if positions:
|
|
position = positions[0]
|
|
print("⚠️ Position still exists (may be partially closed):")
|
|
print(f" Symbol: {position['symbol']}")
|
|
print(f" Side: {position['side']}")
|
|
print(f" Size: {position['size']}")
|
|
print(f" Entry Price: ${position['entry_price']:.2f}")
|
|
print(f" Unrealized PnL: ${position['unrealized_pnl']:.2f}")
|
|
else:
|
|
print("✅ Position successfully closed - no open positions")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Position check error: {e}")
|
|
|
|
def test_order_history(self):
|
|
"""Test getting order history"""
|
|
print(f"\n📋 Checking recent orders...")
|
|
|
|
try:
|
|
# Get open orders
|
|
open_orders = self.bybit.get_open_orders(self.test_symbol)
|
|
print(f"Open orders: {len(open_orders)}")
|
|
for order in open_orders:
|
|
print(f" {order['order_id']}: {order['side']} {order['quantity']} @ ${order['price']:.2f} - {order['status']}")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Order history error: {e}")
|
|
|
|
def main():
|
|
"""Main function"""
|
|
print("Starting Bybit ETH Futures Test...")
|
|
|
|
# Check if API credentials are set
|
|
api_key = os.getenv('BYBIT_API_KEY')
|
|
api_secret = os.getenv('BYBIT_API_SECRET')
|
|
|
|
if not api_key or not api_secret:
|
|
print("❌ Please set BYBIT_API_KEY and BYBIT_API_SECRET environment variables")
|
|
return False
|
|
|
|
# Create test instance
|
|
test = BybitEthFuturesTest(test_mode=True) # Always use testnet for safety
|
|
|
|
# Run tests
|
|
success = test.run_tests()
|
|
|
|
if success:
|
|
print("\n🎉 All tests passed!")
|
|
else:
|
|
print("\n💥 Some tests failed!")
|
|
|
|
return success
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
sys.exit(0 if success else 1)
|