新增投组校验，防止投组全为ft的情况

ai/EstimateMarketTrendV20.py

@@ -2,6 +2,7 @@ from typing import List
 
 from py_jftech import autowired
 
+from ai.dao.robo_datas import get_base_info
 from ai.data_access import DataAccess
 from ai.model_trainer import ModelTrainer
 from ai.noticer import upload_predict
@@ -27,15 +28,17 @@ def sync(syncs: List[DataSync] = None):
         s.do_sync()
 
 
-def predictionFromMoel(the_model, scaledX_forecast, predict_item):
+def predictionFromMoel(the_model, scaledX_forecast, predict_item, indexDict: dict):
     prediction = the_model.predict(scaledX_forecast)
     predictionStr = 'DOWN'
     if (prediction > 0.5):
         predictionStr = 'UP'
     content = f"""\n On day {forecastDay.strftime("%m/%d/%Y")}, the model predicts {predict_item} to be {predictionStr} in {str(numForecastDays)} business days. \n"""
     print(content)
-    # if predict_item == 'SPX':
-    #     upload_predict(f'{predict_item} Index', forecastDay, predictionStr)
+    # 上传预测结果
+    # key = [k for k, v in indexDict.items() if v == predict_item]
+    # index_info = get_base_info(key)[0]
+    # upload_predict(index_info['ticker'], forecastDay, predictionStr)
     # send(content)
     return prediction
 
@@ -113,4 +116,4 @@ if __name__ == '__main__':
         ensemble_model = trainer.ensemble_model(rf_model, gbt_model, svc_model, X_train, y_train, X_test, y_test)
 
         if (toForecast):
-            predictionFromMoel(ensemble_model, scaledX_forecast, indexDict[pid])
+            predictionFromMoel(ensemble_model, scaledX_forecast, indexDict[pid], indexDict)

api.py

@@ -227,10 +227,10 @@ class AssetOptimize(ABC):
     '''
 
     @abstractmethod
-    def find_optimize(self, ids, day):
+    def find_optimize(self, fund_ids, day):
         '''
         从多id中，选出指定日期最优的id
-        :param ids: 待选id列表
+        :param fund_ids: 待选id列表
         :param day: 指定日期
         :return: 最优的id
         '''
@@ -305,6 +305,22 @@ class PortfoliosBuilder(ABC):
         pass
 
 
+class PortfoliosChecker(ABC):
+    '''
+    投组组合检测器
+    '''
+
+    @abstractmethod
+    def check(self, day=None, portfolios=None):
+        """
+        检测避免出现最优投组同时出现全部是ft或美盛基金的情况，增加一步替换动作。
+        @param day:
+        @param portfolios:
+        @return:
+        """
+        pass
+
+
 class Solver(ABC):
     '''
     解算器

asset_pool/asset_optimize.py

@@ -110,7 +110,7 @@ class FundDividendSortinoAssetOptimize(SortinoAssetOptimize):
         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)
         # 取得分数高的前optimize_count个
-        return pct_change.columns[sortino.index[0:self.optimize_count]].values
+        return pct_change.columns[sortino.index[0:self.optimize_count]].values,sortino['score']
 
     def get_optimize_pool(self, day):
         opt_pool = rop.get_one(day=day, type=AssetPoolType.OPTIMIZE)
@@ -125,7 +125,7 @@ class FundDividendSortinoAssetOptimize(SortinoAssetOptimize):
             for fund_group in self.get_groups():
                 fund_group = [x for x in fund_group if min_dates[x] <= max_incept_date]
                 if len(fund_group) > self.optimize_count:
-                    pool.extend(self.find_optimize(tuple(fund_group), day))
+                    pool.extend(self.find_optimize(tuple(fund_group), day)[0])
                 elif len(fund_group) <= self.optimize_count:
                     pool.extend(fund_group)
             rop.insert(day, AssetPoolType.OPTIMIZE, sorted(pool))

config-svrobo4.yml

@@ -107,6 +107,9 @@ portfolios: # 投组模块
 #      high-weight: [ 1 ] # 最高权重比例，可给一个值，也可以给多个值，当多个值时，第一个表示只有一个资产时权重，第二个表示只有两个资产时权重，以此类推，最后一个表示其他资产个数时的权重
     poem: # poem相关
       cvar-scale-factor: 0.1 # 计算时用到的系数
+  checker: #投组检测模块
+    switch: off  #是否开启检查
+    custom-type-priority: [ 3,2,1,4 ]  # 检测优先级
 reports: # 报告模块相关
   navs:
     type: FUND

config-svrobo5.yml

@@ -112,6 +112,9 @@ portfolios: # 投组模块
 #      high-weight: [ 1 ] # 最高权重比例，可给一个值，也可以给多个值，当多个值时，第一个表示只有一个资产时权重，第二个表示只有两个资产时权重，以此类推，最后一个表示其他资产个数时的权重
     poem: # poem相关
       cvar-scale-factor: 0.1 # 计算时用到的系数
+  checker: #投组检测模块
+    switch: off  #是否开启检查
+    custom-type-priority: [ 3,2,1,4 ]  # 检测优先级
 reports: # 报告模块相关
   navs:
     type: FUND

config-svrobo6.yml

@@ -111,6 +111,9 @@ portfolios: # 投组模块
       high-weight: [ 0.35 ] # 最高权重比例，可给一个值，也可以给多个值，当多个值时，第一个表示只有一个资产时权重，第二个表示只有两个资产时权重，以此类推，最后一个表示其他资产个数时的权重
     poem: # poem相关
       cvar-scale-factor: 0.1 # 计算时用到的系数
