""" Integration tests for live trading against Binance testnet. These tests require real testnet API keys set in environment variables: BRIGHTER_BINANCE_TESTNET_API_KEY BRIGHTER_BINANCE_TESTNET_API_SECRET Run with: pytest tests/test_live_integration.py -m live_testnet -v Skip these tests in CI by using: pytest -m "not live_testnet" """ import os import sys import time import pytest from unittest.mock import MagicMock, patch from decimal import Decimal # Add src to path sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) # Check if testnet credentials are available TESTNET_API_KEY = os.environ.get('BRIGHTER_BINANCE_TESTNET_API_KEY', '') TESTNET_API_SECRET = os.environ.get('BRIGHTER_BINANCE_TESTNET_API_SECRET', '') HAS_TESTNET_CREDENTIALS = bool(TESTNET_API_KEY and TESTNET_API_SECRET) # Skip reason for tests requiring credentials SKIP_REASON = "Testnet API keys not configured. Set BRIGHTER_BINANCE_TESTNET_API_KEY and BRIGHTER_BINANCE_TESTNET_API_SECRET." @pytest.fixture def testnet_exchange(): """Create a real Exchange instance connected to Binance testnet.""" if not HAS_TESTNET_CREDENTIALS: pytest.skip(SKIP_REASON) from Exchange import Exchange api_keys = { 'key': TESTNET_API_KEY, 'secret': TESTNET_API_SECRET } exchange = Exchange( name='binance', api_keys=api_keys, exchange_id='binance', testnet=True ) yield exchange @pytest.fixture def testnet_live_broker(testnet_exchange): """Create a LiveBroker connected to Binance testnet.""" from brokers import LiveBroker from DataCache_v3 import DataCache # Create a real data cache for persistence testing data_cache = DataCache() broker = LiveBroker( exchange=testnet_exchange, testnet=True, initial_balance=0.0, commission=0.001, slippage=0.0, data_cache=data_cache, rate_limit=2.0 ) yield broker # Cleanup: disconnect if broker._connected: broker.disconnect() @pytest.fixture def connected_broker(testnet_live_broker): """Provide a connected LiveBroker.""" testnet_live_broker.connect() return testnet_live_broker # ============================================================================= # Exchange Connection Tests # ============================================================================= @pytest.mark.live_testnet class TestExchangeConnection: """Tests for exchange connectivity.""" def test_exchange_connects_to_testnet(self, testnet_exchange): """Verify exchange initializes with sandbox mode enabled.""" assert testnet_exchange is not None assert testnet_exchange.testnet is True assert testnet_exchange.client is not None # Verify testnet URLs are being used assert 'testnet' in testnet_exchange.client.urls.get('api', {}).get('public', '') def test_exchange_is_configured(self, testnet_exchange): """Verify exchange has valid API credentials.""" assert testnet_exchange.configured is True assert testnet_exchange.api_key == TESTNET_API_KEY def test_exchange_can_load_markets(self, testnet_exchange): """Verify exchange can load market data.""" # Markets should be loaded during initialization assert testnet_exchange.exchange_info is not None assert len(testnet_exchange.exchange_info) > 0 # BTC/USDT should be available assert 'BTC/USDT' in testnet_exchange.exchange_info # ============================================================================= # Balance and Price Tests # ============================================================================= @pytest.mark.live_testnet class TestBalanceAndPrice: """Tests for balance and price fetching.""" def test_balance_sync_returns_real_data(self, connected_broker): """Verify sync_balance() returns dict with actual testnet balances.""" balances = connected_broker.sync_balance() assert isinstance(balances, dict) # Testnet accounts typically have some balance # At minimum, we should get a response without errors # Check that balance tracking is working assert connected_broker._balances is not None def test_get_balance_returns_quote_currency(self, connected_broker): """Verify get_balance() returns a numeric value.""" balance = connected_broker.get_balance() assert isinstance(balance, (int, float)) assert balance >= 0 def test_get_available_balance(self, connected_broker): """Verify get_available_balance() works.""" available = connected_broker.get_available_balance() assert isinstance(available, (int, float)) assert available >= 0 def test_price_fetch_returns_valid_price(self, connected_broker): """Verify get_current_price() returns positive float.""" price = connected_broker.get_current_price('BTC/USDT') assert isinstance(price, float) assert price > 0 # BTC price should be in reasonable range (testnet may have different prices) # Just verify it's a sensible number assert price > 100 # BTC should be > $100 assert price < 1000000 # and < $1M def test_price_cache_expires(self, connected_broker): """Verify price cache expires after TTL.""" symbol = 'BTC/USDT' # First fetch price1 = connected_broker.get_current_price(symbol) # Immediate second fetch should use cache price2 = connected_broker.get_current_price(symbol) assert price1 == price2 # Wait for cache to expire (5 seconds + buffer) time.sleep(6) # Third fetch should get fresh price price3 = connected_broker.get_current_price(symbol) # Price may or may not have changed, but fetch should succeed assert isinstance(price3, float) assert price3 > 0 def test_total_equity_calculation(self, connected_broker): """Verify get_total_equity() works.""" equity = connected_broker.get_total_equity() assert isinstance(equity, float) assert equity >= 0 # ============================================================================= # Order Lifecycle Tests # ============================================================================= @pytest.mark.live_testnet class TestOrderLifecycle: """Tests for order placement, monitoring, and cancellation.""" def test_place_limit_order_appears_on_exchange(self, connected_broker): """Place limit order and verify it appears in open orders.""" from brokers import OrderSide, OrderType # Get current price and place limit order below market current_price = connected_broker.get_current_price('BTC/USDT') limit_price = current_price * 0.9 # 10% below market # Place small limit buy order result = connected_broker.place_order( symbol='BTC/USDT', side=OrderSide.BUY, order_type=OrderType.LIMIT, size=0.001, # Minimum size price=limit_price, time_in_force='GTC' ) assert result.success is True assert result.order_id is not None order_id = result.order_id try: # Verify order appears in open orders open_orders = connected_broker.get_open_orders() order_ids = [o['order_id'] for o in open_orders] assert order_id in order_ids # Verify order details from the list order = next((o for o in open_orders if o['order_id'] == order_id), None) assert order is not None assert order['symbol'] == 'BTC/USDT' assert order['side'] == 'buy' finally: # Cleanup: cancel the order connected_broker.cancel_order(order_id) def test_cancel_order_removes_from_exchange(self, connected_broker): """Cancel order and verify it's removed from open orders.""" from brokers import OrderSide, OrderType # Place a limit order current_price = connected_broker.get_current_price('BTC/USDT') limit_price = current_price * 0.9 result = connected_broker.place_order( symbol='BTC/USDT', side=OrderSide.BUY, order_type=OrderType.LIMIT, size=0.001, price=limit_price, time_in_force='GTC' ) assert result.success is True order_id = result.order_id # Cancel the order cancel_result = connected_broker.cancel_order(order_id) assert cancel_result is True # Small delay for exchange to process time.sleep(1) # Sync open orders from exchange connected_broker.sync_open_orders() # Verify order is no longer in open orders open_orders = connected_broker.get_open_orders() order_ids = [o['order_id'] for o in open_orders] assert order_id not in order_ids def test_market_order_fills_immediately(self, connected_broker): """Place small market order and verify it fills.""" from brokers import OrderSide, OrderType, OrderStatus # Check we have balance balance = connected_broker.get_balance() if balance < 20: # Need at least $20 for minimum BTC order pytest.skip("Insufficient testnet balance for market order test") # Place small market buy (no time_in_force for market orders) result = connected_broker.place_order( symbol='BTC/USDT', side=OrderSide.BUY, order_type=OrderType.MARKET, size=0.001 ) assert result.success is True # Market orders should fill immediately assert result.status == OrderStatus.FILLED assert result.filled_qty > 0 assert result.filled_price > 0 def test_order_fill_detected_in_update_cycle(self, connected_broker): """Place market order and verify fill event in update().""" from brokers import OrderSide, OrderType balance = connected_broker.get_balance() if balance < 20: pytest.skip("Insufficient testnet balance") # Place market order (no time_in_force for market orders) result = connected_broker.place_order( symbol='BTC/USDT', side=OrderSide.BUY, order_type=OrderType.MARKET, size=0.001 ) assert result.success is True # Call update to process any events events = connected_broker.update() # For immediate fills, the fill may already be recorded # The important thing is update() doesn't error assert isinstance(events, list) # ============================================================================= # Persistence Tests # ============================================================================= @pytest.mark.live_testnet class TestPersistence: """Tests for state persistence and recovery.""" def test_state_persistence_survives_restart(self, testnet_exchange): """Save state, create new broker, load state, verify orders match.""" from brokers import LiveBroker, OrderSide, OrderType from DataCache_v3 import DataCache data_cache = DataCache() strategy_id = 'test-persistence-001' # Create first broker and place order broker1 = LiveBroker( exchange=testnet_exchange, testnet=True, data_cache=data_cache, rate_limit=2.0 ) broker1.connect() # Place a limit order current_price = broker1.get_current_price('BTC/USDT') limit_price = current_price * 0.85 # Well below market result = broker1.place_order( symbol='BTC/USDT', side=OrderSide.BUY, order_type=OrderType.LIMIT, size=0.001, price=limit_price, time_in_force='GTC' ) assert result.success is True order_id = result.order_id try: # Save state broker1.save_state(strategy_id) # Get state for comparison original_state = broker1.to_state_dict() original_order_count = len(broker1._orders) # Disconnect first broker broker1.disconnect() # Create new broker instance broker2 = LiveBroker( exchange=testnet_exchange, testnet=True, data_cache=data_cache, rate_limit=2.0 ) broker2.connect() # Load state load_success = broker2.load_state(strategy_id) assert load_success is True # Verify orders were restored assert len(broker2._orders) == original_order_count assert order_id in broker2._orders restored_order = broker2._orders[order_id] assert restored_order.symbol == 'BTC/USDT' finally: # Cleanup: cancel order using either broker try: if broker1._connected: broker1.cancel_order(order_id) elif broker2._connected: broker2.cancel_order(order_id) except Exception: pass def test_reconcile_detects_external_changes(self, connected_broker): """Place order via CCXT directly, reconcile, verify order appears.""" from brokers import OrderSide, OrderType # Get current price current_price = connected_broker.get_current_price('BTC/USDT') limit_price = current_price * 0.8 # Well below market # Place order directly via CCXT (simulating external action) exchange = connected_broker._exchange result, order_data = exchange.place_order( symbol='BTC/USDT', side='buy', type='limit', timeInForce='GTC', quantity=0.001, price=limit_price ) assert result == 'Success' external_order_id = order_data['id'] try: # Before reconciliation, broker doesn't know about this order # (unless it was already synced) # Reconcile with exchange reconcile_result = connected_broker.reconcile_with_exchange() assert reconcile_result['success'] is True # After reconciliation, the external order should be tracked # Sync open orders to update local state connected_broker.sync_open_orders() # Check if external order is now in our tracking exchange_ids = [o.exchange_order_id for o in connected_broker._orders.values()] assert external_order_id in exchange_ids finally: # Cleanup: cancel order try: exchange.client.cancel_order(external_order_id, 'BTC/USDT') except Exception: pass # ============================================================================= # Full Trade Lifecycle Test # ============================================================================= @pytest.mark.live_testnet class TestFullTradeLifecycle: """End-to-end trade lifecycle tests.""" def test_full_trade_lifecycle(self, connected_broker): """Open position, hold, close position, verify P&L calculated.""" from brokers import OrderSide, OrderType, OrderStatus balance = connected_broker.get_balance() if balance < 50: pytest.skip("Insufficient testnet balance for full lifecycle test") symbol = 'BTC/USDT' size = 0.001 # Record starting balance starting_balance = connected_broker.get_balance() # Step 1: Open position (buy) buy_result = connected_broker.place_order( symbol=symbol, side=OrderSide.BUY, order_type=OrderType.MARKET, size=size ) assert buy_result.success is True assert buy_result.status == OrderStatus.FILLED entry_price = buy_result.filled_price assert entry_price > 0 # Small delay time.sleep(2) # Step 2: Check position exists connected_broker.sync_balance() # Step 3: Close position (sell) sell_result = connected_broker.place_order( symbol=symbol, side=OrderSide.SELL, order_type=OrderType.MARKET, size=size ) assert sell_result.success is True assert sell_result.status == OrderStatus.FILLED exit_price = sell_result.filled_price assert exit_price > 0 # Step 4: Verify P&L # P&L = (exit - entry) * size - commissions gross_pnl = (exit_price - entry_price) * size # We can't verify exact P&L due to commissions, but the trade completed print(f"Entry: {entry_price}, Exit: {exit_price}, Gross P&L: {gross_pnl}") # Sync final balance connected_broker.sync_balance() final_balance = connected_broker.get_balance() # Balance should have changed by approximately the P&L balance_change = final_balance - starting_balance print(f"Balance change: {balance_change} (includes commissions)") # ============================================================================= # Rate Limiting Tests # ============================================================================= @pytest.mark.live_testnet class TestRateLimiting: """Tests for API rate limiting.""" def test_rapid_requests_dont_cause_errors(self, connected_broker): """Verify rapid API calls are properly throttled.""" # Make multiple rapid price requests errors = [] for i in range(10): try: price = connected_broker.get_current_price('BTC/USDT') assert price > 0 except Exception as e: errors.append(str(e)) # Should have no rate limit errors rate_limit_errors = [e for e in errors if 'rate' in e.lower() or 'limit' in e.lower()] assert len(rate_limit_errors) == 0, f"Rate limit errors: {rate_limit_errors}" # ============================================================================= # Error Handling Tests # ============================================================================= @pytest.mark.live_testnet class TestErrorHandling: """Tests for error handling and edge cases.""" def test_invalid_symbol_handled_gracefully(self, connected_broker): """Verify invalid symbol doesn't crash.""" price = connected_broker.get_current_price('INVALID/PAIR') # Should return 0 or cached value, not crash assert isinstance(price, (int, float)) def test_insufficient_balance_rejected(self, connected_broker): """Verify order with insufficient balance is rejected.""" from brokers import OrderSide, OrderType # Try to buy way more than we have result = connected_broker.place_order( symbol='BTC/USDT', side=OrderSide.BUY, order_type=OrderType.MARKET, size=1000.0 # 1000 BTC - definitely more than testnet balance ) # Should fail gracefully (exchange may return various error messages) assert result.success is False # Binance returns different error codes - just verify it failed with a message assert result.message is not None and len(result.message) > 0 def test_cancel_nonexistent_order(self, connected_broker): """Verify canceling nonexistent order is handled.""" result = connected_broker.cancel_order('nonexistent-order-id') # Should fail gracefully (returns False for nonexistent orders) assert result is False # ============================================================================= # LiveStrategyInstance Integration Tests # ============================================================================= @pytest.mark.live_testnet @pytest.mark.skip(reason="LiveStrategyInstance integration tests require full DataCache setup - tested separately in test_live_strategy_instance.py") class TestLiveStrategyInstanceIntegration: """Integration tests for LiveStrategyInstance with real exchange.""" def test_live_strategy_instance_creation(self, testnet_exchange): """Verify LiveStrategyInstance can be created with real exchange.""" from live_strategy_instance import LiveStrategyInstance from unittest.mock import MagicMock # Use mock data_cache to avoid database initialization hang data_cache = MagicMock() instance = LiveStrategyInstance( strategy_instance_id='test-live-001', strategy_id='test-strategy', strategy_name='Test Strategy', user_id=1, generated_code='pass', data_cache=data_cache, indicators=None, trades=None, exchange=testnet_exchange, testnet=True, initial_balance=0.0, max_position_pct=0.5, circuit_breaker_pct=-0.10 ) assert instance is not None assert instance.is_testnet is True assert instance.live_broker._connected is True # Cleanup instance.disconnect() def test_live_strategy_instance_tick(self, testnet_exchange): """Verify tick() works with real exchange data.""" from live_strategy_instance import LiveStrategyInstance from unittest.mock import MagicMock # Use mock data_cache to avoid database initialization hang data_cache = MagicMock() instance = LiveStrategyInstance( strategy_instance_id='test-live-002', strategy_id='test-strategy', strategy_name='Test Strategy', user_id=1, generated_code='pass', data_cache=data_cache, indicators=None, trades=None, exchange=testnet_exchange, testnet=True ) try: # Get current price for candle data price = instance.get_current_price(symbol='BTC/USDT') candle_data = { 'symbol': 'BTC/USDT', 'open': price, 'high': price * 1.01, 'low': price * 0.99, 'close': price, 'volume': 100.0 } # Run tick events = instance.tick(candle_data) assert isinstance(events, list) # Should complete without circuit breaker (we haven't lost money) assert not any(e.get('type') == 'circuit_breaker' for e in events) finally: instance.disconnect()