import json from datetime import datetime as dt from py_jftech import component from api import DataLogger, LoggerType, PortfoliosRisk, Cleanable, BacktestStep from reports.dao import robo_data_logger as rdl @component(bean_name='data-logger') class DatabaseLogger(DataLogger, Cleanable): def save_record(self, date: dt, risk: PortfoliosRisk, type: LoggerType, datas: dict, exist_merge=True): assert date is not None, "save record, date cannot be null" assert risk is not None, "save record, risk cannot be null" assert type is not None, "save record, type cannot be null" assert datas is not None, "save record, dates cannot be null" exist = rdl.get_one(date=date, risk=risk, type=type) if exist: save_datas = datas if exist_merge: save_datas = {**json.loads(exist['datas']), **datas} rdl.update(exist['id'], save_datas) else: rdl.insert({ 'date': date, 'risk': risk, 'type': type, 'datas': datas }) def load_records(self, max_date=None, min_date=None, risk: PortfoliosRisk = None, type: LoggerType = None): result = rdl.get_list(max_date=max_date, min_date=min_date, risk=risk, type=type, like_type=True) return [{**x, 'datas': json.loads(x['datas'])} for x in result] def clean_up(self, min_date=None, risk: PortfoliosRisk = None): rdl.delete(min_date=min_date, risk=risk) @property def clean_step(self): return BacktestStep.HOLD_PORTFOLIO @property def clean_name(self): return 'data logger'