from datetime import datetime as dt from typing import List import pandas as pd from dateutil.relativedelta import relativedelta from py_jftech import component, autowired, get_config, format_date, filter_weekend from api import RoboReportor @component(bean_name='relative-range-report') class RelativeRangeReport(RoboReportor): @autowired(names={'combo': 'combo-report'}) def __init__(self, combo: RoboReportor = None): self._combo = combo self._config = get_config(__name__) @property def report_name(self) -> str: return '相对区间收益率' @property def range_dates(self): return self._config['range-dates'] def load_report(self, max_date=dt.today(), min_date=None) -> List[dict]: datas = pd.DataFrame(self._combo.load_report(max_date=max_date, min_date=min_date)) if not datas.empty: datas.set_index('date', inplace=True) result = pd.DataFrame(columns=datas.columns) for range in self.range_dates: kwargs = range.copy() del kwargs['name'] start = filter_weekend(max_date - relativedelta(**kwargs)) if kwargs and ('dates' not in kwargs or kwargs['dates'] is not None) else datas.index[0] end = filter_weekend(max_date) row_name = f"{range['name']}({format_date(start)}~{format_date(end)})" result.loc[row_name] = datas[start:end].values[-1] / datas[start:end].values[0] - 1 result = round(result, 4) * 100 result.reset_index(inplace=True) result.rename(columns={'index': 'range-date'}, inplace=True) return result.to_dict('records') return []