但任务回测完成

api.py

@@ -129,7 +129,7 @@ class Navs(ABC):
         pass
 
     @abstractmethod
-    def get_nav_start_date(self, fund_ids = None):
+    def get_nav_start_date(self, fund_ids=None):
         '''
         获取指定id资产的净值开始时间
         :param fund_ids: 指定id资产，如果为None，则返回全部资产的开始时间
@@ -220,6 +220,14 @@ class AssetRisk(ABC):
         '''
         pass
 
+    @abstractmethod
+    def clear(self, day=None):
+        '''
+        清除指定日期之后的资产风控ewma数据，如果没有给日期，则全部清空
+        :param day: 指定清除的开始日期，可选
+        '''
+        pass
+
 
 class AssetPool(ABC):
     '''
@@ -235,6 +243,14 @@ class AssetPool(ABC):
         '''
         pass
 
+    @abstractmethod
+    def clear(self, day=None):
+        '''
+        清除指定日期之后的资产池数据，如果没有给日期，则全部清空
+        :param day: 指定清除的开始日期，可选
+        '''
+        pass
+
 
 class PortfoliosBuilder(ABC):
     '''
@@ -262,6 +278,15 @@ class PortfoliosBuilder(ABC):
         '''
         pass
 
+    @abstractmethod
+    def clear(self, day=None, risk: PortfoliosRisk = None):
+        '''
+        清除指定风险等级，指定日期之后的最优投组
+        :param day: 指定清除的开始日期，可选，如果没给，则清除全部日期
+        :param risk: 指定风险等级，如果没给，则清除全部风险等级
+        '''
+        pass
+
 
 class Solver(ABC):
     '''
@@ -391,6 +416,15 @@ class PortfoliosHolder(ABC):
         '''
         pass
 
+    @abstractmethod
+    def clear(self, day=None, risk: PortfoliosRisk = None):
+        '''
+        清除指定风险等级，指定日期之后的持仓投组
+        :param day: 指定清除的开始日期，可选，如果没给，则清除全部日期
+        :param risk: 指定风险等级，如果没给，则清除全部风险等级
+        '''
+        pass
+
 
 class DriftSolver(ABC):
     '''
@@ -455,6 +489,7 @@ class RoboExecutor(ABC):
     '''
     ROBO执行器，整合以上逻辑，进行实盘或回测
     '''
+
     @abstractmethod
     def start_exec(self):
         '''
@@ -470,4 +505,3 @@ class RoboExecutor(ABC):
     @staticmethod
     def use_name():
         return get_config('robo-executor')['use']
-

asset_pool/asset_pool.py

@@ -2,6 +2,7 @@ from datetime import datetime as dt
 
 from api import AssetPool, AssetOptimize, AssetRisk
 from framework import component, autowired
+from asset_pool.dao import robo_assets_pool as rap
 
 
 @component
@@ -16,3 +17,6 @@ class FundAssetPool(AssetPool):
         opti_pool = self._optimize.get_optimize_pool(day)
         risk_pool = self._risk.get_risk_pool(day)
         return [x for x in opti_pool if x not in risk_pool]
+
+    def clear(self, day=None):
+        rap.delete(day)

asset_pool/asset_risk.py

@@ -7,7 +7,7 @@ from scipy.stats import norm
 
 from api import AssetRisk, Navs, AssetRiskDateType as DateType, Datum, AssetPoolType, RoboExecutor
 from asset_pool.dao import asset_risk_dates as ard, asset_ewma_value as aev, robo_assets_pool as rap
-from framework import component, autowired, get_config, format_date, block_execute, get_logger
+from framework import component, autowired, get_config, format_date, block_execute, get_logger, transaction
 
 logger = get_logger(__name__)
 
@@ -58,6 +58,11 @@ class CvarEwmaAssetRisk(AssetRisk):
         except Exception as e:
             logger.exception(f"build risk date for asset[{asset_id}] after date[{risk_date}] to date[{format_date(day)}] error", e)
 
+    @transaction
+    def clear(self, day=None):
+        ard.delete(day)
+        aev.delete(day)
+
     def get_next_date(self, asset_id, day=dt.today()):
         last = ard.get_last_one(asset_id, day)
         if not last or DateType(last['type']) is DateType.START_DATE:

asset_pool/dao/asset_ewma_value.py

