""" 执行链安全修复回归测试 覆盖重点： - Bitget 单币种平仓，不再误触发全仓平仓 - Bitget 单笔撤单接口可用 - Bitget 移动止损走 set_tp_sl，而不是不存在的方法 """ import asyncio import importlib.util import os import sys import types from pathlib import Path from unittest.mock import AsyncMock, MagicMock sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) def make_bitget_service(): from app.services.bitget_live_trading_service import BitgetLiveTradingService mock_api = MagicMock() mock_exchange = MagicMock() mock_api.exchange = mock_exchange mock_api._standardize_symbol = lambda s: f"{s.replace('USDT', '')}/USDT:USDT" mock_settings = MagicMock() mock_settings.bitget_max_total_leverage = 10.0 mock_settings.bitget_max_single_position = 1000.0 service = BitgetLiveTradingService.__new__(BitgetLiveTradingService) service.settings = mock_settings service.max_total_leverage = mock_settings.bitget_max_total_leverage service.max_single_position = mock_settings.bitget_max_single_position service.circuit_breaker_drawdown = 0.25 service.trading_api = mock_api service.initial_balance = 10000.0 return service, mock_api def load_bitget_executor_class(): """按文件加载执行器，避免触发 app.crypto_agent.__init__ 的重依赖""" executor_dir = Path(__file__).resolve().parents[1] / 'app' / 'crypto_agent' / 'executor' if 'app.crypto_agent' not in sys.modules: crypto_pkg = types.ModuleType('app.crypto_agent') crypto_pkg.__path__ = [str(executor_dir.parent)] sys.modules['app.crypto_agent'] = crypto_pkg if 'app.crypto_agent.executor' not in sys.modules: executor_pkg = types.ModuleType('app.crypto_agent.executor') executor_pkg.__path__ = [str(executor_dir)] sys.modules['app.crypto_agent.executor'] = executor_pkg if 'app.crypto_agent.executor.base_executor' not in sys.modules: base_spec = importlib.util.spec_from_file_location( 'app.crypto_agent.executor.base_executor', executor_dir / 'base_executor.py', ) base_module = importlib.util.module_from_spec(base_spec) sys.modules[base_spec.name] = base_module base_spec.loader.exec_module(base_module) if 'app.crypto_agent.executor.bitget_executor' not in sys.modules: executor_spec = importlib.util.spec_from_file_location( 'app.crypto_agent.executor.bitget_executor', executor_dir / 'bitget_executor.py', ) executor_module = importlib.util.module_from_spec(executor_spec) sys.modules[executor_spec.name] = executor_module executor_spec.loader.exec_module(executor_module) return sys.modules['app.crypto_agent.executor.bitget_executor'].BitgetExecutor def test_bitget_market_close_position_only_closes_requested_symbol(): service, mock_api = make_bitget_service() mock_api.get_position.return_value = [ { 'symbol': 'BTC/USDT:USDT', 'side': 'long', 'entryPrice': 50000.0, 'unrealizedPnl': 100.0, 'leverage': 5, 'liquidationPrice': 45000.0, 'info': {'available': '0.02'}, }, { 'symbol': 'ETH/USDT:USDT', 'side': 'long', 'entryPrice': 3000.0, 'unrealizedPnl': 50.0, 'leverage': 5, 'liquidationPrice': 2500.0, 'info': {'available': '0.5'}, }, ] mock_api.get_open_orders.return_value = [] mock_api.cancel_all_orders.return_value = True mock_api.exchange.create_market_order.return_value = {'id': 'close-btc'} result = service.market_close_position('BTCUSDT') assert result['success'] is True assert result['coin'] == 'BTC' kwargs = mock_api.exchange.create_market_order.call_args.kwargs assert kwargs['symbol'] == 'BTC/USDT:USDT' assert kwargs['side'] == 'sell' def test_bitget_cancel_order_delegates_to_sdk(): service, mock_api = make_bitget_service() mock_api.cancel_order.return_value = True service.get_open_orders = MagicMock(return_value=[{'order_id': 'oid-1'}]) result = service.cancel_order('BTC', 'oid-1') assert result['success'] is True mock_api.cancel_order.assert_called_once_with(symbol='BTC', order_id='oid-1') def test_bitget_cancel_order_treats_missing_open_order_as_already_closed(): service, mock_api = make_bitget_service() service.get_open_orders = MagicMock(return_value=[]) result = service.cancel_order('SOL', 'oid-missing') assert result['success'] is True assert result['already_closed'] is True mock_api.cancel_order.assert_not_called() def test_bitget_executor_close_uses_symbol_close_not_close_all(): BitgetExecutor = load_bitget_executor_class() executor = BitgetExecutor.__new__(BitgetExecutor) executor.bitget = MagicMock() executor.bitget.settings.bitget_default_leverage = 10 executor.send_execution_notification = AsyncMock() executor.bitget.market_close_position.return_value = {'success': True, 'coin': 'BTC'} result = asyncio.run(executor.execute_close({'symbol': 'BTCUSDT'}, 50000.0)) assert result['success'] is True executor.bitget.market_close_position.assert_called_once_with('BTC') executor.bitget.market_close_all.assert_not_called() def test_bitget_executor_move_stop_loss_uses_set_tp_sl(): BitgetExecutor = load_bitget_executor_class() executor = BitgetExecutor.__new__(BitgetExecutor) executor.bitget = MagicMock() executor.bitget.get_position_for_symbol.return_value = {'size': 0.02} executor.bitget.get_tp_sl_prices.return_value = {'take_profit': 55000.0} executor.bitget.set_tp_sl.return_value = {'success': True} result = asyncio.run(executor.move_stop_loss('BTCUSDT', 49000.0)) assert result['success'] is True executor.bitget.set_tp_sl.assert_called_once_with( symbol='BTC', is_long=True, size=0.02, tp_price=55000.0, sl_price=49000.0, ) def test_bitget_executor_open_uses_actual_leverage_for_contracts(): BitgetExecutor = load_bitget_executor_class() executor = BitgetExecutor.__new__(BitgetExecutor) executor.bitget = MagicMock() executor.bitget.settings.bitget_default_leverage = 10 executor.send_execution_notification = AsyncMock() executor.decide_order_type = MagicMock(return_value=('market', 'test')) executor.calculate_effective_margin = MagicMock(return_value=32.0) executor.bitget.get_account_state.return_value = {'available_balance': 1000.0} executor.bitget.get_contract_size.return_value = 0.1 executor.bitget.place_market_order.return_value = { 'success': True, 'order_id': 'oid-1', 'order_status': 'filled', } result = asyncio.run( executor.execute_open( { 'symbol': 'ETHUSDT', 'action': 'buy', 'margin': 32.0, 'entry_price': 2000.0, 'stop_loss': 1980.0, 'take_profit': 2040.0, }, 2000.0, ) ) assert result['success'] is True executor.bitget.update_leverage.assert_called_once_with('ETH', 10) executor.bitget.place_market_order.assert_called_once_with('ETH', is_buy=True, size=1)