import os from abc import abstractmethod from copy import deepcopy from datetime import datetime as dt from shutil import copyfile from tempfile import TemporaryDirectory from typing import List import pandas as pd from dateutil.relativedelta import relativedelta from py_jftech import component, autowired, get_config, get_instance_name, get_project_path, format_date, sendmail from api import RoboReportor, RoboExportor, RoboExecutor def include_report(): return get_config(__name__)['include-report'] class DefaultExportor(RoboExportor): @autowired def __init__(self, reportors: List[RoboReportor] = None, exec: RoboExecutor = None): self._reportors = {get_instance_name(x): x for x in reportors} self._exec = exec def export(self, max_date=dt.today(), min_date=None): if not self.include_report: return None with TemporaryDirectory() as tmpdir: filename = f"{self.file_name}_{format_date(self._exec.curt_date)}" filepath = os.path.join(tmpdir, f"{filename}.xlsx") with pd.ExcelWriter(filepath) as writer: for reportor_name in self.include_report: mindate = min_date if isinstance(reportor_name, dict): reportor = self._reportors[reportor_name['name']] if reportor_name['min-date'] is None: mindate = None elif isinstance(reportor_name['min-date'], dict): mindate = max_date - relativedelta(**reportor_name['min-date']) datas = pd.DataFrame(reportor.load_report(max_date=max_date, min_date=mindate)) else: reportor = self._reportors[reportor_name] datas = pd.DataFrame(reportor.load_report(max_date=max_date, min_date=mindate)) sheet_name = reportor.report_name if mindate and mindate > self._exec.start_date: sheet_name = f'{sheet_name}(近{(max_date-mindate).days}天)' if not datas.empty: datas.to_excel(writer, sheet_name=sheet_name, index=False) email = self.get_email(filepath) if email and 'receives' in email and email['receives']: receives = email['receives'] copies = email['copies'] if 'copies' in email and email['copies'] is not None else [] attach_paths = [filepath] subject = email['subject'].format(today=format_date(dt.today())) content = email['content'].format(today=format_date(dt.today())) sendmail(receives=receives, copies=copies, attach_paths=attach_paths, subject=subject, content=content) if self.save_path is not None: os.makedirs(self.save_path, exist_ok=True) save_file = os.path.join(self.save_path, f"{filename}.xlsx") copyfile(filepath, save_file) if self.save_config: profile_active = os.environ.get('PROFILE_ACTIVE') config_name = f'config-{profile_active}.yml' if profile_active is not None else 'config.yml' src_path = f'{get_project_path()}{os.path.sep}{config_name}' save_path = os.path.join(self.save_path, f"{filename}.yml") copyfile(src_path, save_path) def get_email(self, file): return deepcopy(self.config['email']) if 'email' in self.config else None @property def save_path(self): if 'save-path' not in self.config: return None save_path: str = self.config['save-path'] if save_path.startswith('.'): return os.path.abspath(os.path.join(os.path.dirname(__file__), save_path)) elif save_path.startswith('/'): return os.path.abspath(save_path) return os.path.abspath(os.path.join(get_project_path(), save_path)) @property def exist_build(self): return self.config['exist-build'] if 'exist-build' in self.config else False @property def file_name(self): return self.config['file-name'] if 'file-name' in self.config else 'export' @property def include_report(self): return self.config['include-report'] if 'include-report' in self.config else [] @property def save_config(self): return self.config['save-config'] if 'save-config' in self.config else False @property @abstractmethod def config(self): pass @component(bean_name='backtest-export') class BacktestExportor(DefaultExportor): def __init__(self): super(BacktestExportor, self).__init__() self.__config = deepcopy(get_config(__name__)) @property def config(self): return self.__config['backtest'] @component(bean_name='daily-real-export') class DailyRealExportor(DefaultExportor): @autowired(names={'signal_reportor': 'daily-signal-report'}) def __init__(self, signal_reportor: RoboReportor = None): super(DailyRealExportor, self).__init__() self.__config = deepcopy(get_config(__name__)) self._signal_reportor = signal_reportor def get_email(self, file): result = super(DailyRealExportor, self).get_email(file) if result is None: return None content = pd.read_excel(file, sheet_name=None) if self._signal_reportor.report_name in content: result['subject'] = str(result['subject']['rebalance']) result['content'] = result['content']['rebalance'] else: result['subject'] = result['subject']['default'] result['content'] = result['content']['rebalance'] return result @property def config(self): return self.__config['real-daily'] @component(bean_name='daily-monitor-export') class DailyMonitorExportor(DefaultExportor): def __init__(self): super(DailyMonitorExportor, self).__init__() self.__config = deepcopy(get_config(__name__)) @property def config(self): return self.__config['daily-monitor']