@@ -16,6 +16,14 @@ def insert(asset_id, date, value):
     '''
 
 
+@write
+def delete(day=None):
+    if day:
+        return f"delete from asset_ewma_value where aev_date >= '{format_date(day)}'"
+    else:
+        return 'truncate table asset_ewma_value'
+
+
 @read(one=True)
 def get_last_one(asset_id, max_date=None):
     sqls = []

asset_pool/dao/asset_risk_dates.py

@@ -28,3 +28,11 @@ def get_last_one(fund_id, date=None, type: DateType = None):
     select {','.join([f"`{x[0]}` as `{x[1]}`" for x in __COLUMNS__.items()])} 
     from asset_risk_dates {where(sql, **kwargs)} order by ard_date desc, ard_type asc limit 1
     '''
+
+
+@write
+def delete(day=None):
+    if day:
+        return f"delete from asset_risk_dates where ard_date >= '{format_date(day)}'"
+    else:
+        return 'truncate table asset_risk_dates'

asset_pool/dao/robo_assets_pool.py

@@ -28,3 +28,11 @@ def insert(day, type: AssetPoolType, pool: list):
         insert into robo_assets_pool(rap_date, rap_type, rap_asset_ids)
         values ('{format_date(day)}', {type.value},'{json.dumps(pool)}')
     '''
+
+
+@write
+def delete(day=None):
+    if day:
+        return f"delete from robo_assets_pool where rap_date >= '{format_date(day)}'"
+    else:
+        return 'truncate table robo_assets_pool'

config.yml

@@ -167,6 +167,7 @@ robo-executor:
   backtest:
     start-date: 2008-01-02
     end-date: 2009-01-01
+    start-step: 4
 
 
 

portfolios/builder.py

@@ -70,6 +70,9 @@ class MptPortfoliosBuilder(PortfoliosBuilder):
             }
         return result, detail
 
+    def clear(self, day=None, risk: PortfoliosRisk = None):
+        rmp.delete(min_date=day, risk=risk)
+
 
 @component(bean_name='poem')
 class PoemPortfoliosBuilder(MptPortfoliosBuilder):

portfolios/dao/robo_hold_portfolios.py

@@ -43,3 +43,12 @@ def insert(datas):
     insert into robo_hold_portfolios({','.join([x for x in datas.keys()])})
     values ({','.join([f"'{x[1]}'" for x in datas.items()])})
     '''
+
+
+@write
+def delete(min_date=None, risk: PortfoliosRisk = None):
+    if min_date is None and risk is None:
+        return 'truncate table robo_hold_portfolios'
+    else:
+        sql = f"rhp_date >= '{format_date(min_date)}'" if min_date else None
+        return f"delete from robo_hold_portfolios {where(sql, rhp_risk=risk)}"
\ No newline at end of file

portfolios/dao/robo_mpt_portfolios.py

@@ -24,6 +24,15 @@ def insert(datas):
     '''
 
 
+@write
+def delete(min_date=None, risk: PortfoliosRisk = None):
+    if min_date is None and risk is None:
+        return 'truncate table robo_mpt_portfolios'
+    else:
+        sql = f"rmp_date >= '{format_date(min_date)}'" if min_date else None
+        return f"delete from robo_mpt_portfolios {where(sql, rmp_risk=risk)}"
+
+
 @read(one=True)
 def get_one(day, type: PortfoliosType, risk: PortfoliosRisk):
     return f'''

portfolios/holder.py

@@ -110,6 +110,9 @@ class NextReblanceHolder(PortfoliosHolder):
         navs.fillna(method='ffill', inplace=True)
         return dict(navs.iloc[-1])
 
+    def clear(self, day=None, risk: PortfoliosRisk = None):
+        rhp.delete(min_date=day, risk=risk)
+
     @property
     def interval_days(self):
         return self._config['min-interval-days']

robo_executor.py

@@ -4,10 +4,22 @@ from framework import component, autowired, block_execute, get_config, get_logge
 from api import RoboExecutor, AssetRisk, Datum, AssetPool, PortfoliosBuilder, PortfoliosRisk, PortfoliosHolder
 from datetime import datetime as dt
 import time
+from enum import Enum, unique
 
 logger = get_logger(__name__)
 
 
