from datetime import datetime as dt from typing import List import pandas as pd from empyrical import annual_return, annual_volatility, max_drawdown, sharpe_ratio from py_jftech import component, autowired from api import RoboReportor @component(bean_name='indicators-report') class IndicatorsReportor(RoboReportor): @autowired(names={'combo': 'combo-report'}) def __init__(self, combo: RoboReportor = None): self._combo = combo @property def report_name(self) -> str: return '指标' 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) returns = round(datas.pct_change(), 5) indicators = { 'annual_return': list(annual_return(returns, period='daily', annualization=None) * 100), 'annual_volatility': annual_volatility(returns, period='daily', annualization=None) * 100, 'max_drawdown': max_drawdown(returns, out=None) * 100, 'sharp': sharpe_ratio(returns, risk_free=0, period='daily', annualization=None), } indicators['calmar'] = abs(indicators['annual_return'] / indicators['max_drawdown']) result = pd.DataFrame(indicators.values(), index=indicators.keys(), columns=list(returns.columns)).round(2) result.reset_index(inplace=True) result.rename(columns={'index': 'indicators'}, inplace=True) return result.to_dict('records') return []