mirror of
https://github.com/silencesdg/mt5_python_ea_suite.git
synced 2026-08-12 18:28:09 +00:00
重构项目架构,新增 MT5 代理服务
- 重构核心模块:DataProvider 依赖注入、RiskController 门面、信号注册表 - 新增 FastAPI 代理服务 (run/server.py),支持局域网远程调用 MT5 - 新增 RemoteDataProvider + AttrDict,远端无缝替代 LiveDataProvider - 新增序列化模块,MT5 对象转 JSON 兼容格式 - 重构入口点至 run/ 包,支持 python -m run.realtime/server/backtest/optimize - 更新 CLAUDE.md 文档 Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude <noreply@anthropic.com> Co-Authored-By: Happy <yesreply@happy.engineering>
This commit is contained in:
co-authored by
Claude
Happy
parent
e8226edd96
commit
4cb4f4a15e
@@ -0,0 +1,70 @@
|
||||
"""策略注册表 — 单例,消除代码中多处的策略列表重复
|
||||
|
||||
用法:
|
||||
registry = StrategyRegistry()
|
||||
registry.register('ma_cross', MACrossStrategy)
|
||||
registry.instantiate_all(SYMBOL, TIMEFRAME) # 或传入参数字典
|
||||
config_key = registry.class_name_to_config_key('MACrossStrategy')
|
||||
"""
|
||||
|
||||
from config import STRATEGY_CONFIG
|
||||
|
||||
|
||||
class StrategyRegistry:
|
||||
"""策略注册表单例 — 所有策略信息的唯一数据源"""
|
||||
|
||||
_instance = None
|
||||
|
||||
def __new__(cls):
|
||||
if cls._instance is None:
|
||||
cls._instance = super().__new__(cls)
|
||||
cls._instance._strategies = {} # config_key → StrategyClass
|
||||
cls._instance._instances = {} # config_key → instance
|
||||
return cls._instance
|
||||
|
||||
def register(self, config_key: str, strategy_class: type) -> None:
|
||||
self._strategies[config_key] = strategy_class
|
||||
|
||||
def instantiate_all(self, symbol: str, timeframe: int,
|
||||
params_dict: dict = None,
|
||||
data_provider=None) -> dict:
|
||||
"""实例化所有已注册策略
|
||||
|
||||
Args:
|
||||
data_provider: 回测传None(run_backtest不需要),实盘传DataProvider实例
|
||||
Returns:
|
||||
{config_key: strategy_instance}
|
||||
"""
|
||||
if params_dict is None:
|
||||
params_dict = STRATEGY_CONFIG
|
||||
|
||||
instances = {}
|
||||
for key, cls in self._strategies.items():
|
||||
params = params_dict.get(key, {})
|
||||
instances[key] = cls(data_provider, symbol, timeframe, **params)
|
||||
self._instances = instances
|
||||
return instances
|
||||
|
||||
@property
|
||||
def instances(self) -> dict:
|
||||
return self._instances
|
||||
|
||||
@property
|
||||
def all_strategies(self) -> dict:
|
||||
"""返回 {config_key: StrategyClass}"""
|
||||
return dict(self._strategies)
|
||||
|
||||
def class_name_to_config_key(self, class_name: str) -> str | None:
|
||||
for key, cls in self._strategies.items():
|
||||
if cls.__name__ == class_name:
|
||||
return key
|
||||
return None
|
||||
|
||||
def config_key_to_class_name(self, config_key: str) -> str:
|
||||
return self._strategies[config_key].__name__
|
||||
|
||||
def get_instance(self, config_key: str):
|
||||
return self._instances.get(config_key)
|
||||
|
||||
def __contains__(self, config_key: str) -> bool:
|
||||
return config_key in self._strategies
|
||||
Reference in New Issue
Block a user