+@unique
+class BacktestStep(Enum):
+    EWMA_VALUE = 1
+    ASSET_POOL = 2
+    NORMAL_PORTFOLIO = 3
+    HOLD_PORTFOLIO = 4
+
+    def within(self, step: Enum):
+        return self.value <= step.value
+
+
 @component(bean_name='backtest')
 class BacktestExector(RoboExecutor):
 
@@ -25,29 +37,53 @@ class BacktestExector(RoboExecutor):
     def start_date(self):
         return pd.to_datetime(filter_weekend(self._config['start-date']))
 
+    @property
+    def start_step(self) -> BacktestStep:
+        return BacktestStep(self._config['start-step'])
+
     @property
     def end_date(self):
         return pd.to_datetime(self._config['end-date'])
 
+    def clear_datas(self):
+        if self.start_step.within(BacktestStep.EWMA_VALUE):
+            logger.info('start to clear fund ewma value'.center(50, '-'))
+            self._risk.clear()
+        if self.start_step.within(BacktestStep.ASSET_POOL):
+            logger.info('start to clear asset pool'.center(50, '-'))
+            self._pool.clear()
+        if self.start_step.within(BacktestStep.NORMAL_PORTFOLIO):
+            logger.info('start to clear normal portfolios'.center(50, '-'))
+            self._builder.clear()
+        if self.start_step.within(BacktestStep.HOLD_PORTFOLIO):
+            logger.info('start to clear hold portfolios'.center(50, '-'))
+            self._hold.clear()
+
     def start_exec(self):
-        logger.info("start to build fund ewma value.".center(50, '-'))
-        now = dt.now()
-        block_execute(self._risk.build_risk_date, {x['id']: (x['id'], self.end_date) for x in self._datum.get_fund_datums(risk=(3, 4, 5))}, isolate=True, result=False)
-        logger.info(f"build fund ewma value success, use[{(dt.now() - now).seconds}s]")
-        logger.info("start to build asset pool".center(50, '-'))
-        now = dt.now()
-        workdays = workday_range(self.start_date, self.end_date)
-        for date in workdays:
-            self._risk.get_risk_pool(date)
-            time.sleep(0.05)
-        for date in workdays:
-            self._pool.get_pool(date)
-        logger.info(f"build fund ewma value success, use[{(dt.now() - now).seconds}s]")
-        logger.info("start to build normal portfolios".center(50, '-'))
-        now = dt.now()
-        block_execute(self._builder.get_portfolios, {f'{x.name}_{format_date(j)}': (j, x) for x in PortfoliosRisk for j in workday_range(self.start_date, self.end_date)}, isolate=True, result=False)
-        logger.info(f"build normal portfolios success, use[{(dt.now() - now).seconds}s]")
-        logger.info("start to build hold portfolios".center(50, '-'))
-        now = dt.now()
-        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]")
+        self.clear_datas()
+        if self.start_step.within(BacktestStep.EWMA_VALUE):
+            logger.info("start to build fund ewma value.".center(50, '-'))
+            now = dt.now()
+            block_execute(self._risk.build_risk_date, {x['id']: (x['id'], self.end_date) for x in self._datum.get_fund_datums(risk=(3, 4, 5))}, isolate=True, result=False)
+            logger.info(f"build fund ewma value success, use[{(dt.now() - now).seconds}s]")
+        if self.start_step.within(BacktestStep.ASSET_POOL):
+            logger.info("start to build asset pool".center(50, '-'))
+            now = dt.now()
+            workdays = workday_range(self.start_date, self.end_date)
+            for date in workdays:
+                self._risk.get_risk_pool(date)
+                time.sleep(0.05)
+            for date in workdays:
+                self._pool.get_pool(date)
+            logger.info(f"build asset pool success, use[{(dt.now() - now).seconds}s]")
+        if self.start_step.within(BacktestStep.NORMAL_PORTFOLIO):
+            logger.info("start to build normal portfolios".center(50, '-'))
+            now = dt.now()
+            block_execute(self._builder.get_portfolios, {f'{x.name}_{format_date(j)}': (j, x) for x in PortfoliosRisk for j in workday_range(self.start_date, self.end_date)}, isolate=True, result=False)
+            logger.info(f"build normal portfolios success, use[{(dt.now() - now).seconds}s]")
+        if self.start_step.within(BacktestStep.HOLD_PORTFOLIO):
+            logger.info("start to build hold portfolios".center(50, '-'))
+            now = dt.now()
+            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]")
+