实盘完成待测试

api.py

@@ -368,6 +368,16 @@ class PortfoliosHolder(ABC):
     投资组合持仓器
     '''
 
+    @abstractmethod
+    def get_portfolio_type(self, day, risk: PortfoliosRisk) -> PortfoliosType:
+        '''
+        获取指定日期指定风险等级持仓投组的类型
+        :param day: 指定日期
+        :param risk: 指定风险等级
+        :return: 持仓投组类型
+        '''
+        pass
+
     @abstractmethod
     def get_portfolios_weight(self, day, risk: PortfoliosRisk):
         '''
@@ -378,6 +388,7 @@ class PortfoliosHolder(ABC):
         '''
         pass
 
+    @abstractmethod
     def has_hold(self, risk: PortfoliosRisk) -> bool:
         '''
         是否存在指定分线等级的投组持仓
@@ -484,6 +495,15 @@ class RebalanceRuler(ABC):
         '''
         pass
 
+    @abstractmethod
+    def get_signal_type(self, sign_id) -> SignalType:
+        '''
+        获取指定id的信号类型
+        :param sign_id: 信号id
+        :return: 信号类型
+        '''
+        pass
+
 
 class RoboExecutor(ABC):
     '''

config.yml

@@ -168,6 +168,8 @@ robo-executor:
     start-date: 2008-01-02
     end-date: 2009-01-01
     start-step: 4
+  real:
+    start-date: 2022-11-01
 
 
 

portfolios/holder.py

@@ -2,7 +2,7 @@ import json
 
 import pandas as pd
 
-from api import PortfoliosHolder, PortfoliosRisk, RebalanceRuler, Navs, SignalType, RoboExecutor
+from api import PortfoliosHolder, PortfoliosRisk, RebalanceRuler, Navs, SignalType, RoboExecutor, PortfoliosType
 from framework import (
     component, autowired, get_config, next_workday, filter_weekend,
     prev_workday, transaction, workday_range, format_date, get_logger
@@ -23,6 +23,13 @@ class NextReblanceHolder(PortfoliosHolder):
         self._executor = executor
         self._config = get_config(__name__)
 
+    def get_portfolio_type(self, day, risk: PortfoliosRisk) -> PortfoliosType:
+        hold = rhp.get_one(day, risk)
+        if hold:
+            signal_type = self._rule.get_signal_type(hold['signal_id'])
+            return signal_type.p_type if signal_type else PortfoliosType.NORMAL
+        return PortfoliosType.NORMAL
+
     def get_last_rebalance_date(self, risk: PortfoliosRisk, max_date=None, signal_id=None):
         assert risk, f"get last rebalance date, risk can not be none"
         last = rhp.get_last_one(max_date=max_date, risk=risk, signal_id=signal_id, rebalance=True)

rebalance/ruler.py

@@ -62,5 +62,9 @@ class LevelRebalanceRuler(RebalanceRuler):
                 return signal
         return None
 
+    def get_signal_type(self, sign_id) -> SignalType:
+        signal = rrs.get_by_id(sign_id)
+        return SignalType(signal['type']) if signal else None
+
     def commit_signal(self, sign_id):
         rrs.update(sign_id, {'effective': True})

robo_executor.py

 import pandas as pd
 
-from framework import component, autowired, block_execute, get_config, get_logger, filter_weekend, workday_range, format_date
-from api import RoboExecutor, AssetRisk, Datum, AssetPool, PortfoliosBuilder, PortfoliosRisk, PortfoliosHolder
+from framework import component, autowired, block_execute, get_config, get_logger, filter_weekend, workday_range, format_date, prev_workday
+from api import RoboExecutor, AssetRisk, Datum, AssetPool, PortfoliosBuilder, PortfoliosRisk, PortfoliosHolder, PortfoliosType
 from datetime import datetime as dt
 import time
 from enum import Enum, unique
@@ -72,7 +72,7 @@ class BacktestExector(RoboExecutor):
             workdays = workday_range(self.start_date, self.end_date)
             for date in workdays:
                 self._risk.get_risk_pool(date)
-                time.sleep(0.05)
+                time.sleep(0.05) # 这里需要sleep，否则里面多进程太快，数据库连接容易超时
             for date in workdays:
                 self._pool.get_pool(date)
             logger.info(f"build asset pool success, use[{(dt.now() - now).seconds}s]")
@@ -87,3 +87,29 @@ class BacktestExector(RoboExecutor):
             block_execute(self._hold.build_hold_portfolio, {x: (self.end_date, x) for x in PortfoliosRisk}, isolate=True, result=False)
             logger.info(f"build hold portfolios success, use[{(dt.now() - now).seconds}s]")
 
+
+@component(bean_name='real')
+class RealExecutor(RoboExecutor):
+
+    @autowired
+    def __init__(self, builder: PortfoliosBuilder = None, hold: PortfoliosHolder = None):
+        self._builder = builder
+        self._hold = hold
+        self._config = get_config(__name__)['real']
+
+    @property
+    def start_date(self):
+        return pd.to_datetime(filter_weekend(self._config['start-date']))
+
+    def start_exec(self):
+        date = dt.combine(dt.today().date(), dt.min.time())
+        for risk in PortfoliosRisk:
+            logger.info(f"start to build risk[{risk.name}] real for date[{format_date(date)}]".center(50, '-'))
+            now = dt.now()
+            self._builder.get_portfolios(prev_workday(date), risk)
+            self._hold.build_hold_portfolio(date, risk)
+            p_type = self._hold.get_portfolio_type(date, risk)
+            if p_type is not PortfoliosType.NORMAL:
+                self._builder.get_portfolios(prev_workday(date), p_type)
+            logger.info(f"build risk[{risk.name}] real for date[{format_date(date)}] success, use[{(dt.now() - now).seconds}s]")
+