Coverage for api/v1/endpoints/strategies.py: 58.70%

1""" 

2策略管理API端点 

3""" 

4 

5from typing import Any, Dict, List 

6 

7from fastapi import APIRouter, Depends, HTTPException 

8from pydantic import BaseModel 

9 

10from core.middleware.auth_middleware import get_current_user 

11from core.models.user import User 

12from core.trading.strategies.strategy_factory import ( 

13 get_available_strategies, get_strategy_parameters, 

14 get_strategy_recommended_risk_config, get_strategy_template, 

15 validate_strategy_config) 

16 

17router = APIRouter() 

18 

19 

20class StrategyConfigRequest(BaseModel): 

21 """策略配置请求""" 

22 

23 strategy_name: str 

24 config: Dict[str, Any] 

25 

26 

27class StrategyConfigResponse(BaseModel): 

28 """策略配置响应""" 

29 

30 valid: bool 

31 errors: List[str] = [] 

32 warnings: List[str] = [] 

33 config: Dict[str, Any] = {} 

34 

35 

36@router.get("/", response_model=Dict[str, Any]) 

37async def get_available_strategies_list(current_user: User = Depends(get_current_user)): 

38 """获取可用策略列表""" 

39 return get_available_strategies() 

40 

41 

42@router.get("/{strategy_name}") 

43async def get_strategy_info( 

44 strategy_name: str, current_user: User = Depends(get_current_user) 

45): 

46 """获取策略详情""" 

47 available_strategies = get_available_strategies() 

48 

49 if strategy_name not in available_strategies: 

50 raise HTTPException(status_code=404, detail="策略不存在") 

51 

52 return available_strategies[strategy_name] 

53 

54 

55@router.post("/validate", response_model=StrategyConfigResponse) 

56async def validate_strategy_configuration( 

57 request: StrategyConfigRequest, current_user: User = Depends(get_current_user) 

58): 

59 """验证策略配置""" 

60 result = validate_strategy_config(request.strategy_name, request.config) 

61 

62 return StrategyConfigResponse( 

63 valid=result["valid"], 

64 errors=result["errors"], 

65 warnings=result["warnings"], 

66 config=result.get("config", {}), 

67 ) 

68 

69 

70@router.get("/{strategy_name}/risk-config") 

71async def get_strategy_recommended_risk_configuration( 

72 strategy_name: str, current_user: User = Depends(get_current_user) 

73): 

74 """获取策略推荐的风险配置""" 

75 risk_config = get_strategy_recommended_risk_config(strategy_name) 

76 

77 if not risk_config: 

78 raise HTTPException(status_code=404, detail="策略不存在") 

79 

80 return {"strategy_name": strategy_name, "recommended_risk_config": risk_config} 

81 

82 

83@router.get("/{strategy_name}/parameters") 

84async def get_strategy_parameters_list( 

85 strategy_name: str, current_user: User = Depends(get_current_user) 

86): 

87 """获取策略参数定义""" 

88 parameters = get_strategy_parameters(strategy_name) 

89 

90 if not parameters: 

91 raise HTTPException(status_code=404, detail="策略不存在") 

92 

93 return {"strategy_name": strategy_name, "parameters": parameters} 

94 

95 

96@router.get("/{strategy_name}/template") 

97async def get_strategy_template_info( 

98 strategy_name: str, current_user: User = Depends(get_current_user) 

99): 

100 """获取策略配置模板""" 

101 template = get_strategy_template(strategy_name) 

102 

103 if not template: 

104 raise HTTPException(status_code=404, detail="策略不存在") 

105 

106 return template