test_position_roe.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114
  1. import unittest
  2. import asyncio
  3. from unittest.mock import Mock, patch, AsyncMock
  4. from datetime import datetime, timezone
  5. from src.monitoring.simple_position_tracker import SimplePositionTracker
  6. from src.commands.info.positions import PositionsCommands
  7. class TestPositionROE(unittest.TestCase):
  8. def setUp(self):
  9. self.trading_engine = Mock()
  10. self.notification_manager = Mock()
  11. self.position_tracker = SimplePositionTracker(self.trading_engine, self.notification_manager)
  12. self.stats = Mock()
  13. self.timestamp = datetime.now(timezone.utc)
  14. async def async_test_position_size_change_roe_calculation(self):
  15. """Test ROE calculation during position size changes."""
  16. # Mock exchange position data
  17. exchange_pos = {
  18. 'contracts': 0.01,
  19. 'side': 'long',
  20. 'entryPrice': 50000.0,
  21. 'unrealizedPnl': -16.21,
  22. 'info': {
  23. 'position': {
  24. 'returnOnEquity': '-0.324' # -32.4%
  25. }
  26. }
  27. }
  28. # Mock database position data
  29. db_pos = {
  30. 'trade_lifecycle_id': 'test_lifecycle',
  31. 'current_position_size': 0.005,
  32. 'position_side': 'long',
  33. 'entry_price': 50000.0
  34. }
  35. # Mock stats manager
  36. self.stats.trade_manager.update_trade_market_data = AsyncMock(return_value=True)
  37. # Call the method
  38. await self.position_tracker._handle_position_size_change('BTC/USD', exchange_pos, db_pos, self.stats, self.timestamp)
  39. # Verify the call
  40. call_args = self.stats.trade_manager.update_trade_market_data.call_args[1]
  41. self.assertEqual(call_args['roe_percentage'], -32.4) # Should match exchange ROE
  42. def test_position_size_change_roe_calculation(self):
  43. """Test ROE calculation during position size changes."""
  44. asyncio.run(self.async_test_position_size_change_roe_calculation())
  45. async def async_test_position_opened_roe_calculation(self):
  46. """Test ROE calculation when position is opened."""
  47. # Mock exchange position data
  48. exchange_pos = {
  49. 'contracts': 0.01,
  50. 'side': 'long',
  51. 'entryPrice': 50000.0,
  52. 'unrealizedPnl': -16.21,
  53. 'info': {
  54. 'position': {
  55. 'returnOnEquity': '-0.324' # -32.4%
  56. }
  57. }
  58. }
  59. # Mock stats manager
  60. self.stats.create_trade_lifecycle = AsyncMock(return_value='test_lifecycle')
  61. self.stats.update_trade_position_opened = AsyncMock(return_value=True)
  62. # Call the method
  63. await self.position_tracker._handle_position_opened('BTC/USD', exchange_pos, self.stats, self.timestamp)
  64. # Verify the call
  65. call_args = self.stats.update_trade_position_opened.call_args[1]
  66. self.assertEqual(call_args['roe_percentage'], -32.4) # Should match exchange ROE
  67. def test_position_opened_roe_calculation(self):
  68. """Test ROE calculation when position is opened."""
  69. asyncio.run(self.async_test_position_opened_roe_calculation())
  70. async def async_test_positions_command_roe_display(self):
  71. """Test ROE display in positions command."""
  72. # Mock open positions data
  73. open_positions = [{
  74. 'symbol': 'BTC/USD',
  75. 'position_side': 'long',
  76. 'current_position_size': 0.01,
  77. 'entry_price': 50000.0,
  78. 'duration': '1h',
  79. 'unrealized_pnl': -16.21,
  80. 'roe_percentage': -32.4, # ROE from database
  81. 'position_value': 500.0,
  82. 'margin_used': 50.0,
  83. 'leverage': 10.0,
  84. 'liquidation_price': 45000.0
  85. }]
  86. # Mock stats manager
  87. self.stats.get_open_positions = AsyncMock(return_value=open_positions)
  88. # Call the positions command
  89. result = await positions_command(None, self.stats)
  90. # Verify the output
  91. self.assertIn('-32.4%', result) # Should display ROE correctly
  92. def test_positions_command_roe_display(self):
  93. """Test ROE display in positions command."""
  94. asyncio.run(self.async_test_positions_command_roe_display())
  95. if __name__ == '__main__':
  96. unittest.main()