#!/usr/bin/env python3 """ P0问题修复验证测试 测试3个严重问题的修复情况 """ import sys import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import asyncio import json from pathlib import Path from datetime import datetime from typing import Dict, Any class TestP0Fixes: """P0问题修复测试类""" def __init__(self): self.test_results = [] self.passed = 0 self.failed = 0 def log(self, test_name: str, passed: bool, message: str): """记录测试结果""" status = "✅ PASS" if passed else "❌ FAIL" result = f"{status} - {test_name}: {message}" self.test_results.append(result) if passed: self.passed += 1 else: self.failed += 1 print(result) def test_1_max_margin_pct(self): """测试1: 配置不一致修复（max_margin_pct 应该是 25%）""" print("\n" + "="*60) print("测试1: max_margin_pct 配置（应该支持25%）") print("="*60) try: from app.crypto_agent.crypto_agent import CryptoAgent agent = CryptoAgent() # 检查 Bitget bitget_max = agent.PLATFORM_RULES['Bitget']['max_margin_pct'] self.log( "Bitget max_margin_pct", bitget_max == 0.25, f"期望 0.25, 实际 {bitget_max}" ) # 检查 PaperTrading paper_max = agent.PLATFORM_RULES['PaperTrading']['max_margin_pct'] self.log( "PaperTrading max_margin_pct", paper_max == 0.25, f"期望 0.25, 实际 {paper_max}" ) # 检查 Hyperliquid hyper_max = agent.PLATFORM_RULES['Hyperliquid']['max_margin_pct'] self.log( "Hyperliquid max_margin_pct", hyper_max == 0.25, f"期望 0.25, 实际 {hyper_max}" ) except Exception as e: self.log("max_margin_pct 测试", False, f"异常: {e}") def test_2_position_sizing(self): """测试2: A级信号应该能使用20%保证金（不被截断到10%）""" print("\n" + "="*60) print("测试2: A级信号仓位计算（$1000账户应该能用$200保证金）") print("="*60) try: from app.crypto_agent.crypto_agent import CryptoAgent agent = CryptoAgent() # 模拟A级信号 signal = { 'symbol': 'BTCUSDT', 'confidence': 92, # A级 } # 模拟账户 account = { 'available': 1000.0, 'current_balance': 1000.0, 'current_total_leverage': 0, 'max_total_leverage': 10, } # 计算仓位 margin, reason = agent._calculate_position_size(signal, account, 'PaperTrading') # A级信号期望 20% = $200 expected_min = 190 # 考虑一些边界情况，至少应该接近$200 expected_max = 210 is_correct = expected_min <= margin <= expected_max self.log( "A级信号保证金计算", is_correct, f"期望 ${expected_min}-${expected_max}, 实际 ${margin:.2f} ({reason})" ) except Exception as e: self.log("仓位计算测试", False, f"异常: {e}") def test_3_initial_balance_persistence(self): """测试3: 初始余额持久化机制""" print("\n" + "="*60) print("测试3: 初始余额持久化（应该保存到data/initial_balances.json）") print("="*60) try: from app.crypto_agent.crypto_agent import CryptoAgent # 清理测试文件 test_file = Path("data/initial_balances.json") if test_file.exists(): test_file.unlink() # 创建新的agent实例 agent = CryptoAgent() # 第一次获取（应该记录） platform_name = "TestPlatform" current_balance = 10000.0 initial_1 = agent._get_initial_balance(platform_name, current_balance) self.log( "首次获取初始余额", initial_1 == current_balance, f"期望 {current_balance}, 实际 {initial_1}" ) # 检查文件是否创建 self.log( "持久化文件创建", test_file.exists(), f"文件存在: {test_file.exists()}" ) # 检查文件内容 with open(test_file, 'r') as f: saved_data = json.load(f) self.log( "持久化内容正确", saved_data.get(platform_name) == current_balance, f"文件中 {platform_name}: {saved_data.get(platform_name)}" ) # 模拟余额变化，再次获取（应该返回初始值） current_balance_2 = 9000.0 # 亏损10% initial_2 = agent._get_initial_balance(platform_name, current_balance_2) self.log( "再次获取初始余额（余额变化后）", initial_2 == current_balance, # 仍然是初始值 f"期望 {current_balance}, 实际 {initial_2}" ) # 计算回撤 drawdown = (initial_2 - current_balance_2) / initial_2 expected_drawdown = 0.10 # 10% self.log( "回撤计算正确", abs(drawdown - expected_drawdown) < 0.001, f"期望 {expected_drawdown*100:.1f}%, 实际 {drawdown*100:.1f}%" ) except Exception as e: self.log("初始余额持久化测试", False, f"异常: {e}") import traceback traceback.print_exc() async def test_4_emergency_close_await(self): """测试4: 紧急平仓async/await处理""" print("\n" + "="*60) print("测试4: 紧急平仓async/await检测") print("="*60) try: import asyncio # Mock同步方法 def sync_close(symbol: str): return {'success': True, 'symbol': symbol} # Mock异步方法 async def async_close(symbol: str): await asyncio.sleep(0.01) return {'success': True, 'symbol': symbol} # 测试asyncio.iscoroutinefunction is_sync = asyncio.iscoroutinefunction(sync_close) is_async = asyncio.iscoroutinefunction(async_close) self.log( "同步方法检测", not is_sync, f"sync_close 是协程: {is_sync} (应该是False)" ) self.log( "异步方法检测", is_async, f"async_close 是协程: {is_async} (应该是True)" ) # 测试调用 try: # 同步调用 result_sync = sync_close("BTC") self.log( "同步方法调用", result_sync['success'], f"结果: {result_sync}" ) # 异步调用 result_async = await async_close("ETH") self.log( "异步方法调用", result_async['success'], f"结果: {result_async}" ) except Exception as e: self.log("方法调用测试", False, f"异常: {e}") except Exception as e: self.log("紧急平仓await测试", False, f"异常: {e}") def test_5_account_drawdown_calculation(self): """测试5: 账户回撤计算""" print("\n" + "="*60) print("测试5: 账户回撤计算（模拟各种场景）") print("="*60) try: from app.crypto_agent.crypto_agent import CryptoAgent # 测试场景 scenarios = [ {"initial": 10000, "current": 10000, "expected_dd": 0.0, "desc": "无回撤"}, {"initial": 10000, "current": 8500, "expected_dd": 0.15, "desc": "15%回撤(警告)"}, {"initial": 10000, "current": 7500, "expected_dd": 0.25, "desc": "25%回撤(止损)"}, {"initial": 10000, "current": 5000, "expected_dd": 0.50, "desc": "50%回撤(严重)"}, ] agent = CryptoAgent() for scenario in scenarios: initial = scenario['initial'] current = scenario['current'] expected = scenario['expected_dd'] # 模拟计算 drawdown = (initial - current) / initial is_correct = abs(drawdown - expected) < 0.001 self.log( f"回撤计算: {scenario['desc']}", is_correct, f"期望 {expected*100:.1f}%, 实际 {drawdown*100:.1f}%" ) except Exception as e: self.log("账户回撤计算测试", False, f"异常: {e}") def print_summary(self): """打印测试总结""" print("\n" + "="*60) print("📊 测试总结") print("="*60) total = self.passed + self.failed pass_rate = (self.passed / total * 100) if total > 0 else 0 print(f"总测试数: {total}") print(f"通过: {self.passed} ✅") print(f"失败: {self.failed} ❌") print(f"通过率: {pass_rate:.1f}%") if self.failed == 0: print("\n🎉 所有测试通过！P0问题修复验证成功！") else: print(f"\n⚠️ 有 {self.failed} 个测试失败，请检查！") print("="*60 + "\n") async def run_all_tests(self): """运行所有测试""" print("\n" + "🚀 " * 30) print("P0问题修复验证测试") print("测试时间:", datetime.now().strftime("%Y-%m-%d %H:%M:%S")) print("🚀 " * 30) # 运行测试 self.test_1_max_margin_pct() self.test_2_position_sizing() self.test_3_initial_balance_persistence() await self.test_4_emergency_close_await() self.test_5_account_drawdown_calculation() # 打印总结 self.print_summary() return self.failed == 0 def main(): """主函数""" test = TestP0Fixes() # 运行异步测试 success = asyncio.run(test.run_all_tests()) # 返回退出码 sys.exit(0 if success else 1) if __name__ == "__main__": main()