""" Tests for paper trading functionality. """ import pytest from paper_strategy_instance import PaperStrategyInstance from brokers import OrderSide, OrderType, OrderStatus class TestPaperStrategyInstance: """Tests for PaperStrategyInstance.""" def test_create_instance(self): """Test creating a paper strategy instance.""" instance = PaperStrategyInstance( strategy_instance_id='test-instance-1', strategy_id='test-strategy-1', strategy_name='Test Strategy', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, initial_balance=10000.0, ) assert instance.strategy_instance_id == 'test-instance-1' assert instance.starting_balance == 10000.0 assert instance.get_current_balance() == 10000.0 assert instance.get_available_balance() == 10000.0 def test_update_prices(self): """Test updating prices in paper broker.""" instance = PaperStrategyInstance( strategy_instance_id='test-1', strategy_id='strat-1', strategy_name='Test', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, ) instance.update_prices({'BTC/USDT': 50000.0, 'ETH/USDT': 3000.0}) assert instance.get_current_price(symbol='BTC/USDT') == 50000.0 assert instance.get_current_price(symbol='ETH/USDT') == 3000.0 def test_trade_order_market_buy(self): """Test placing a market buy order.""" instance = PaperStrategyInstance( strategy_instance_id='test-1', strategy_id='strat-1', strategy_name='Test', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, initial_balance=10000.0, commission=0.001, ) # Set price instance.update_prices({'BTC/USDT': 50000.0}) # Place order result = instance.trade_order( trade_type='buy', size=0.1, order_type='MARKET', source={'symbol': 'BTC/USDT'} ) assert result.success assert result.status == OrderStatus.FILLED # Check position exists pos = instance.get_position('BTC/USDT') assert pos is not None assert pos.size == 0.1 # Check balance reduced assert instance.get_available_balance() < 10000.0 def test_trade_order_sell(self): """Test selling a position.""" instance = PaperStrategyInstance( strategy_instance_id='test-1', strategy_id='strat-1', strategy_name='Test', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, initial_balance=10000.0, commission=0, slippage=0, ) # Buy first instance.update_prices({'BTC/USDT': 50000.0}) instance.trade_order( trade_type='buy', size=0.1, order_type='MARKET', source={'symbol': 'BTC/USDT'} ) # Now sell at higher price instance.update_prices({'BTC/USDT': 51000.0}) result = instance.trade_order( trade_type='sell', size=0.1, order_type='MARKET', source={'symbol': 'BTC/USDT'} ) assert result.success # Position should be closed pos = instance.get_position('BTC/USDT') assert pos is None def test_close_position(self): """Test closing a position via close_position method.""" instance = PaperStrategyInstance( strategy_instance_id='test-1', strategy_id='strat-1', strategy_name='Test', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, ) instance.update_prices({'BTC/USDT': 50000.0}) instance.trade_order( trade_type='buy', size=0.1, order_type='MARKET', source={'symbol': 'BTC/USDT'} ) result = instance.close_position('BTC/USDT') assert result.success def test_close_all_positions(self): """Test closing all positions.""" instance = PaperStrategyInstance( strategy_instance_id='test-1', strategy_id='strat-1', strategy_name='Test', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, ) instance.update_prices({'BTC/USDT': 50000.0, 'ETH/USDT': 3000.0}) instance.trade_order( trade_type='buy', size=0.1, order_type='MARKET', source={'symbol': 'BTC/USDT'} ) instance.trade_order( trade_type='buy', size=1.0, order_type='MARKET', source={'symbol': 'ETH/USDT'} ) assert instance.get_active_trades() == 2 results = instance.close_all_positions() assert len(results) == 2 assert all(r.success for r in results) assert instance.get_active_trades() == 0 def test_reset(self): """Test resetting paper trading state.""" instance = PaperStrategyInstance( strategy_instance_id='test-1', strategy_id='strat-1', strategy_name='Test', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, initial_balance=10000.0, ) instance.update_prices({'BTC/USDT': 50000.0}) instance.trade_order( trade_type='buy', size=0.1, order_type='MARKET', source={'symbol': 'BTC/USDT'} ) assert instance.get_available_balance() < 10000.0 instance.reset() assert instance.get_available_balance() == 10000.0 assert instance.get_active_trades() == 0 def test_get_trade_history(self): """Test getting trade history.""" instance = PaperStrategyInstance( strategy_instance_id='test-1', strategy_id='strat-1', strategy_name='Test', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, ) instance.update_prices({'BTC/USDT': 50000.0}) instance.trade_order( trade_type='buy', size=0.1, order_type='MARKET', source={'symbol': 'BTC/USDT'} ) history = instance.get_trade_history() assert len(history) == 1 assert history[0]['symbol'] == 'BTC/USDT' assert history[0]['side'] == 'buy' def test_failed_margin_open_does_not_credit_spot_cash(self): """Insufficient-collateral failures must not mint spot cash.""" instance = PaperStrategyInstance( strategy_instance_id='test-margin-open-fail', strategy_id='strat-margin-open-fail', strategy_name='Test Margin Fail', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, initial_balance=100.0, ) instance.update_prices({'BTC/USDT': 50000.0}) instance.exec_context['current_symbol'] = 'BTC/USDT' spot_before = instance.paper_broker.get_balance() margin_before = instance.paper_margin_broker.get_balance() with pytest.raises(ValueError, match='Insufficient balance'): instance.open_margin_position(side='long', collateral=150.0, leverage=3.0) assert instance.paper_broker.get_balance() == pytest.approx(spot_before) assert instance.paper_margin_broker.get_balance() == pytest.approx(margin_before) def test_margin_limit_open_uses_target_market_and_name_order(self): """Limit-style margin opens should respect shared target-market and order-name options.""" instance = PaperStrategyInstance( strategy_instance_id='test-margin-limit-open', strategy_id='strat-margin-limit-open', strategy_name='Test Margin Limit', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, initial_balance=10000.0, ) instance.exec_context['current_symbol'] = 'BTC/USDT' instance.update_prices({'BTC/USDT': 50000.0, 'ETH/USDT': 3000.0}) instance.paper_margin_broker.update_price('BTC/USDT', 50000.0) instance.paper_margin_broker.update_price('ETH/USDT', 3000.0, 'binance') result = instance.open_margin_position( side='long', collateral=200.0, leverage=3.0, limit={'limit': 3050.0}, tif='IOC', stop_loss={'value': 2900.0}, take_profit={'value': 3400.0}, target_market={'exchange': 'binance', 'symbol': 'ETH/USDT', 'time_frame': '15m'}, name_order={'order_name': 'ETH breakout'}, ) position = instance.paper_margin_broker.get_position('ETH/USDT') assert result['success'] is True assert result['symbol'] == 'ETH/USDT' assert result['order_name'] == 'ETH breakout' assert position is not None assert instance.paper_margin_broker.get_position('BTC/USDT') is None assert instance.paper_margin_broker._position_sltp['ETH/USDT'] == { 'stop_loss': 2900.0, 'take_profit': 3400.0, } def test_margin_trailing_limit_entry_fills_on_later_tick(self): """Trailing-limit margin entries should persist, arm, and fill on a later tick.""" instance = PaperStrategyInstance( strategy_instance_id='test-margin-trailing-limit', strategy_id='strat-margin-trailing-limit', strategy_name='Test Margin Trail Limit', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, initial_balance=10000.0, ) instance.exec_context['current_symbol'] = 'BTC/USDT' instance.update_prices({'BTC/USDT': 50000.0}) instance.paper_margin_broker.update_price('BTC/USDT', 50000.0) result = instance.open_margin_position( side='long', collateral=200.0, leverage=3.0, tif='GTC', trailing_limit={'trail_limit_distance': 100.0}, target_market={'symbol': 'BTC/USDT'}, name_order={'order_name': 'Trail Entry'}, ) assert result['success'] is True assert result['pending'] is True assert result['pending_type'] == 'trailing_limit' assert instance.variables['_pending_margin_entries']['BTC/USDT']['order_name'] == 'Trail Entry' first_tick_events = instance.tick({'symbol': 'BTC/USDT', 'close': 49800.0}) second_tick_events = instance.tick({'symbol': 'BTC/USDT', 'close': 49900.0}) assert not any(event.get('type') == 'margin_entry_filled' for event in first_tick_events) assert any( event.get('type') == 'margin_entry_filled' and event.get('order_name') == 'Trail Entry' for event in second_tick_events ) assert instance.paper_margin_broker.get_position('BTC/USDT') is not None assert 'BTC/USDT' not in instance.variables['_pending_margin_entries'] def test_margin_trailing_stop_closes_position_after_reversal(self): """Trailing-stop margin exits should trigger from shared trade-option payloads.""" instance = PaperStrategyInstance( strategy_instance_id='test-margin-trailing-stop', strategy_id='strat-margin-trailing-stop', strategy_name='Test Margin Trail Stop', user_id=1, generated_code='def next(self): pass', data_cache=None, indicators=None, trades=None, initial_balance=10000.0, ) instance.exec_context['current_symbol'] = 'BTC/USDT' instance.update_prices({'BTC/USDT': 50000.0}) instance.paper_margin_broker.update_price('BTC/USDT', 50000.0) result = instance.open_margin_position( side='long', collateral=200.0, leverage=3.0, trailing_stop={'trail_distance': 100.0}, name_order={'order_name': 'Trail Exit'}, ) assert result['success'] is True assert 'BTC/USDT' in instance.variables['_margin_trailing_stops'] instance.tick({'symbol': 'BTC/USDT', 'close': 50500.0}) trigger_events = instance.tick({'symbol': 'BTC/USDT', 'close': 50400.0}) assert any( event.get('type') == 'margin_trailing_stop_triggered' and event.get('order_name') == 'Trail Exit' for event in trigger_events ) assert instance.paper_margin_broker.get_position('BTC/USDT') is None assert 'BTC/USDT' not in instance.variables['_margin_trailing_stops'] class TestStrategiesModeSeletion: """Tests for strategy mode selection.""" def test_mode_selection_imports(self): """Test that mode selection imports work.""" from Strategies import Strategies from brokers import TradingMode assert TradingMode.PAPER == 'paper' assert TradingMode.BACKTEST == 'backtest' assert TradingMode.LIVE == 'live'