404 lines
14 KiB
Python
404 lines
14 KiB
Python
"""
|
|
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'
|