Coverage for core/trading/strategies/strategy_factory.py: 48.72%

1""" 

2策略工厂 

3""" 

4 

5from typing import Any, Dict, List, Optional 

6 

7from core.models.strategy_config import StrategyConfigPreset 

8 

9from .base_strategy import BaseStrategy 

10from .bollinger_strategy import BollingerBandsStrategy 

11from .ma_crossover_strategy import MovingAverageCrossoverStrategy 

12from .macd_strategy import MACDStrategy 

13from .rsi_strategy import RSIStrategy 

14 

15 

16def create_strategy( 

17 strategy_name: str, config: Dict[str, Any] 

18) -> Optional[BaseStrategy]: 

19 """创建策略工厂函数""" 

20 

21 strategy_mapping = { 

22 "ma_crossover": MovingAverageCrossoverStrategy, 

23 "rsi": RSIStrategy, 

24 "macd": MACDStrategy, 

25 "bollinger": BollingerBandsStrategy, 

26 "bollinger_bands": BollingerBandsStrategy, # 别名 

27 } 

28 

29 strategy_class = strategy_mapping.get(strategy_name.lower()) 

30 

31 if strategy_class: 

32 try: 

33 return strategy_class(strategy_name, config) 

34 except Exception as e: 

35 # 策略工厂暂时使用print，因为此时还没有StrategyEngine 

36 print(f"❌ 创建策略失败: {strategy_name} - {e}") 

37 return None 

38 else: 

39 # 策略工厂暂时使用print，因为此时还没有StrategyEngine 

40 print(f"❌ 未知策略类型: {strategy_name}") 

41 print(f" 支持的策略类型: {list(strategy_mapping.keys())}") 

42 return None 

43 

44 

45def get_available_strategies() -> Dict[str, Dict[str, Any]]: 

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

47 templates = StrategyConfigPreset.get_strategy_templates() 

48 

49 result = {} 

50 for strategy_name, template in templates.items(): 

51 result[strategy_name] = { 

52 "name": template.display_name, 

53 "description": template.description, 

54 "version": template.version, 

55 "author": template.author, 

56 "parameters": { 

57 param.name: { 

58 "type": param.type.value, 

59 "default": param.default_value, 

60 "description": param.description, 

61 "min_value": param.min_value, 

62 "max_value": param.max_value, 

63 "allowed_values": param.allowed_values, 

64 "required": param.required, 

65 "category": param.category, 

66 } 

67 for param in template.parameters 

68 }, 

69 "recommended_risk_config": template.recommended_risk_config, 

70 "supports_realtime": template.supports_realtime, 

71 "supports_backtest": template.supports_backtest, 

72 "min_capital": float(template.min_capital), 

73 "recommended_capital": float(template.recommended_capital), 

74 "supported_markets": template.supported_markets, 

75 "supported_symbols": template.supported_symbols, 

76 } 

77 

78 return result 

79 

80 

81def validate_strategy_config( 

82 strategy_name: str, config: Dict[str, Any] 

83) -> Dict[str, Any]: 

84 """验证策略配置""" 

85 template = StrategyConfigPreset.get_strategy_template(strategy_name) 

86 

87 if not template: 

88 return { 

89 "valid": False, 

90 "errors": [f"未知策略类型: {strategy_name}"], 

91 "warnings": [], 

92 "config": {}, 

93 } 

94 

95 return template.validate_config(config) 

96 

97 

98def get_strategy_recommended_risk_config(strategy_name: str) -> Dict[str, Any]: 

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

100 return StrategyConfigPreset.get_recommended_risk_config(strategy_name) 

101 

102 

103def get_strategy_parameters(strategy_name: str) -> List[Dict[str, Any]]: 

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

105 return StrategyConfigPreset.get_strategy_parameters(strategy_name) 

106 

107 

108def get_strategy_template(strategy_name: str) -> Optional[Dict[str, Any]]: 

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

110 template = StrategyConfigPreset.get_strategy_template(strategy_name) 

111 if template: 

112 return { 

113 "strategy_name": template.strategy_name, 

114 "display_name": template.display_name, 

115 "description": template.description, 

116 "version": template.version, 

117 "author": template.author, 

118 "parameters": [ 

119 { 

120 "name": param.name, 

121 "type": param.type.value, 

122 "default_value": param.default_value, 

123 "description": param.description, 

124 "min_value": param.min_value, 

125 "max_value": param.max_value, 

126 "allowed_values": param.allowed_values, 

127 "required": param.required, 

128 "category": param.category, 

129 } 

130 for param in template.parameters 

131 ], 

132 "recommended_risk_config": template.recommended_risk_config, 

133 "supports_realtime": template.supports_realtime, 

134 "supports_backtest": template.supports_backtest, 

135 "min_capital": float(template.min_capital), 

136 "recommended_capital": float(template.recommended_capital), 

137 "supported_markets": template.supported_markets, 

138 "supported_symbols": template.supported_symbols, 

139 } 

140 return None