from typing import Any from config.strategy_state import state from kit.conf import conf from kit.logger import Logger from config.network_config import network_config class strategy_config: def __init__(self, strategy_name: str, strategy_template_id: str, config_save: network_config, log): """ 基础必要初始化策略配置 """ # 策略名 self.strategy_name = strategy_name # 配置模版id self.template_id = strategy_template_id # 配置变更保存方式 self.connection = config_save # 日志记录器 self.logger = Logger(log) """ 实例化strategy_config类时,初始化必要的属性 """ # 策略ID self.strategy_id = None # 配置对应主键 self.configkey_id = {} # 配置信息 self.config = {} # 核心缓存 self.strategy_state = state() # 初始化配置文件 self.init_from_db() def init_from_db(self): self.strategy_id = self.connection.register_strategy(self.strategy_name, self.template_id) self.logger.info("初始化策略配置", self.strategy_id) self.logger.info(' 注册策略序列:' + str(self.strategy_id), self.strategy_id) config_list = self.connection.get_data_list(self.strategy_id, 'config') self.config = {item['name']: item for item in config_list} self.configkey_id = {item['name']: item['id'] for item in config_list} self.logger.info('确认配置参数:', self.strategy_id) for config in config_list: self.logger.info(f" 参数名: {config['name']}, 值: {config['new_value']}, 备注: {config['memo']}", self.strategy_id) self.logger.info('确认配置参数完毕, 如有错误请立刻停止执行', self.strategy_id) def get_all_configs(self) -> dict: """ 获取所有策略配置参数 :return: 包含所有配置参数的字典 """ return self.config def set_config(self, key: str, value: Any) -> None: """ 通过字典语法设置配置项,并自动触发数据库更新 Args: key (str): 配置项名称 value (Any): 配置项的新值 """ id = self.configkey_id.get(key) self.connection.update_config(id, value) self.config[key]['new_value'] = value def get_config(self, key: str): func = conf.config_type(key) return func(key) # if __name__ == "__main__": # config_instance = strategy_config('aaa', '10000000', network_config(), None)