回测

api.py

@@ -246,6 +246,14 @@ class PortfoliosBuilder(ABC):
         '''
         pass
 
+    @abstractmethod
+    def get_all_portfolios(self, risk: PortfoliosRisk = None):
+        """
+        查询所有优选基金
+        @param risk:
+        """
+        pass
+
 
 class Solver(ABC):
     '''
@@ -424,6 +432,14 @@ class PortfoliosHolder(ABC):
         '''
         pass
 
+    @property
+    @abstractmethod
+    def month_dividend(self):
+        """
+        获取当月配息
+        """
+        pass
+
 
 class RoboExecutor(ABC):
     '''

basic/dao/robo_fund_navs.py

@@ -4,6 +4,8 @@ __COLUMNS__ = {
     'rfn_fund_id': 'fund_id',
     'rfn_date': 'nav_date',
     'rfn_nav_cal': 'nav_cal',
+    'rfn_av': 'av',
+    'rfn_div': 'dividend',
 }
 
 __INSERT_COLUMNS__ = {

config-svrobo5.yml

@@ -75,6 +75,7 @@ portfolios: # 投组模块
   holder: # 持仓投组相关
     init-nav: 100 # 初始金额
     min-interval-days: 10 # 两次实际调仓最小间隔期，单位交易日
+    dividend-rate: 0.09 #设定年化配息率
   solver: # 解算器相关
     tol: 1E-10 # 误差满足条件
     navs: # 净值要求
@@ -87,12 +88,10 @@ portfolios: # 投组模块
       US_STOCK: [ 0.3, 0.5, 0.7 ]
       US_HY_BOND: [ 0.6, 0.4, 0.2 ]
       US_IG_BOND: [ 0.1, 0.1, 0.1 ]
-      dividend-rate: 0.09
     riskctl-ratio:
       US_STOCK: [ 0.2, 0.4, 0.6 ]
       US_HY_BOND: [ 0.5, 0.3, 0.1 ]
       US_IG_BOND: [ 0.3, 0.3, 0.3 ]
-      dividend-rate: 0.09
     matrix-rtn-days: 20 # 计算回报率矩阵时，回报率滚动天数
     asset-count: [1,3] # 投组资产个数。e.g. count 或 [min, max] 分别表示 最大最小都为count 或 最小为min 最大为max，另外这里也可以类似上面给不同风险等级分别配置
     mpt: # mpt计算相关

portfolios/builder.py

@@ -6,6 +6,7 @@ from pymysql import IntegrityError, constants
 
 from api import PortfoliosBuilder, PortfoliosRisk, AssetPool, Navs, PortfoliosType, Datum, SolveType, SolverFactory
 from portfolios.dao import robo_mpt_portfolios as rmp
+from portfolios.dao.robo_mpt_portfolios import get_list
 
 logger = logging.getLogger(__name__)
 
@@ -78,6 +79,9 @@ class MptPortfoliosBuilder(PortfoliosBuilder):
     def clear(self, day=None, risk: PortfoliosRisk = None):
         rmp.delete(min_date=day, risk=risk)
 
+    def get_all_portfolios(self, risk: PortfoliosRisk = None):
+        return get_list(risk=risk)
+
 
 @component(bean_name='poem')
 class PoemPortfoliosBuilder(MptPortfoliosBuilder):
@@ -97,10 +101,11 @@ class PoemPortfoliosBuilder(MptPortfoliosBuilder):
                 portfolio, cvar = solver.solve_poem(min_rtn, max_rtn, mpt_cvar, maxCVaR_whenMinV)
                 if not portfolio:
                     portfolio = mpt_portfolio
-                portfolios = {**portfolios, **portfolio }
+                portfolios = {**portfolios, **portfolio}
             if portfolios:
                 result[risk] = {
                     'solve': SolveType.POEM,
                     'portfolio': json.dumps(portfolios),
                 }
         return result
+

portfolios/dao/mysql.sql

@@ -27,6 +27,9 @@ CREATE TABLE IF NOT EXISTS robo_hold_portfolios
     rhp_rebalance   TINYINT         NOT NULL DEFAULT 0 COMMENT '是否调仓',
     rhp_portfolios  JSON            NOT NULL COMMENT '投组信息',
     rhp_nav         DOUBLE(12, 4)          NOT NULL COMMENT '资产值',
+    `rhp_div` double(12, 4) NOT NULL COMMENT '配息金额',
+    `rhp_div_acc` double(12, 4) NOT NULL COMMENT '累计配息金额',
+    `v_nav_div_acc` double(12, 4) GENERATED ALWAYS AS ((`rhp_div_acc` + `rhp_nav`)) VIRTUAL COMMENT '配息金额+净值' NOT NULL,
     rhp_create_time DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
     rhp_update_time DATETIME                 DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
     PRIMARY KEY (rhp_id),

portfolios/dao/robo_hold_portfolios.py

@@ -6,6 +6,8 @@ __COLUMNS__ = {
     'rhp_id': 'id',
     'rhp_date': 'date',
     'rhp_risk': 'risk',
+    'rhp_div': 'dividend',
+    'rhp_div_acc': 'div_acc',
     'rhp_rrs_id': 'signal_id',
     'rhp_rebalance': 'rebalance',
     'rhp_portfolios': 'portfolios',

portfolios/dao/robo_mpt_portfolios.py

@@ -49,4 +49,5 @@ def get_list(max_date=None, min_date=None, type: PortfoliosType = None, risk: Po
     return f'''
     select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_mpt_portfolios
     {where(*sqls, rmp_risk=risk, rmp_type=type)}
+    order by rmp_date
     '''

portfolios/holder.py

@@ -6,7 +6,7 @@ from py_jftech import (
     component, autowired, get_config, next_workday, format_date
 )
 
-from api import PortfoliosHolder, PortfoliosRisk, Navs, RoboExecutor, PortfoliosType
+from api import PortfoliosHolder, PortfoliosRisk, Navs, RoboExecutor, PortfoliosType, PortfoliosBuilder
 from portfolios.dao import robo_hold_portfolios as rhp
 from portfolios.utils import format_weight
 
@@ -17,9 +17,10 @@ logger = logging.getLogger(__name__)
 class DividendPortfoliosHolder(PortfoliosHolder):
 
     @autowired(names={'executor': RoboExecutor.use_name()})
-    def __init__(self, navs: Navs = None, executor: RoboExecutor = None):
+    def __init__(self, navs: Navs = None, executor: RoboExecutor = None, builder: PortfoliosBuilder = None):
         self._navs = navs
         self._executor = executor
+        self._builder = builder
         self._config = get_config(__name__)
 
     def get_portfolio_type(self, day, risk: PortfoliosRisk) -> PortfoliosType:
@@ -42,27 +43,70 @@ class DividendPortfoliosHolder(PortfoliosHolder):
 
     def build_hold_portfolio(self, day, risk: PortfoliosRisk):
         last_nav = rhp.get_last_one(max_date=day, risk=risk)
-        start = next_workday(last_nav['date'] if last_nav else self._executor.start_date)
+        # 从基金优选池选取所有调仓日基金
+        portfolios = self._builder.get_all_portfolios(risk)
+        portfoliosMap = {p['date']: p['portfolio'] for p in portfolios}
+        start = last_nav['date'] if last_nav else list(portfoliosMap.keys())[0]
         try:
-            if not last_nav:
-                pass
             while start <= day:
                 logger.info(f"start to build hold portfolio[{risk.name}] for date[{format_date(start)}]")
+                if start in portfoliosMap.keys():
+                    self.do_rebalance(start, risk, portfoliosMap[start], last_nav)
+                else:
                     self.no_rebalance(start, risk, last_nav)
                 start = next_workday(start)
                 last_nav = rhp.get_last_one(max_date=day, risk=risk)
         except Exception as e:
             logger.exception(f"build hold portfolio[{risk.name}] for date[{format_date(start)}] failure.", e)
 
+    def do_rebalance(self, day, risk: PortfoliosRisk, portfolio, last_nav):
+        weight = {int(x[0]): x[1] for x in json.loads(portfolio).items()}
+        dividend_acc = 0
+        if last_nav:
+            share = {int(x): y for x, y in json.loads(last_nav['portfolios'])['share'].items()}
+            fund_div_tuple = self.get_navs_and_div(fund_ids=tuple(set(weight) | set(share)), day=day)
+            navs = fund_div_tuple[0]
+            fund_dividend = fund_div_tuple[1]
+            nav = round(sum([navs[x] * y for x, y in share.items()]), 4)
+            dividend_acc = last_nav['div_acc']
+        else:
+            nav = self.init_nav
+            fund_div_tuple =self.get_navs_and_div(fund_ids=tuple(weight), day=day)
+            navs = fund_div_tuple[0]
+            fund_dividend = fund_div_tuple[1]
+        dividend = nav * self.month_dividend
+        nav = nav - dividend
+        share = {x: nav * w / navs[x] for x, w in weight.items()}
+        fund_dividend = sum(map(lambda k: share[k] * fund_dividend[k], filter(lambda k: k in fund_dividend, share.keys())))
+        dividend_acc = dividend + dividend_acc + fund_dividend
+        rhp.insert({
+            'date': day,
+            'risk': risk,
+            'dividend': dividend,
+            'div_acc': dividend_acc,
+            'rebalance': True,
+            'portfolios': {
+                'weight': weight,
+                'share': share,
+            },
+            'nav': nav,
+        })
+
     def no_rebalance(self, day, risk: PortfoliosRisk, last_nav):
         share = {int(x): y for x, y in json.loads(last_nav['portfolios'])['share'].items()}
-        navs = self.get_navs(fund_ids=tuple(share), day=day)
+        fund_div_tuple = self.get_navs_and_div(fund_ids=tuple(share), day=day)
+        navs = fund_div_tuple[0]
+        fund_dividend = fund_div_tuple[1]
         nav = round(sum([navs[x] * y for x, y in share.items()]), 4)
         weight = {x: round(y * navs[x] / nav, 2) for x, y in share.items()}
         weight = format_weight(weight)
+        fund_dividend = sum(map(lambda k: share[k] * fund_dividend[k], filter(lambda k: k in fund_dividend, share.keys())))
+        dividend_acc = last_nav['div_acc'] + fund_dividend
         rhp.insert({
             'date': day,
             'risk': risk,
+            'dividend': 0,
+            'div_acc': dividend_acc,
             'signal_id': last_nav['signal_id'],
             'rebalance': False,
             'portfolios': {
@@ -72,15 +116,21 @@ class DividendPortfoliosHolder(PortfoliosHolder):
             'nav': nav,
         })
 
-    def get_navs(self, day, fund_ids):
+    def get_navs_and_div(self, day, fund_ids):
         navs = pd.DataFrame(self._navs.get_fund_navs(fund_ids=fund_ids, max_date=day))
-        navs = navs.pivot_table(index='nav_date', columns='fund_id', values='nav_cal')
+        dividend = navs.pivot_table(index='nav_date', columns='fund_id', values='dividend')
+        navs = navs.pivot_table(index='nav_date', columns='fund_id', values='av')
         navs.fillna(method='ffill', inplace=True)
-        return dict(navs.iloc[-1])
+        dividend.fillna(method='ffill', inplace=True)
+        return dict(navs.iloc[-1]), dict(dividend.iloc[-1])
 
     def clear(self, day=None, risk: PortfoliosRisk = None):
         rhp.delete(min_date=day, risk=risk)
 
+    @property
+    def month_dividend(self):
+        return self._config['dividend-rate'] / 12
+
     @property
     def interval_days(self):
         return self._config['min-interval-days']

robo_executor.py

@@ -119,7 +119,6 @@ class BacktestExecutor(RoboExecutor):
             now = dt.now()
             wait([self.async_build_hold(x) for x in PortfoliosRisk])
             logger.info(f"build hold portfolios success, use[{(dt.now() - now).seconds}s]")
-        logger.info("start to export report".center(50, '-'))
 
     @asynchronized(isolate=True)
     def async_build_risk_date(self, asset_id):