依赖注入实现中

api.py

-from abc import ABCMeta, abstractmethod
+from abc import ABC, abstractmethod
 from enum import Enum, unique
 
 
@@ -15,7 +15,7 @@ class BusinessException(Exception):
         return self.__msg
 
 
-class Datum(metaclass=ABCMeta):
+class Datum(ABC):
     '''
     基础资料服务，基金资料数据，各种指数，指标资料数据
     '''
@@ -31,7 +31,7 @@ class Datum(metaclass=ABCMeta):
         pass
 
 
-class Navs(metaclass=ABCMeta):
+class Navs(ABC):
     '''
     基础数据相关服务，基金净值，各种指标 高开低收
     '''
@@ -47,25 +47,37 @@ class Navs(metaclass=ABCMeta):
         pass
 
 
-class FundOptimize(metaclass=ABCMeta):
+class Optimize(ABC):
     '''
-    基金优选相关服务
+    优选相关服务ABC
     '''
     @abstractmethod
-    def find_optimize(self, fund_ids, day):
+    def find_optimize(self, ids, day):
         '''
-        从多个基金id中，选出指定日期最优的基金id
-        :param fund_ids: 待选基金id列表
+        从多id中，选出指定日期最优的id
+        :param ids: 待选id列表
         :param day: 指定日期
-        :return: 最优的基金id
+        :return: 最优的id
         '''
         pass
 
     @abstractmethod
     def get_optimize_pool(self, day):
         '''
-        根据基金优选获取指定日期的基金池
+        根据优选规则获取指定日期的优选池
         :param day: 指定日期
-        :return: 基金id列表
+        :return: 优选id列表
         '''
         pass
+
+
+class EwmaCvar(metaclass=ABCMeta):
+    '''
+    ewma相关服务
+    '''
+
+    def build_ewma_cvar(self):
+        pass
+
+
+

fund_pool/fund_optimize.py

 import json
-
 import pandas as pd
+from abc import ABC, abstractmethod
 from dateutil.relativedelta import relativedelta
 from empyrical import sortino_ratio
 from framework import filter_weekend, dict_remove, get_config, component, autowired
-from api import FundOptimize, Navs, BusinessException, Datum
+from api import Optimize, Navs, BusinessException, Datum
 from fund_pool.dao import robo_optimize_pool as rop
 
 
-@component
-class SortinoFundOptimize(FundOptimize):
-
-    _navs: Navs = None
-    _datum: Datum = None
+class SortinoOptimize(Optimize, ABC):
 
-    @autowired
-    def __init__(self, navs: Navs = None, datum: Datum = None):
-        self._navs = navs
-        self._datum = datum
+    def __init__(self):
         optimize_config = get_config(__name__)
         self._config = [{
             **x,
@@ -27,35 +20,78 @@ class SortinoFundOptimize(FundOptimize):
     def find_optimize(self, fund_ids, day):
         if not self._config:
             raise BusinessException(f"find optimize, but not found sortino config.")
-        start = filter_weekend(sorted([day - relativedelta(days=1, **dict_remove(x, ('weight', 'name'))) for x in self._config])[0])
-        fund_navs = pd.DataFrame(self._navs.get_fund_navs(fund_ids=tuple(fund_ids), min_date=start, max_date=day))
-        fund_navs.sort_values('nav_date', inplace=True)
-        fund_navs = fund_navs.pivot_table(index='nav_date', columns='fund_id', values='nav_cal')
-        fund_navs.fillna(method='ffill', inplace=True)
-        day_income = round(fund_navs.pct_change(), 4)
+        pct_change = pd.DataFrame(self.get_pct_change(fund_ids, day))
+        pct_change.set_index('date', inplace=True)
         sortino = pd.DataFrame()
         for item in self._config:
             delta_kwargs = item.copy()
             del delta_kwargs['weight'], delta_kwargs['name']
-            ratio = dict(sortino_ratio(day_income.truncate(before=(day - relativedelta(**delta_kwargs)))))
+            ratio = dict(sortino_ratio(pct_change.truncate(before=(day - relativedelta(**delta_kwargs)))))
             sortino = pd.concat([sortino, pd.DataFrame([ratio], index=[item['name']])])
         sortino = sortino.T
         sortino['score'] = sortino.apply(lambda r: sum([x['weight'] * r[x['name']] for x in self._config]), axis=1)
         sortino.sort_values('score', ascending=False, inplace=True)
-        return day_income.columns[sortino.index[0]]
+        return pct_change.columns[sortino.index[0]]
 
     def get_optimize_pool(self, day):
         last_one = rop.get_last_one(day)
         if not last_one:
-            funds = pd.DataFrame(self._datum.get_fund_datums())
             pool = []
-            for (category, asset_type), fund_group in funds.groupby(by=['category', 'assetType']):
+            for fund_group in self.get_groups():
                 if len(fund_group) > 1:
-                    pool.append(self.find_optimize(tuple(fund_group['id']), day))
+                    pool.append(self.find_optimize(fund_group, day))
                 else:
-                    pool.append(fund_group.iloc[0].id)
+                    pool.append(fund_group[0])
             rop.insert(day, sorted(pool))
             last_one = rop.get_last_one(day)
         return json.loads(last_one['fund_ids'])
 
+    @abstractmethod
+    def get_groups(self):
+        '''
+        :return: 返回待处理的id数组
+        '''
+        pass
+
+    @abstractmethod
+    def get_pct_change(self, fund_ids, day):
+        '''
+        根据id数组，返回指定日期的收益率
+        :param fund_ids: id数组
+        :param day: 指定的日期
+        :return: 收益率
+        '''
+        pass
 
+
+@component
+class FundSortinoOptimize(SortinoOptimize):
+    '''
+    根据索提诺比率计算基金优选的优选实现
+    '''
+
+    @autowired
+    def __init__(self, navs: Navs = None, datum: Datum = None):
+        super().__init__()
+        self._navs = navs
+        self._datum = datum
+
+    def get_groups(self):
+        funds = pd.DataFrame(self._datum.get_fund_datums())
+        result = []
+        for (category, asset_type), fund_group in funds.groupby(by=['category', 'assetType']):
+            result.append(tuple(fund_group['id']))
+        return result
+
+    def get_pct_change(self, fund_ids, day):
+        if not self._config:
+            raise BusinessException(f"find optimize, but not found sortino config.")
+        start = filter_weekend(sorted([day - relativedelta(days=1, **dict_remove(x, ('weight', 'name'))) for x in self._config])[0])
+        fund_navs = pd.DataFrame(self._navs.get_fund_navs(fund_ids=tuple(fund_ids), min_date=start, max_date=day))
+        fund_navs.sort_values('nav_date', inplace=True)
+        fund_navs = fund_navs.pivot_table(index='nav_date', columns='fund_id', values='nav_cal')
+        fund_navs.fillna(method='ffill', inplace=True)
+        result = round(fund_navs.pct_change().dropna(), 4)
+        result.reset_index(inplace=True)
+        result.rename(columns={'nav_date': 'date'}, inplace=True)
+        return result.to_dict('records')

main.py

 from framework import autowired, parse_date, logger
-from api import FundOptimize
+from api import Optimize
 
 
 @autowired
-def start(optimize: FundOptimize = None):
+def start(optimize: Optimize = None):
     pool = optimize.get_optimize_pool(parse_date('2022-11-07'))
     logger.info(pool)