+  checker: #投组检测模块
+    switch: on  #是否开启检查
+    custom-type-priority: [ 3,2,1,4 ]  # 检测优先级
 reports: # 报告模块相关
   navs:
     type: FUND

portfolios/builder.py

@@ -4,7 +4,8 @@ import logging
 from py_jftech import component, autowired, format_date
 from pymysql import IntegrityError, constants
 
-from api import PortfoliosBuilder, PortfoliosRisk, AssetPool, Navs, PortfoliosType, Datum, SolveType, SolverFactory
+from api import PortfoliosBuilder, PortfoliosRisk, AssetPool, Navs, PortfoliosType, Datum, SolveType, SolverFactory, \
+    PortfoliosChecker
 from portfolios.dao import robo_mpt_portfolios as rmp
 
 logger = logging.getLogger(__name__)
@@ -14,11 +15,13 @@ logger = logging.getLogger(__name__)
 class MptPortfoliosBuilder(PortfoliosBuilder):
 
     @autowired
-    def __init__(self, assets: AssetPool = None, navs: Navs = None, datum: Datum = None, factory: SolverFactory = None):
+    def __init__(self, assets: AssetPool = None, navs: Navs = None, datum: Datum = None, factory: SolverFactory = None,
+                 checker: PortfoliosChecker = None):
         self._assets = assets
         self._navs = navs
         self._datum = datum
         self._factory = factory
+        self._checker = checker
 
     def get_portfolios(self, day, risk: PortfoliosRisk, type: PortfoliosType = PortfoliosType.NORMAL):
         try:
@@ -26,6 +29,7 @@ class MptPortfoliosBuilder(PortfoliosBuilder):
             if not portfolio:
                 result = self.build_portfolio(day, type)
                 for build_risk, datas in result.items():
+                    datas['portfolio'] = self._checker.check(day, json.loads(datas['portfolio']))
                     try:
                         rmp.insert({
                             **datas,
@@ -44,7 +48,7 @@ class MptPortfoliosBuilder(PortfoliosBuilder):
             return None
         except Exception as e:
             logger.exception(
-                f"build protfolio of type[{type.name}] and risk[{risk.name}] with date[{format_date(day)}] failure.", e)
+                f"build portfolio of type[{type.name}] and risk[{risk.name}] with date[{format_date(day)}] failure.", e)
             raise e
 
     def build_portfolio(self, day, type: PortfoliosType):
@@ -119,6 +123,7 @@ class MptARCPortfoliosBuilder(MptPortfoliosBuilder):
             if not portfolio:
                 result, detail = self.build_portfolio(day, type)
                 for build_risk, datas in result.items():
+                    datas['portfolio'] = self._checker.check(day, json.loads(datas['portfolio']))
                     try:
                         rmp.insert({
                             **datas,

portfolios/checker.py

+import logging
+
+from py_jftech import autowired, component, get_config
+
+from api import AssetOptimize, PortfoliosChecker, Datum, Navs, DatumType
+
+logger = logging.getLogger(__name__)
+
+
+@component(bean_name='checker')
+class DefaultPortfoliosChecker(PortfoliosChecker):
+
+    @autowired
+    def __init__(self, asset: AssetOptimize = None, navs: Navs = None, datum: Datum = None):
+        self._asset = asset
+        self._navs = navs
+        self._datum = datum
+        self._config = get_config(__name__)
+
+    def check(self, day=None, portfolios: dict = None):
+        if not self._config.get('switch'):
+            return portfolios
+        funds = self._datum.get_datums(type=DatumType.FUND)
+        company = {f"{fund['id']}": fund['companyType'] for fund in funds}
+        customType = {f"{fund['id']}": fund['customType'] for fund in funds}
+        companies = set(company[key] for key in portfolios.keys())
+        # 同时出现全部是ft或美盛基金的情况
+        if len(companies) == 1:
+            # step1:	检查原始投组的customType。检查顺序用列表呈现，依序进行
+            priority = self._config.get('custom-type-priority')
+            for p in priority:
+                keys = [key for key in portfolios.keys() if customType[key] == p]
+                # 若存在匹配值则执行后跳出循环
+                if len(keys) > 0:
+                    ids = [fund['id'] for fund in funds if fund['companyType'] != list(companies)[0]]
+                    best = self.find_highest_score(ids, day)
+                    # 若刚好有一个匹配，直接替换
+                    if len(keys) == 1:
+                        portfolios[best] = portfolios[keys[0]]
+                        # 删除原始键
+                        del portfolios[keys[0]]
+                    else:
+                        # 算分，把分低的替换掉
+                        scores = self.do_score(keys, day)
+                        weight_scores = {key: scores[key]*portfolios[key] for key in keys}
+                        lowest = min(scores, key=lambda k: weight_scores[k])
+                        portfolios[best] = portfolios[lowest]
+                        # 删除原始键
+                        del portfolios[lowest]
+                    break
+        return portfolios
+
+    def do_score(self, ids, day):
+        optimize = self._asset.find_optimize(fund_ids=ids, day=day)
+        scores = optimize[1].to_dict()
+        id_score = {}
+        for k, v in scores.items():
+            id_score[f'{ids[k]}'] = v
+        return id_score
+
+    def find_highest_score(self, ids, day):
+        optimize = self._asset.find_optimize(fund_ids=ids, day=day)
+        return optimize[0][0]