"""Tests for paper trading engine. This module contains tests for PaperTradingEngine, Trade, FeeConfig, SlippageConfig, PositionSummary, and AccountSummary. """ import asyncio import pytest from datetime import datetime, timedelta from openclaw.exchange.paper_trading import ( PaperTradingEngine, PaperTradingError, PriceFeedError, Trade, FeeConfig, SlippageConfig, PositionSummary, AccountSummary, ) from openclaw.exchange.models import OrderSide, OrderStatus, OrderType from openclaw.exchange.base import InsufficientFundsError, InvalidOrderError class TestPaperTradingEngine: """Tests for PaperTradingEngine class.""" @pytest.fixture def engine(self): """Fixture to create a PaperTradingEngine instance.""" return PaperTradingEngine( name="test_engine", initial_capital={"USDT": 10000.0, "BTC": 1.0}, ) def test_engine_initialization(self, engine): """Test PaperTradingEngine initialization.""" assert engine.name == "test_engine" assert "USDT" in engine.balances assert "BTC" in engine.balances assert engine.balances["USDT"].free == 10000.0 assert engine.balances["BTC"].free == 1.0 def test_default_initial_capital(self): """Test default initial capital.""" engine = PaperTradingEngine() assert "USDT" in engine.balances assert engine.balances["USDT"].free == 10000.0 def test_custom_fee_config(self): """Test custom fee configuration.""" fee_config = FeeConfig(maker_fee=0.0005, taker_fee=0.0015) engine = PaperTradingEngine(fee_config=fee_config) assert engine.fee_config.maker_fee == 0.0005 assert engine.fee_config.taker_fee == 0.0015 def test_custom_slippage_config(self): """Test custom slippage configuration.""" slippage_config = SlippageConfig(enabled=True, base_slippage_pct=0.1) engine = PaperTradingEngine(slippage_config=slippage_config) assert engine.slippage_config.enabled is True assert engine.slippage_config.base_slippage_pct == 0.1 @pytest.mark.asyncio async def test_update_price(self, engine): """Test updating price for a symbol.""" await engine.update_price("BTC/USDT", 50000.0) # Price should be cached price = engine._get_price("BTC/USDT") assert price == 50000.0 @pytest.mark.asyncio async def test_place_market_buy_order(self, engine): """Test placing a market buy order.""" await engine.update_price("BTC/USDT", 50000.0) order = await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) assert order.symbol == "BTC/USDT" assert order.side == OrderSide.BUY assert order.amount == 0.1 assert order.status == OrderStatus.FILLED assert order.filled_amount == 0.1 assert order.order_type == OrderType.MARKET @pytest.mark.asyncio async def test_place_market_sell_order(self, engine): """Test placing a market sell order.""" await engine.update_price("BTC/USDT", 50000.0) order = await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.SELL, amount=0.5, ) assert order.symbol == "BTC/USDT" assert order.side == OrderSide.SELL assert order.amount == 0.5 assert order.status == OrderStatus.FILLED @pytest.mark.asyncio async def test_market_buy_without_price_feed(self, engine): """Test that market buy fails without price.""" with pytest.raises(PriceFeedError): await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) @pytest.mark.asyncio async def test_insufficient_funds_buy(self, engine): """Test buy order with insufficient funds.""" await engine.update_price("BTC/USDT", 50000.0) with pytest.raises(InsufficientFundsError): await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=1.0, # Costs $50k + fees, only have $10k ) @pytest.mark.asyncio async def test_insufficient_funds_sell(self, engine): """Test sell order with insufficient funds.""" await engine.update_price("BTC/USDT", 50000.0) with pytest.raises(InsufficientFundsError): await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.SELL, amount=2.0, # Only have 1 BTC ) @pytest.mark.asyncio async def test_buy_updates_balances(self, engine): """Test that buy order updates balances correctly.""" await engine.update_price("BTC/USDT", 50000.0) initial_usdt = engine.balances["USDT"].free await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) # USDT should decrease, BTC should increase assert engine.balances["BTC"].free == 1.0 + 0.1 assert engine.balances["USDT"].free < initial_usdt @pytest.mark.asyncio async def test_sell_updates_balances(self, engine): """Test that sell order updates balances correctly.""" await engine.update_price("BTC/USDT", 50000.0) initial_btc = engine.balances["BTC"].free await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.SELL, amount=0.5, ) # BTC should decrease, USDT should increase assert engine.balances["BTC"].free == initial_btc - 0.5 assert engine.balances["USDT"].free > 10000.0 @pytest.mark.asyncio async def test_place_limit_order(self, engine): """Test placing a limit order.""" await engine.update_price("BTC/USDT", 50000.0) order = await engine.place_limit_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, price=45000.0, ) assert order.symbol == "BTC/USDT" assert order.side == OrderSide.BUY assert order.amount == 0.1 assert order.price == 45000.0 assert order.status == OrderStatus.OPEN assert order.filled_amount == 0.0 @pytest.mark.asyncio async def test_limit_order_invalid_price(self, engine): """Test limit order with invalid price.""" with pytest.raises(InvalidOrderError): await engine.place_limit_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, price=-100.0, ) @pytest.mark.asyncio async def test_limit_order_fills_when_price_reached(self, engine): """Test that limit order fills when price is reached.""" await engine.update_price("BTC/USDT", 50000.0) # Place buy limit at $45k order = await engine.place_limit_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, price=45000.0, ) assert order.status == OrderStatus.OPEN # Price drops to $44k (below limit) await engine.update_price("BTC/USDT", 44000.0) # Order should now be filled filled_order = engine.get_order(order.order_id) assert filled_order.status == OrderStatus.FILLED @pytest.mark.asyncio async def test_limit_sell_order_fills_when_price_reached(self, engine): """Test that sell limit order fills when price is reached.""" await engine.update_price("BTC/USDT", 50000.0) # Place sell limit at $55k order = await engine.place_limit_order( symbol="BTC/USDT", side=OrderSide.SELL, amount=0.1, price=55000.0, ) assert order.status == OrderStatus.OPEN # Price rises to $56k (above limit) await engine.update_price("BTC/USDT", 56000.0) # Order should now be filled filled_order = engine.get_order(order.order_id) assert filled_order.status == OrderStatus.FILLED @pytest.mark.asyncio async def test_cancel_order(self, engine): """Test cancelling a pending order.""" await engine.update_price("BTC/USDT", 50000.0) order = await engine.place_limit_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, price=45000.0, ) initial_usdt_locked = engine.balances["USDT"].locked assert initial_usdt_locked > 0 cancelled = await engine.cancel_order(order.order_id) assert cancelled is True # Funds should be unlocked assert engine.balances["USDT"].locked == 0.0 # Order status should be cancelled updated_order = engine.get_order(order.order_id) assert updated_order.status == OrderStatus.CANCELLED def test_cancel_nonexistent_order(self, engine): """Test cancelling nonexistent order.""" cancelled = asyncio.run(engine.cancel_order("nonexistent")) assert cancelled is False @pytest.mark.asyncio async def test_get_open_orders(self, engine): """Test getting open orders.""" await engine.update_price("BTC/USDT", 50000.0) # Place limit orders (use smaller amounts to stay within balance) await engine.place_limit_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.05, price=45000.0, ) await engine.place_limit_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.05, price=44000.0, ) open_orders = engine.get_open_orders() assert len(open_orders) == 2 @pytest.mark.asyncio async def test_get_open_orders_filtered(self, engine): """Test getting open orders filtered by symbol.""" await engine.update_price("BTC/USDT", 50000.0) await engine.update_price("ETH/USDT", 3000.0) await engine.place_limit_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, price=45000.0, ) await engine.place_limit_order( symbol="ETH/USDT", side=OrderSide.BUY, amount=1.0, price=2800.0, ) btc_orders = engine.get_open_orders("BTC/USDT") assert len(btc_orders) == 1 assert btc_orders[0].symbol == "BTC/USDT" @pytest.mark.asyncio async def test_position_created_after_buy(self, engine): """Test that position is created after buy order.""" await engine.update_price("BTC/USDT", 50000.0) # No positions initially assert len(engine.get_positions()) == 0 await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) # Should have position now positions = engine.get_positions() assert len(positions) == 1 position = engine.get_position("BTC/USDT") assert position is not None assert position.symbol == "BTC/USDT" assert position.side == OrderSide.BUY assert position.amount == 0.1 @pytest.mark.asyncio async def test_position_updated_on_additional_buy(self, engine): """Test position averaging on additional buy.""" await engine.update_price("BTC/USDT", 50000.0) await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.05, ) # Update price and buy more (use smaller amount to stay within balance) await engine.update_price("BTC/USDT", 55000.0) await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.05, ) position = engine.get_position("BTC/USDT") assert position.amount == 0.1 # Average entry price should be between 50k and 55k assert 50000.0 < position.entry_price < 55000.0 @pytest.mark.asyncio async def test_realized_pnl_on_sell(self, engine): """Test realized PnL calculation on sell.""" await engine.update_price("BTC/USDT", 50000.0) # Buy at $50k await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) # Sell at $55k (10% gain) await engine.update_price("BTC/USDT", 55000.0) await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.SELL, amount=0.1, ) # Should have realized profit assert engine.total_realized_pnl > 0 # Check trade record assert len(engine.trades) == 2 # Buy and sell sell_trade = engine.trades[1] assert sell_trade.realized_pnl > 0 @pytest.mark.asyncio async def test_unrealized_pnl_calculation(self, engine): """Test unrealized PnL calculation.""" await engine.update_price("BTC/USDT", 50000.0) await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) # Price rises to $55k await engine.update_price("BTC/USDT", 55000.0) position = engine.get_position("BTC/USDT") expected_pnl = (55000.0 - 50000.0) * 0.1 # $500 assert abs(position.unrealized_pnl - expected_pnl) < 100 # Allow for slippage def test_get_unrealized_pnl_no_positions(self, engine): """Test unrealized PnL with no positions.""" assert engine.get_unrealized_pnl() == 0.0 @pytest.mark.asyncio async def test_get_position_summary(self, engine): """Test position summary generation.""" await engine.update_price("BTC/USDT", 50000.0) await engine.update_price("ETH/USDT", 3000.0) await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) await engine.place_market_order( symbol="ETH/USDT", side=OrderSide.BUY, amount=1.0, ) summary = engine.get_position_summary() assert isinstance(summary, PositionSummary) assert summary.total_positions == 2 assert summary.total_market_value > 0 @pytest.mark.asyncio async def test_get_account_summary(self, engine): """Test account summary generation.""" await engine.update_price("BTC/USDT", 50000.0) await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) summary = engine.get_account_summary() assert isinstance(summary, AccountSummary) assert summary.initial_capital == 10001.0 # 10000 USDT + 1 BTC worth assert summary.total_trades == 1 assert summary.open_positions == 1 def test_trade_history_empty(self, engine): """Test trade history when no trades.""" history = engine.get_trade_history() assert len(history) == 0 @pytest.mark.asyncio async def test_trade_history_with_trades(self, engine): """Test trade history with trades.""" await engine.update_price("BTC/USDT", 50000.0) await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) history = engine.get_trade_history() assert len(history) == 1 assert history[0].symbol == "BTC/USDT" assert history[0].side == OrderSide.BUY @pytest.mark.asyncio async def test_trade_history_filtered(self, engine): """Test filtered trade history.""" await engine.update_price("BTC/USDT", 50000.0) await engine.update_price("ETH/USDT", 3000.0) await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) await engine.place_market_order( symbol="ETH/USDT", side=OrderSide.BUY, amount=1.0, ) btc_trades = engine.get_trade_history(symbol="BTC/USDT") assert len(btc_trades) == 1 assert btc_trades[0].symbol == "BTC/USDT" @pytest.mark.asyncio async def test_trade_history_time_filtered(self, engine): """Test trade history filtered by time.""" await engine.update_price("BTC/USDT", 50000.0) before = datetime.now() await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) after = datetime.now() # Get trades before order early_trades = engine.get_trade_history(end_time=before) assert len(early_trades) == 0 # Get trades after order recent_trades = engine.get_trade_history(start_time=before) assert len(recent_trades) == 1 @pytest.mark.asyncio async def test_generate_report(self, engine): """Test report generation.""" await engine.update_price("BTC/USDT", 50000.0) await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) report = engine.generate_report() assert isinstance(report, str) assert "Paper Trading Report" in report assert "BTC/USDT" in report assert "Account Summary" in report @pytest.mark.asyncio async def test_reset(self, engine): """Test engine reset.""" await engine.update_price("BTC/USDT", 50000.0) await engine.place_market_order( symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, ) assert len(engine.get_positions()) == 1 assert len(engine.trades) == 1 assert engine.total_realized_pnl == 0.0 # No sells yet engine.reset() # Everything should be cleared assert len(engine.get_positions()) == 0 assert len(engine.trades) == 0 assert engine.total_realized_pnl == 0.0 assert engine.balances["USDT"].free == 10000.0 assert engine.balances["BTC"].free == 1.0 @pytest.mark.asyncio async def test_price_feed_callable(self): """Test engine with price feed callable.""" def price_feed(symbol): return {"BTC/USDT": 50000.0}.get(symbol) engine = PaperTradingEngine(price_feed=price_feed) price = engine._get_price("BTC/USDT") assert price == 50000.0 class TestFeeConfig: """Tests for FeeConfig dataclass.""" def test_default_fees(self): """Test default fee configuration.""" config = FeeConfig() assert config.maker_fee == 0.001 assert config.taker_fee == 0.001 def test_custom_fees(self): """Test custom fee configuration.""" config = FeeConfig(maker_fee=0.0005, taker_fee=0.0015) assert config.maker_fee == 0.0005 assert config.taker_fee == 0.0015 class TestSlippageConfig: """Tests for SlippageConfig dataclass.""" def test_default_slippage(self): """Test default slippage configuration.""" config = SlippageConfig() assert config.enabled is True assert config.base_slippage_pct == 0.05 def test_disabled_slippage(self): """Test disabled slippage.""" config = SlippageConfig(enabled=False) assert config.enabled is False class TestTrade: """Tests for Trade dataclass.""" def test_trade_creation(self): """Test creating a Trade.""" trade = Trade( trade_id="trade123", order_id="order123", symbol="BTC/USDT", side=OrderSide.BUY, amount=0.1, price=50000.0, fee=5.0, fee_asset="USDT", timestamp=datetime.now(), ) assert trade.trade_id == "trade123" assert trade.symbol == "BTC/USDT" assert trade.side == OrderSide.BUY assert trade.realized_pnl == 0.0 def test_trade_with_pnl(self): """Test creating a Trade with realized PnL.""" trade = Trade( trade_id="trade123", order_id="order123", symbol="BTC/USDT", side=OrderSide.SELL, amount=0.1, price=55000.0, fee=5.5, fee_asset="USDT", timestamp=datetime.now(), realized_pnl=500.0, ) assert trade.realized_pnl == 500.0 class TestPaperTradingError: """Tests for PaperTradingError exception.""" def test_error_inheritance(self): """Test that PaperTradingError inherits from ExchangeError.""" assert issubclass(PaperTradingError, Exception) def test_price_feed_error(self): """Test PriceFeedError exception.""" assert issubclass(PriceFeedError, PaperTradingError)