123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114 |
- import unittest
- import asyncio
- from unittest.mock import Mock, patch, AsyncMock
- from datetime import datetime, timezone
- from src.monitoring.simple_position_tracker import SimplePositionTracker
- from src.commands.info.positions import PositionsCommands
- class TestPositionROE(unittest.TestCase):
- def setUp(self):
- self.trading_engine = Mock()
- self.notification_manager = Mock()
- self.position_tracker = SimplePositionTracker(self.trading_engine, self.notification_manager)
- self.stats = Mock()
- self.timestamp = datetime.now(timezone.utc)
- async def async_test_position_size_change_roe_calculation(self):
- """Test ROE calculation during position size changes."""
- # Mock exchange position data
- exchange_pos = {
- 'contracts': 0.01,
- 'side': 'long',
- 'entryPrice': 50000.0,
- 'unrealizedPnl': -16.21,
- 'info': {
- 'position': {
- 'returnOnEquity': '-0.324' # -32.4%
- }
- }
- }
-
- # Mock database position data
- db_pos = {
- 'trade_lifecycle_id': 'test_lifecycle',
- 'current_position_size': 0.005,
- 'position_side': 'long',
- 'entry_price': 50000.0
- }
-
- # Mock stats manager
- self.stats.trade_manager.update_trade_market_data = AsyncMock(return_value=True)
-
- # Call the method
- await self.position_tracker._handle_position_size_change('BTC/USD', exchange_pos, db_pos, self.stats, self.timestamp)
-
- # Verify the call
- call_args = self.stats.trade_manager.update_trade_market_data.call_args[1]
- self.assertEqual(call_args['roe_percentage'], -32.4) # Should match exchange ROE
- def test_position_size_change_roe_calculation(self):
- """Test ROE calculation during position size changes."""
- asyncio.run(self.async_test_position_size_change_roe_calculation())
- async def async_test_position_opened_roe_calculation(self):
- """Test ROE calculation when position is opened."""
- # Mock exchange position data
- exchange_pos = {
- 'contracts': 0.01,
- 'side': 'long',
- 'entryPrice': 50000.0,
- 'unrealizedPnl': -16.21,
- 'info': {
- 'position': {
- 'returnOnEquity': '-0.324' # -32.4%
- }
- }
- }
-
- # Mock stats manager
- self.stats.create_trade_lifecycle = AsyncMock(return_value='test_lifecycle')
- self.stats.update_trade_position_opened = AsyncMock(return_value=True)
-
- # Call the method
- await self.position_tracker._handle_position_opened('BTC/USD', exchange_pos, self.stats, self.timestamp)
-
- # Verify the call
- call_args = self.stats.update_trade_position_opened.call_args[1]
- self.assertEqual(call_args['roe_percentage'], -32.4) # Should match exchange ROE
- def test_position_opened_roe_calculation(self):
- """Test ROE calculation when position is opened."""
- asyncio.run(self.async_test_position_opened_roe_calculation())
- async def async_test_positions_command_roe_display(self):
- """Test ROE display in positions command."""
- # Mock open positions data
- open_positions = [{
- 'symbol': 'BTC/USD',
- 'position_side': 'long',
- 'current_position_size': 0.01,
- 'entry_price': 50000.0,
- 'duration': '1h',
- 'unrealized_pnl': -16.21,
- 'roe_percentage': -32.4, # ROE from database
- 'position_value': 500.0,
- 'margin_used': 50.0,
- 'leverage': 10.0,
- 'liquidation_price': 45000.0
- }]
-
- # Mock stats manager
- self.stats.get_open_positions = AsyncMock(return_value=open_positions)
-
- # Call the positions command
- result = await positions_command(None, self.stats)
-
- # Verify the output
- self.assertIn('-32.4%', result) # Should display ROE correctly
- def test_positions_command_roe_display(self):
- """Test ROE display in positions command."""
- asyncio.run(self.async_test_positions_command_roe_display())
- if __name__ == '__main__':
- unittest.main()
|