""" CryptoAgent 平台熔断回归测试 覆盖重点： - 账户级止损只暂停单个平台，不再要求全局停机 - 手动恢复平台会重置该平台初始权益基线 """ 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 load_crypto_agent_class(): agent_path = Path(__file__).resolve().parents[1] / 'app' / 'crypto_agent' / 'crypto_agent.py' if 'app' not in sys.modules: app_pkg = types.ModuleType('app') app_pkg.__path__ = [str(agent_path.parents[2] / 'app')] sys.modules['app'] = app_pkg for pkg_name, pkg_path in [ ('app.crypto_agent', agent_path.parent), ('app.services', agent_path.parents[1] / 'services'), ('app.utils', agent_path.parents[1] / 'utils'), ]: if pkg_name not in sys.modules: pkg = types.ModuleType(pkg_name) pkg.__path__ = [str(pkg_path)] sys.modules[pkg_name] = pkg logger_module = types.ModuleType('app.utils.logger') logger_module.logger = MagicMock() sys.modules['app.utils.logger'] = logger_module config_module = types.ModuleType('app.config') config_module.get_settings = MagicMock() sys.modules['app.config'] = config_module bitget_service_module = types.ModuleType('app.services.bitget_service') bitget_service_module.bitget_service = MagicMock() sys.modules['app.services.bitget_service'] = bitget_service_module feishu_module = types.ModuleType('app.services.feishu_service') feishu_module.get_feishu_service = MagicMock() feishu_module.get_feishu_paper_trading_service = MagicMock() feishu_module.get_feishu_error_service = MagicMock() sys.modules['app.services.feishu_service'] = feishu_module telegram_module = types.ModuleType('app.services.telegram_service') telegram_module.get_telegram_service = MagicMock() sys.modules['app.services.telegram_service'] = telegram_module dingtalk_module = types.ModuleType('app.services.dingtalk_service') dingtalk_module.get_dingtalk_service = MagicMock() sys.modules['app.services.dingtalk_service'] = dingtalk_module paper_module = types.ModuleType('app.services.paper_trading_service') paper_module.get_paper_trading_service = MagicMock() sys.modules['app.services.paper_trading_service'] = paper_module signal_db_module = types.ModuleType('app.services.signal_database_service') signal_db_module.get_signal_db_service = MagicMock() sys.modules['app.services.signal_database_service'] = signal_db_module position_sizing_module = types.ModuleType('app.services.position_sizing') position_sizing_module.DEFAULT_SIGNAL_POSITION_SIZE_BY_TIMEFRAME = {} position_sizing_module.DEFAULT_TIMEFRAME_MARGIN_MULTIPLIERS = {} position_sizing_module.calculate_margin_and_position_value = MagicMock() position_sizing_module.resolve_target_margin_pct = MagicMock() sys.modules['app.services.position_sizing'] = position_sizing_module market_analyzer_module = types.ModuleType('app.crypto_agent.market_signal_analyzer') market_analyzer_module.MarketSignalAnalyzer = MagicMock() sys.modules['app.crypto_agent.market_signal_analyzer'] = market_analyzer_module system_status_module = types.ModuleType('app.utils.system_status') system_status_module.get_system_monitor = MagicMock() system_status_module.AgentStatus = types.SimpleNamespace(RUNNING='running', STOPPED='stopped') sys.modules['app.utils.system_status'] = system_status_module module_name = 'app.crypto_agent.crypto_agent_test' spec = importlib.util.spec_from_file_location(module_name, agent_path) module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) return module.CryptoAgent def make_agent(): CryptoAgent = load_crypto_agent_class() agent = CryptoAgent.__new__(CryptoAgent) agent.settings = types.SimpleNamespace( account_max_drawdown=0.25, account_drawdown_alert=0.15, feishu_enabled=False, crypto_intraday_llm_cooldown_minutes=15, crypto_trend_llm_cooldown_minutes=60, crypto_force_llm_surge_threshold=2.0, crypto_force_llm_trade_zone_pct=0.4, crypto_event_analysis_enabled=True, crypto_event_analysis_window_minutes=5, crypto_event_analysis_price_change_percent=1.0, crypto_event_analysis_cooldown_minutes=10, paper_trading_enabled=True, bitget_trading_enabled=True, ) agent.paper_trading = None agent.bitget = None agent.bitget_services = {} agent.bitget_executors = {} agent.symbols = ['BTCUSDT'] agent.executors = {} agent._platform_halts = {} from collections import deque agent._execution_events = deque(maxlen=120) agent._analysis_events = deque(maxlen=240) agent._analysis_monitor = {} agent._analysis_notification_state = {} agent._lane_analysis_state = {} agent._event_analysis_state = {} agent.execution_guardian = MagicMock() agent.execution_guardian.get_status.return_value = {"last_status": "idle", "targets": [], "last_actions": []} agent._initial_balances = {} agent._target_execution_controls = {} agent._save_platform_halts = MagicMock() agent._save_initial_balances = MagicMock() agent._save_target_execution_controls = MagicMock() agent._send_alert_notification = AsyncMock() agent._emergency_close_all_positions = AsyncMock() return agent def test_account_stop_loss_halts_only_triggered_platform(): agent = make_agent() bitget = MagicMock() bitget.get_account_state.return_value = { 'account_value': 700.0, 'current_balance': 700.0, } agent.bitget = bitget agent._get_risk_platforms = MagicMock(return_value=[('Bitget:default', bitget)]) agent._get_initial_balance = MagicMock(return_value=1000.0) agent._get_bitget_target_key = MagicMock(return_value='Bitget:default') should_stop, reason = asyncio.run(agent._check_account_level_stop_loss()) assert should_stop is True assert 'Bitget:default' in reason assert agent._platform_halts['Bitget:default']['halted'] is True agent._emergency_close_all_positions.assert_awaited_once() def test_resume_platform_resets_initial_balance_and_clears_halt(): agent = make_agent() bitget = MagicMock() bitget.get_account_state.return_value = { 'account_value': 888.0, 'current_balance': 888.0, } agent.bitget = bitget agent.bitget_services = {'default': bitget} agent._get_bitget_target_key = MagicMock(return_value='Bitget:default') agent._platform_halts = { 'Bitget:default': { 'halted': True, 'reason': 'drawdown', 'drawdown_pct': 25.1, } } result = agent.resume_platform('Bitget') assert result['halted'] is False assert agent._initial_balances['Bitget:default'] == 888.0 assert result['initial_balance'] == 888.0 assert result['current_balance'] == 888.0 def test_execution_events_are_recorded_and_returned_in_reverse_time_order(): agent = make_agent() agent._record_execution_event( 'Bitget', 'open_failed', symbol='ETHUSDT', reason='余额不足', status='error', decision={ 'setup_type': 'breakout_confirmation', 'setup_basis': 'setup=breakout_confirmation | breakout_quality=acceptance_breakout_up', 'entry_basis': 'breakout_acceptance_follow_through', }, ) agent._record_execution_event('PaperTrading', 'hold', symbol='BTCUSDT', reason='已有盈利反向仓', status='hold') events = agent.get_recent_execution_events(limit=10) assert len(events) == 2 assert events[0]['platform'] == 'PaperTrading' assert events[0]['event_type'] == 'hold' assert events[1]['platform'] == 'Bitget' assert events[1]['reason'] == '余额不足' assert events[1]['setup_type'] == 'breakout_confirmation' def test_get_status_contains_last_execution_preview(): agent = make_agent() agent.running = True agent.symbols = ['BTCUSDT'] agent.last_signals = { 'BTCUSDT': {'type': 'medium_term', 'action': 'sell', 'confidence': 78, 'grade': 'B'} } agent.last_execution_preview = { 'BTCUSDT': { 'timestamp': '2026-04-22T12:00:00', 'current_price': 65000.0, 'paper': {'decision': 'OPEN', 'reason': '正常开仓'}, 'bitget': {'decision': 'CANCEL_PENDING', 'reason': '替换旧挂单'}, } } status = agent.get_status() assert status['last_execution_preview']['BTCUSDT']['paper']['decision'] == 'OPEN' assert status['last_execution_preview']['BTCUSDT']['bitget']['reason'] == '替换旧挂单' def test_target_execution_status_uses_settings_defaults_until_overridden(): agent = make_agent() agent.bitget_services = {'default': MagicMock()} agent._get_bitget_target_key = MagicMock(return_value='Bitget:default') agent._iter_bitget_accounts = MagicMock(return_value=['default']) status = agent.get_target_execution_status() assert status['PaperTrading']['enabled'] is True assert status['PaperTrading']['source'] == 'default' assert status['Bitget:default']['enabled'] is True assert status['Bitget:default']['source'] == 'default' def test_set_target_execution_enabled_persists_manual_override(): agent = make_agent() agent.bitget_services = {'default': MagicMock()} agent._iter_bitget_accounts = MagicMock(return_value=['default']) agent._get_bitget_target_key = MagicMock(return_value='Bitget:default') result = agent.set_target_execution_enabled('Bitget', False, 'manual off') assert result['enabled'] is False assert result['source'] == 'manual' assert result['reason'] == 'manual off' assert agent._target_execution_controls['Bitget:default']['enabled'] is False agent._save_target_execution_controls.assert_called_once()