82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
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)
|