101 lines
4.0 KiB
Python
101 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify CoinEx API connection and credentials.
|
|
Run this before using the live trader to ensure everything works.
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Add parent to path for imports
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from coinex_live_trader import CoinExAPI, load_coinex_credentials, setup_logging
|
|
|
|
def test_api():
|
|
"""Test CoinEx API connection and basic functionality."""
|
|
try:
|
|
print("=== CoinEx API Test ===")
|
|
|
|
# Setup logging directories
|
|
setup_logging()
|
|
print("Log directories created")
|
|
|
|
# Load credentials
|
|
print("Loading credentials...")
|
|
access_id, secret_key = load_coinex_credentials()
|
|
print(f"Access ID: {access_id[:8]}...")
|
|
|
|
# Initialize API
|
|
api = CoinExAPI(access_id, secret_key)
|
|
print("API client initialized")
|
|
|
|
# Test balance endpoint
|
|
print("\nTesting futures balance endpoint...")
|
|
balance = api.get_futures_balance()
|
|
print(f"Raw balance response: {balance}")
|
|
|
|
# Handle different response formats
|
|
if isinstance(balance, list):
|
|
if balance:
|
|
balance_data = balance[0] # Take first item if it's a list
|
|
# CoinEx uses different field names
|
|
total_balance = float(balance_data.get("available", 0)) + float(balance_data.get("frozen", 0))
|
|
available_balance = float(balance_data.get("available", 0))
|
|
else:
|
|
print("⚠️ Empty balance list returned")
|
|
total_balance = available_balance = 0
|
|
elif isinstance(balance, dict):
|
|
total_balance = float(balance.get("available", 0)) + float(balance.get("frozen", 0))
|
|
available_balance = float(balance.get("available", 0))
|
|
else:
|
|
print(f"⚠️ Unexpected balance format: {type(balance)}")
|
|
total_balance = available_balance = 0
|
|
|
|
print(f"✅ Balance retrieved successfully")
|
|
print(f" Total Balance: ${total_balance:.2f}")
|
|
print(f" Available Balance: ${available_balance:.2f}")
|
|
|
|
# Test market data endpoint
|
|
print("\nTesting market data endpoint...")
|
|
market_data = api.get_market_price("BTCUSDT")
|
|
if market_data:
|
|
print(f"✅ Market data retrieved successfully")
|
|
if isinstance(market_data, list):
|
|
print(f" Market data is a list with {len(market_data)} items")
|
|
if market_data:
|
|
print(f" First item keys: {list(market_data[0].keys()) if market_data[0] else 'Empty'}")
|
|
else:
|
|
print(f" Market data keys: {list(market_data.keys())}")
|
|
|
|
# Test positions endpoint
|
|
print("\nTesting positions endpoint...")
|
|
positions = api.get_positions()
|
|
if isinstance(positions, list):
|
|
print(f"✅ Positions retrieved successfully")
|
|
print(f" Number of open positions: {len(positions)}")
|
|
if positions:
|
|
for pos in positions[:3]: # Show first 3
|
|
market = pos.get("market", "Unknown")
|
|
side = pos.get("side", "Unknown")
|
|
amount = pos.get("amount", "0")
|
|
print(f" {market} {side} {amount}")
|
|
else:
|
|
print(f"✅ Positions endpoint working (empty/different format)")
|
|
|
|
print("\n🎉 All tests passed! API is working correctly.")
|
|
print("\nYou can now run the live trader with:")
|
|
print(" python3 coinex_live_trader.py --dry-run (for testing)")
|
|
print(" python3 coinex_live_trader.py (for live trading)")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ Test failed: {e}")
|
|
# Just print error instead of using log_error which isn't available in this context
|
|
print(f"API test failed: {e}")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
success = test_api()
|
|
sys.exit(0 if success else 1) |