""" Tests for margin trading support in backtest mode. """ import pytest from datetime import datetime, timezone, timedelta from unittest.mock import MagicMock, patch import sys import os # Add src to path sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) from brokers.paper_margin_broker import PaperMarginBroker class TestPaperMarginBrokerTimeInjection: """Test that PaperMarginBroker accepts and uses time_provider.""" def test_time_provider_injection(self): """PaperMarginBroker should use injected time provider.""" simulated_time = datetime(2023, 6, 15, 12, 0, 0, tzinfo=timezone.utc) broker = PaperMarginBroker( time_provider=lambda: simulated_time, initial_balance=10000.0 ) # Verify the injected time is used assert broker._get_current_time() == simulated_time def test_default_time_provider_uses_wall_clock(self): """Without time_provider, should use wall-clock time.""" broker = PaperMarginBroker(initial_balance=10000.0) now = datetime.now(timezone.utc) broker_time = broker._get_current_time() # Should be within a few seconds of now delta = abs((broker_time - now).total_seconds()) assert delta < 5.0 def test_interest_accrues_with_simulated_time(self): """Interest should accrue based on simulated time progression.""" times = [ datetime(2023, 6, 15, 12, 0, 0, tzinfo=timezone.utc), datetime(2023, 6, 15, 13, 0, 0, tzinfo=timezone.utc), # +1 hour ] time_index = [0] def time_provider(): return times[time_index[0]] broker = PaperMarginBroker( time_provider=time_provider, initial_balance=10000.0 ) # Open position at t=0 broker.update_price('BTC/USDT', 50000.0) position = broker.open_position( symbol='BTC/USDT', side='long', collateral=1000.0, leverage=3.0, current_price=50000.0 ) initial_interest = position.interest_accrued # Advance simulated time by 1 hour time_index[0] = 1 broker.update() # Interest should have accrued position = broker.get_position('BTC/USDT') assert position.interest_accrued > initial_interest def test_set_time_provider_method(self): """Can change time provider after initialization.""" broker = PaperMarginBroker(initial_balance=10000.0) new_time = datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc) broker.set_time_provider(lambda: new_time) assert broker._get_current_time() == new_time class TestBacktestMarginBasics: """Test basic margin operations in backtest context.""" def setup_method(self): """Set up mock backtrader strategy for tests.""" # Create mock backtrader strategy self.mock_bt_strategy = MagicMock() self.mock_bt_strategy.broker.getcash.return_value = 10000.0 self.mock_bt_strategy.broker.getvalue.return_value = 10000.0 self.mock_bt_strategy.data.close = [50000.0] # Current price # Mock datetime to return simulated time mock_datetime_attr = MagicMock() mock_datetime_attr.datetime.return_value = datetime(2023, 6, 15, 12, 0, 0) self.mock_bt_strategy.data.datetime = mock_datetime_attr @patch('backtest_strategy_instance.PaperMarginBroker') def test_margin_broker_initialized_with_time_provider(self, mock_broker_class): """BacktestStrategyInstance should initialize margin broker with time_provider.""" from backtest_strategy_instance import BacktestStrategyInstance instance = BacktestStrategyInstance( strategy_instance_id='test-123', strategy_id='strat-1', strategy_name='Test Strategy', user_id=1, generated_code='', data_cache=MagicMock(), indicators=None, trades=None ) # Set backtrader strategy which should initialize margin broker instance.set_backtrader_strategy(self.mock_bt_strategy) # Verify margin broker was created with time_provider mock_broker_class.assert_called_once() call_kwargs = mock_broker_class.call_args.kwargs assert 'time_provider' in call_kwargs assert call_kwargs['time_provider'] is not None class TestBacktestMarginMethods: """Test margin trading methods in BacktestStrategyInstance.""" def setup_method(self): """Set up BacktestStrategyInstance for tests.""" from backtest_strategy_instance import BacktestStrategyInstance # Create instance self.instance = BacktestStrategyInstance( strategy_instance_id='test-123', strategy_id='strat-1', strategy_name='Test Strategy', user_id=1, generated_code='', data_cache=MagicMock(), indicators=None, trades=None ) # Create mock backtrader strategy self.mock_bt_strategy = MagicMock() self.mock_bt_strategy.broker.getcash.return_value = 10000.0 self.mock_bt_strategy.broker.getvalue.return_value = 10000.0 self.mock_bt_strategy.broker.setcash = MagicMock() self.mock_bt_strategy.data.close = [50000.0] # Mock datetime mock_datetime_attr = MagicMock() mock_datetime_attr.datetime.return_value = datetime(2023, 6, 15, 12, 0, 0) self.mock_bt_strategy.data.datetime = mock_datetime_attr # Set backtrader strategy self.instance.set_backtrader_strategy(self.mock_bt_strategy) # Set up exec_context self.instance.exec_context['current_symbol'] = 'BTC/USDT' def test_has_margin_position_returns_false_when_no_position(self): """has_margin_position should return False when no position exists.""" assert self.instance.has_margin_position() is False assert self.instance.has_margin_position('BTC/USDT') is False def test_get_unrealized_pnl_returns_zero_when_no_position(self): """get_unrealized_pnl should return 0 when no position exists.""" assert self.instance.get_unrealized_pnl() == 0.0 def test_get_total_margin_used_returns_zero_when_no_position(self): """get_total_margin_used should return 0 when no position exists.""" assert self.instance.get_total_margin_used() == 0.0 def test_get_liquidation_buffer_pct_returns_100_when_no_position(self): """get_liquidation_buffer_pct should return 100 when no position exists.""" assert self.instance.get_liquidation_buffer_pct() == 100.0 def test_open_margin_position_clamps_leverage(self): """open_margin_position should clamp leverage to valid range (like paper/live).""" # Leverage 15 should be clamped to 10 (max) # This matches paper/live behavior result = self.instance.open_margin_position( side='long', collateral=100.0, leverage=15.0 # Too high - should be clamped to 10 ) assert result['success'] is True assert result['leverage'] == 10 # Clamped def test_open_margin_position_validates_collateral(self): """open_margin_position should reject invalid collateral.""" with pytest.raises(ValueError, match="Collateral must be positive"): self.instance.open_margin_position( side='long', collateral=-100.0, leverage=2.0 ) def test_margin_rejects_non_active_symbol(self): """Backtest margin should reject symbols that don't match active feed.""" with pytest.raises(ValueError, match="only supports active symbol"): self.instance._get_margin_price('ETH/USDT') # Different from BTC/USDT class TestBacktestMarginBlocklyPayloads: """Test margin trading with actual Blockly payload shapes.""" def setup_method(self): """Set up BacktestStrategyInstance for tests.""" from backtest_strategy_instance import BacktestStrategyInstance self.instance = BacktestStrategyInstance( strategy_instance_id='test-123', strategy_id='strat-1', strategy_name='Test Strategy', user_id=1, generated_code='', data_cache=MagicMock(), indicators=None, trades=None ) # Create mock backtrader strategy self.mock_bt_strategy = MagicMock() self.mock_bt_strategy.broker.getcash.return_value = 10000.0 self.mock_bt_strategy.broker.getvalue.return_value = 10000.0 self.mock_bt_strategy.broker.setcash = MagicMock() self.mock_bt_strategy.data.close = [50000.0] # Mock datetime mock_datetime_attr = MagicMock() mock_datetime_attr.datetime.return_value = datetime(2023, 6, 15, 12, 0, 0) self.mock_bt_strategy.data.datetime = mock_datetime_attr # Set backtrader strategy self.instance.set_backtrader_strategy(self.mock_bt_strategy) # Set up exec_context self.instance.exec_context['current_symbol'] = 'BTC/USDT' def test_sltp_with_dict_payloads(self): """SL/TP with Blockly dict payloads like {'value': 49000.0} should work.""" result = self.instance.open_margin_position( side='long', collateral=100.0, leverage=2.0, stop_loss={'value': 49000.0}, # Blockly payload shape take_profit={'value': 55000.0} # Blockly payload shape ) assert result['success'] is True def test_limit_entry_with_blockly_payload(self): """Limit entry with Blockly payload {'limit': 48000.0} should work.""" result = self.instance.open_margin_position( side='long', collateral=100.0, leverage=2.0, limit={'limit': 48000.0} # Blockly payload shape ) assert result['success'] is True assert result['pending'] is True assert result['entry_type'] == 'limit' def test_trailing_limit_with_blockly_payload(self): """Trailing limit with Blockly payload {'trail_limit_distance': 500.0} should work.""" result = self.instance.open_margin_position( side='long', collateral=100.0, leverage=2.0, trailing_limit={'trail_limit_distance': 500.0} # Blockly payload shape ) assert result['success'] is True assert result['pending'] is True assert result['entry_type'] == 'trailing_limit' def test_trailing_stop_with_blockly_payload(self): """Trailing stop with Blockly payload {'trail_distance': 500.0} should work.""" result = self.instance.open_margin_position( side='long', collateral=100.0, leverage=2.0, trailing_stop={'trail_distance': 500.0} # Blockly payload shape ) assert result['success'] is True # Trailing stop is set up for the position trailing_stops = self.instance._get_margin_trailing_stops() assert 'BTC/USDT' in trailing_stops def test_target_market_matching_active_symbol(self): """target_market matching active symbol should work.""" result = self.instance.open_margin_position( side='long', collateral=100.0, leverage=2.0, target_market={'symbol': 'BTC/USDT'} # Matches active symbol ) assert result['success'] is True def test_target_market_different_from_active_symbol_rejected(self): """target_market with different symbol should be rejected.""" with pytest.raises(ValueError, match="only supports active symbol"): self.instance.open_margin_position( side='long', collateral=100.0, leverage=2.0, target_market={'symbol': 'ETH/USDT'} # Different from BTC/USDT ) def test_default_leverage_uses_base_class_method(self): """Default leverage should use get_margin_leverage() from base class.""" # Set a specific default leverage using the correct variable name self.instance.variables['_margin_leverage_default'] = 5 result = self.instance.open_margin_position( side='long', collateral=100.0, leverage=None # Use default ) assert result['success'] is True assert result['leverage'] == 5 # Should use the variable class TestBacktestMarginPendingEntries: """Regression tests for pending backtest margin entries.""" def setup_method(self): """Set up BacktestStrategyInstance with stateful cash for reservation checks.""" from backtest_strategy_instance import BacktestStrategyInstance self.instance = BacktestStrategyInstance( strategy_instance_id='test-123', strategy_id='strat-1', strategy_name='Test Strategy', user_id=1, generated_code='', data_cache=MagicMock(), indicators=None, trades=None ) self.cash_state = {'cash': 10000.0} self.mock_bt_strategy = MagicMock() self.mock_bt_strategy.broker.getcash.side_effect = lambda: self.cash_state['cash'] self.mock_bt_strategy.broker.getvalue.side_effect = lambda: self.cash_state['cash'] self.mock_bt_strategy.broker.setcash.side_effect = lambda value: self.cash_state.__setitem__('cash', value) self.mock_bt_strategy.data.close = [50000.0] mock_datetime_attr = MagicMock() mock_datetime_attr.datetime.return_value = datetime(2023, 6, 15, 12, 0, 0) self.mock_bt_strategy.data.datetime = mock_datetime_attr self.instance.set_backtrader_strategy(self.mock_bt_strategy) self.instance.exec_context['current_symbol'] = 'BTC/USDT' def test_trailing_limit_pending_survives_flat_bar_without_refunding_collateral(self): """Trailing-limit entries should stay pending and keep collateral reserved until triggered.""" result = self.instance.open_margin_position( side='long', collateral=100.0, leverage=2.0, trailing_limit={'trail_limit_distance': 500.0} ) assert result['pending'] is True assert self.cash_state['cash'] == 9900.0 assert self.instance.paper_margin_broker.get_balance() == 100.0 events = self.instance.process_margin_tick() assert events == [] assert self.instance.has_margin_position() is False assert 'BTC/USDT' in self.instance._get_pending_margin_entries() assert self.cash_state['cash'] == 9900.0 assert self.instance.paper_margin_broker.get_balance() == 100.0 def test_duplicate_pending_margin_entry_rejected_without_extra_reservation(self): """A second pending entry for the same symbol should fail without reserving more collateral.""" self.instance.open_margin_position( side='long', collateral=100.0, leverage=2.0, limit={'limit': 48000.0} ) assert self.cash_state['cash'] == 9900.0 assert self.instance.paper_margin_broker.get_balance() == 100.0 with pytest.raises(ValueError, match='Pending margin entry already exists'): self.instance.open_margin_position( side='long', collateral=200.0, leverage=2.0, limit={'limit': 47000.0} ) pending = self.instance._get_pending_margin_entries() assert len(pending) == 1 assert pending['BTC/USDT']['collateral'] == 100.0 assert self.cash_state['cash'] == 9900.0 assert self.instance.paper_margin_broker.get_balance() == 100.0 def test_ioc_non_marketable_limit_fails_without_pending_entry(self): """IOC/FOK limit orders should fail immediately when not marketable.""" with pytest.raises(RuntimeError, match='IOC margin limit order could not be filled immediately'): self.instance.open_margin_position( side='long', collateral=100.0, leverage=2.0, tif='IOC', limit={'limit': 48000.0} ) assert self.instance._get_pending_margin_entries() == {} assert self.cash_state['cash'] == 10000.0 assert self.instance.paper_margin_broker.get_balance() == 0.0 def test_pending_limit_fills_at_limit_price(self): """Pending limit entries should fill at the requested limit price, not the bar price.""" result = self.instance.open_margin_position( side='long', collateral=100.0, leverage=2.0, limit={'limit': 48000.0} ) assert result['pending'] is True assert self.cash_state['cash'] == 9900.0 self.mock_bt_strategy.data.close = [47000.0] events = self.instance.process_margin_tick() position = self.instance.paper_margin_broker.get_position('BTC/USDT') assert position is not None assert position.entry_price == 48000.0 assert events[0]['type'] == 'margin_entry_filled' assert events[0]['entry_price'] == 48000.0 assert self.instance._get_pending_margin_entries() == {} assert self.cash_state['cash'] == 9900.0 assert self.instance.paper_margin_broker.get_balance() == 0.0 class TestBacktestMarginIntegration: """Integration tests for margin trading in backtests.""" def test_process_margin_tick_returns_empty_when_no_broker(self): """process_margin_tick should return empty list when no margin broker.""" from backtest_strategy_instance import BacktestStrategyInstance instance = BacktestStrategyInstance( strategy_instance_id='test-123', strategy_id='strat-1', strategy_name='Test Strategy', user_id=1, generated_code='', data_cache=MagicMock(), indicators=None, trades=None ) # paper_margin_broker is None before set_backtrader_strategy events = instance.process_margin_tick() assert events == [] def test_get_current_time_utc_returns_aware_datetime(self): """_get_current_time_utc should return timezone-aware datetime.""" from backtest_strategy_instance import BacktestStrategyInstance instance = BacktestStrategyInstance( strategy_instance_id='test-123', strategy_id='strat-1', strategy_name='Test Strategy', user_id=1, generated_code='', data_cache=MagicMock(), indicators=None, trades=None ) # Mock backtrader strategy with naive datetime mock_bt_strategy = MagicMock() mock_datetime_attr = MagicMock() mock_datetime_attr.datetime.return_value = datetime(2023, 6, 15, 12, 0, 0) # Naive mock_bt_strategy.data.datetime = mock_datetime_attr mock_bt_strategy.broker.getcash.return_value = 10000.0 mock_bt_strategy.broker.getvalue.return_value = 10000.0 instance.set_backtrader_strategy(mock_bt_strategy) # Should return UTC-aware datetime result = instance._get_current_time_utc() assert result.tzinfo is not None assert result.tzinfo == timezone.utc class TestUndefinedValueBehavior: """Test that UndefinedValue prevents crashes and warns appropriately.""" def test_undefined_value_comparison_returns_false(self): """Comparisons with UndefinedValue should return False.""" from StrategyInstance import UndefinedValue undef = UndefinedValue('test_var') assert (100.0 < undef) is False assert (100.0 > undef) is False assert (100.0 <= undef) is False assert (100.0 >= undef) is False assert (undef < 100.0) is False assert (undef > 100.0) is False def test_undefined_value_math_returns_zero(self): """Math operations with UndefinedValue should return 0.""" from StrategyInstance import UndefinedValue undef = UndefinedValue('test_var') assert undef + 10 == 0 assert undef - 10 == 0 assert undef * 10 == 0 assert undef / 10 == 0 assert 10 + undef == 10 # 10 + undefined = 10 assert 10 - undef == 10 # 10 - undefined = 10 assert 10 * undef == 0 # 10 * undefined = 0 def test_undefined_value_equality_with_none(self): """UndefinedValue == None should be True for backwards compat.""" from StrategyInstance import UndefinedValue undef = UndefinedValue('test_var') assert (undef == None) is True assert (undef != None) is False def test_undefined_value_warns_via_callback(self): """UndefinedValue should call warn callback when used.""" from StrategyInstance import UndefinedValue warnings = [] undef = UndefinedValue('my_var', warn_callback=warnings.append) _ = undef < 100.0 assert len(warnings) == 1 assert 'my_var' in warnings[0] assert 'comparison' in warnings[0] def test_strategy_variables_returns_undefined_for_missing_key(self): """StrategyVariables.get() should return UndefinedValue for missing keys.""" from StrategyInstance import StrategyVariables, UndefinedValue variables = StrategyVariables() result = variables.get('nonexistent') assert isinstance(result, UndefinedValue) def test_strategy_variables_returns_real_value_for_existing_key(self): """StrategyVariables.get() should return actual value for existing keys.""" from StrategyInstance import StrategyVariables variables = StrategyVariables() variables['my_key'] = 42.5 result = variables.get('my_key') assert result == 42.5 assert not isinstance(result, type(None)) def test_strategy_variables_honors_explicit_non_none_default(self): """Real dict.get(default) call sites should still receive their explicit fallback.""" from StrategyInstance import StrategyVariables variables = StrategyVariables() assert variables.get('missing', 3) == 3 def test_strategy_variables_keeps_none_default_undefined_safe_for_old_generated_code(self): """Existing generated code using get(..., None) should still receive UndefinedValue.""" from StrategyInstance import StrategyVariables, UndefinedValue variables = StrategyVariables() assert isinstance(variables.get('missing', None), UndefinedValue) def test_undefined_warning_is_deduplicated_for_same_variable_in_one_tick(self): """Repeated reads of the same missing variable should only warn once per tick.""" from StrategyInstance import StrategyVariables warnings = [] variables = StrategyVariables(warn_callback=warnings.append) result = 100 < variables.get('sl_value') or 100 > variables.get('sl_value') assert result is False assert len(warnings) == 1 assert 'sl_value' in warnings[0] def test_undefined_warning_resets_between_ticks(self): """Missing-variable warnings should be able to fire again on a later tick.""" from StrategyInstance import StrategyVariables warnings = [] variables = StrategyVariables(warn_callback=warnings.append) _ = 100 < variables.get('sl_value') variables.reset_undefined_warnings() _ = 100 < variables.get('sl_value') assert len(warnings) == 2 def test_undefined_in_complex_expression_doesnt_crash(self): """Complex expressions with undefined variables should not crash.""" from StrategyInstance import StrategyVariables variables = StrategyVariables() price = 70000.0 # This would crash with None: price < variables.get('sl_value') or price > variables.get('tp_value') # With UndefinedValue, it should return False without crashing result = price < variables.get('sl_value') or price > variables.get('tp_value') assert result is False def test_undefined_value_format_string(self): """UndefinedValue should handle f-string formatting without crashing.""" from StrategyInstance import UndefinedValue undef = UndefinedValue('trade_value') # These would crash without __format__ method formatted_float = f"{undef:.2f}" formatted_int = f"{undef:d}" formatted_plain = f"{undef}" assert formatted_float == "0.00" assert formatted_int == "0" assert formatted_plain == "" def test_default_margin_leverage_without_override_stays_three(self): """Backtest margin should still use the documented 3x default leverage.""" from backtest_strategy_instance import BacktestStrategyInstance instance = BacktestStrategyInstance( strategy_instance_id='test-123', strategy_id='strat-1', strategy_name='Test Strategy', user_id=1, generated_code='', data_cache=MagicMock(), indicators=None, trades=None ) mock_bt_strategy = MagicMock() mock_bt_strategy.broker.getcash.return_value = 10000.0 mock_bt_strategy.broker.getvalue.return_value = 10000.0 mock_bt_strategy.broker.setcash = MagicMock() mock_bt_strategy.data.close = [50000.0] mock_datetime_attr = MagicMock() mock_datetime_attr.datetime.return_value = datetime(2023, 6, 15, 12, 0, 0) mock_bt_strategy.data.datetime = mock_datetime_attr instance.set_backtrader_strategy(mock_bt_strategy) instance.exec_context['current_symbol'] = 'BTC/USDT' result = instance.open_margin_position( side='long', collateral=100.0, leverage=None ) assert result['success'] is True assert result['leverage'] == 3 assert not any('_margin_leverage_default' in alert['message'] for alert in instance.collected_alerts)