from datetime import datetime as dt from typing import List import pandas as pd from py_jftech import component, autowired from api import RoboReportor @component(bean_name='combo-report') class DivAlligamComboDatasReportor(RoboReportor): @autowired(names={'hold_reportor': 'hold-report', 'benchmark': 'benckmark-report'}) def __init__(self, hold_reportor: RoboReportor = None, benchmark: RoboReportor = None): self._hold_reportor = hold_reportor self._benchmark = benchmark @property def report_name(self) -> str: return '对比报告' def load_report(self, max_date=dt.today(), min_date=None) -> List[dict]: holds = pd.DataFrame(self._hold_reportor.load_report(max_date=max_date, min_date=min_date)) if not holds.empty: holds.set_index('date', inplace=True) holds = holds[['real_av', 'acc_av', 'nav', 'fund_nav']] holds.rename(columns={'real_av': 'av', 'acc_av': 'acc'}, inplace=True) benchmark = pd.DataFrame(self._benchmark.load_report(max_date=max_date, min_date=min_date)) benchmark.set_index('date', inplace=True) benchmark = benchmark[['alligam_av', 'alligam_acc', 'alligam_nav']] datas = holds.join(benchmark) datas.fillna(method='ffill', inplace=True) datas.dropna(inplace=True) datas.reset_index(inplace=True) return datas.to_dict('records') return []