加入信号，调整和robo3一致

api.py

@@ -55,6 +55,12 @@ class LoggerType(Enum):
     SIGNAL = 'signal'
 
 
+@unique
+class SignalType(Enum):
+    NORMAL = 1
+
+SignalType.NORMAL.p_type = PortfoliosType.NORMAL
+
 class DataSync(ABC):
     '''
     数据同步服务，需要同步数据的服务，可以实现该接口
@@ -438,11 +444,12 @@ class PortfoliosHolder(ABC):
         pass
 
     @abstractmethod
-    def build_hold_portfolio(self, day, risk: PortfoliosRisk):
+    def build_hold_portfolio(self, day, risk: PortfoliosRisk, force_mpt=False):
         '''
         构建指定日期，指定风险等级的持仓投组，以day为截止日期，会持续补满
         :param day: 指定日期
         :param risk: 指定风险等级
+        :param force_mpt: 如果为True，则强制计算当天mpt，否则不强制计算
         :return:
         '''
         pass
@@ -457,6 +464,15 @@ class PortfoliosHolder(ABC):
         '''
         pass
 
+    @abstractmethod
+    def get_rebalance_date_by_signal(self, signal_id):
+        '''
+        获取指定调仓信号触发的实际调仓日期
+        :param signal_id: 指定的调仓信号
+        :return: 实际调仓日期
+        '''
+        pass
+
     @property
     @abstractmethod
     def interval_days(self):
@@ -570,7 +586,6 @@ class RoboExportor(ABC):
         pass
 
 
-
 class DataLogger(ABC):
 
     @abstractmethod
@@ -596,3 +611,37 @@ class DataLogger(ABC):
         :return: 日志数据列表
         '''
         pass
+
+
+class RebalanceSignal(ABC):
+    '''
+    控制信号，发起是否调仓服务
+    '''
+
+    @abstractmethod
+    def get_signal(self, day, risk: PortfoliosRisk):
+        '''
+        根据日期和风险等级，返回当天的调仓信号，如果没有则返回None
+        :param day: 指定的日期，净值日
+        :param risk: 指定的风险等级
+        :return: 如果有信号，则返回信号数据，否则返回None
+        '''
+        pass
+
+    @property
+    @abstractmethod
+    def signal_type(self) -> SignalType:
+        '''
+        返回信号类型
+        :return: 信号类型
+        '''
+        pass
+
+    @abstractmethod
+    def get_last_signal(self, day, risk: PortfoliosRisk):
+        '''
+        根据日期和风险等级，返回最近的调仓信号，如果没有则返回None
+        :param day: 指定的日期，净值日
+        :param risk: 指定的风险等级
+        :return: 如果有信号，则返回信号数据，否则返回None
+        '''

config-svrobo6.yml

@@ -242,7 +242,7 @@ robo-executor: # 执行器相关
     clean-up: true
   real: # 实盘执行器
     export: ${EXPORT_ENABLE:on} # 是否开启报告
-    start-date: 2022-09-01 # 实盘开始时间
+    start-date: 2023-05-08 # 实盘开始时间
     include-date: []
 
 

portfolios/builder.py

 import json
 import logging
-from datetime import datetime as dt, timedelta
-from typing import List
 
-import pandas as pd
-from py_jftech import component, autowired, format_date, prev_workday, is_workday, get_config
+from py_jftech import component, autowired, format_date
 from pymysql import IntegrityError, constants
 
-from api import PortfoliosBuilder, PortfoliosRisk, AssetPool, Navs, PortfoliosType, Datum, SolveType, SolverFactory, \
-    RoboReportor, DatumType
+from api import PortfoliosBuilder, PortfoliosRisk, AssetPool, Navs, PortfoliosType, Datum, SolveType, SolverFactory
 from portfolios.dao import robo_mpt_portfolios as rmp
 
 logger = logging.getLogger(__name__)
@@ -26,18 +22,7 @@ class MptPortfoliosBuilder(PortfoliosBuilder):
 
     def get_portfolios(self, day, risk: PortfoliosRisk, type: PortfoliosType = PortfoliosType.NORMAL):
         try:
-            # 若记录为空则，将传入日期作为初始日期，进行build
-            portfolio = rmp.get_last_one(day, type, risk)
-            if portfolio:
-                frequency = get_config('portfolios')['holder']['warehouse-frequency']
-                date = pd.to_datetime(day.replace(day=1)) + pd.DateOffset(months=frequency)
-                date = date - timedelta(days=1)
-                # 指定周期末的工作日
-                date = date if is_workday(date) else prev_workday(date)
-                if date == day:
-                    portfolio = None
-                elif portfolio['date'] != day:
-                    return None
+            portfolio = rmp.get_one(day, type, risk)
             if not portfolio:
                 result = self.build_portfolio(day, type)
                 for build_risk, datas in result.items():
@@ -125,65 +110,3 @@ class PoemPortfoliosBuilder(MptPortfoliosBuilder):
         return result
 
 
-@component(bean_name='signal-report')
-class SignalReportor(RoboReportor):
-
-    @autowired
-    def __init__(self, datum: Datum = None):
-        self._datum = datum
-
-    @property
-    def report_name(self) -> str:
-        return '调仓信号'
-
-    def load_report(self, max_date=dt.today(), min_date=None) -> List[dict]:
-        result = []
-        datums = {str(x['id']): x for x in self._datum.get_datums(type=DatumType.FUND, exclude=False)}
-        for signal in rmp.get_list(max_date=max_date, min_date=min_date):
-            for fund_id, weight in json.loads(signal['portfolio']).items():
-                result.append({
-                    'risk': PortfoliosRisk(signal['risk']).name,
-                    'rebalance_date': signal['date'],
-                    'portfolio_type': PortfoliosType.NORMAL.name,
-                    'ft_ticker': datums[fund_id]['ftTicker'],
-                    'blooberg_ticker': datums[fund_id]['bloombergTicker'],
-                    'fund_name': datums[fund_id]['chineseName'],
-                    'weight': weight
-                })
-        return result
-
-
-
-
-@component(bean_name='daily-signal-report')
-class DailySignalReportor(RoboReportor):
-
-    @autowired
-    def __init__(self, datum: Datum = None):
-        self._datum = datum
-
-    @property
-    def report_name(self) -> str:
-        return '每日调仓信号'
-
-    def load_report(self, max_date=prev_workday(dt.today()), min_date=None) -> List[dict]:
-        portfolio = rmp.get_one(max_date, type=PortfoliosType.NORMAL, risk=PortfoliosRisk.FT3)
-        result = {}
-        if portfolio:
-            datum_ids = list(json.loads(portfolio['portfolio']).keys())
-            datums = pd.DataFrame(self._datum.get_datums(type=DatumType.FUND, datum_ids=datum_ids))
-            datums.set_index('id', inplace=True)
-
-            result['risk'] = [portfolio['risk'] for i in datum_ids]
-            result['rebalance_type'] = [portfolio['type'] for i in datum_ids]
-
-            result['weight'] = [format(i, '.0%') for i in json.loads(portfolio['portfolio']).values()]
-            result['asset_ids'] = [datums.loc[int(i)]['ftTicker'] for i in datum_ids]
-            result['name'] = [datums.loc[int(i)]['chineseName'] for i in datum_ids]
-            result['lipper_id'] = [datums.loc[int(i)]['lipperKey'] for i in datum_ids]
-            result['date'] = [max_date for i in datum_ids]
-            result = pd.DataFrame(result)
-            result = result[['lipper_id', 'asset_ids', 'name', 'weight', 'risk', 'date', 'rebalance_type']]
-            return result.to_dict('records')
-
-        return []

portfolios/holder.py

@@ -10,7 +10,7 @@ from py_jftech import (
 )
 
 from api import PortfoliosHolder, PortfoliosRisk, Navs, RoboExecutor, PortfoliosType, PortfoliosBuilder, RoboReportor, \
-    DatumType, Datum
+    DatumType, Datum, RebalanceSignal
 from portfolios.dao import robo_hold_portfolios as rhp
 from portfolios.utils import format_weight
 
@@ -22,13 +22,14 @@ class DividendPortfoliosHolder(PortfoliosHolder):
 
     @autowired(names={'executor': RoboExecutor.use_name()})
     def __init__(self, navs: Navs = None, executor: RoboExecutor = None, builder: PortfoliosBuilder = None,
-                 datum: Datum = None, mpt: PortfoliosBuilder = None):
+                 datum: Datum = None, mpt: PortfoliosBuilder = None, signal: RebalanceSignal = None):
         self._navs = navs
         self._executor = executor
         self._builder = builder
         self._config = get_config(__name__)
         self._datum = datum
         self._mpt = mpt
+        self._signal = signal
 
     def get_portfolio_type(self, day, risk: PortfoliosRisk) -> PortfoliosType:
         return PortfoliosType.NORMAL
@@ -38,6 +39,10 @@ class DividendPortfoliosHolder(PortfoliosHolder):
         last = rhp.get_last_one(max_date=max_date, risk=risk, rebalance=True)
         return last['date'] if last else None
 
+    def get_rebalance_date_by_signal(self, signal_id):
+        last = rhp.get_last_one(signal_id=signal_id, rebalance=True)
+        return last['date'] if last else None
+
     def get_portfolios_weight(self, day, risk: PortfoliosRisk):
         hold = rhp.get_one(day, risk)
         if hold:
@@ -48,21 +53,23 @@ class DividendPortfoliosHolder(PortfoliosHolder):
     def has_hold(self, risk: PortfoliosRisk) -> bool:
         return rhp.get_count(risk=risk) > 0
 
-    def build_hold_portfolio(self, day, risk: PortfoliosRisk):
+    def build_hold_portfolio(self, day, risk: PortfoliosRisk, force_mpt=False):
         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
         try:
             while start <= day:
-                logger.info(f'start to get normal portfolio for date[{format_date(start)}]')
-                portfolios = self._mpt.get_portfolios(day=prev_workday(start), type=PortfoliosType.NORMAL, risk=risk)
+                if force_mpt:
+                    logger.info(f'start to get normal portfolio for date[{format_date(start)}]')
+                    self._mpt.get_portfolios(day=prev_workday(start), type=PortfoliosType.NORMAL, risk=risk)
                 logger.info(f"start to build hold portfolio[{risk.name}] for date[{format_date(start)}]")
-                if portfolios:
+                signal = self._signal.get_signal(prev_workday(start), risk)
+                if signal:
                     last_re_date = self.get_last_rebalance_date(risk=risk, max_date=start)
                     # 两次实际调仓最小间隔期，单位交易日
                     if last_re_date and len(workday_range(last_re_date, start)) <= self.interval_days:
                         self.no_rebalance(start, risk, last_nav)
                     else:
-                        self.do_rebalance(start, risk, portfolios, last_nav)
+                        self.do_rebalance(start, risk, signal, last_nav)
                 else:
                     self.no_rebalance(start, risk, last_nav)
                 start = next_workday(start)
@@ -70,8 +77,8 @@ class DividendPortfoliosHolder(PortfoliosHolder):
         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()}
+    def do_rebalance(self, day, risk: PortfoliosRisk, signal, last_nav):
+        weight = {int(x[0]): x[1] for x in json.loads(signal['portfolio']).items()}
         dividend_acc = 0
         fund_dividend = 0
         if last_nav:
@@ -129,6 +136,7 @@ class DividendPortfoliosHolder(PortfoliosHolder):
         rhp.insert({
             'date': day,
             'risk': risk,
+            'signal_id': signal['id'],
             'dividend': dividend,
             'div_forecast': div_forecast if div_forecast else last_nav['div_forecast'] if last_nav else None,
             'fund_div': fund_dividend,
@@ -219,8 +227,8 @@ class DividendPortfoliosHolder(PortfoliosHolder):
 @component(bean_name='dividend-holder')
 class InvTrustPortfoliosHolder(DividendPortfoliosHolder):
 
-    def do_rebalance(self, day, risk: PortfoliosRisk, portfolio, last_nav):
-        weight = portfolio
+    def do_rebalance(self, day, risk: PortfoliosRisk, signal, last_nav):
+        weight = {int(x[0]): x[1] for x in json.loads(signal['portfolio']).items()}
         dividend_acc = 0
         fund_dividend = 0
         if last_nav:
@@ -255,6 +263,7 @@ class InvTrustPortfoliosHolder(DividendPortfoliosHolder):
         rhp.insert({
             'date': day,
             'risk': risk,
+            'signal_id': signal['id'],
             'dividend': 0,
             'fund_div': fund_dividend,
             'div_forecast': div_forecast if div_forecast else last_nav['div_forecast'] if last_nav else None,
@@ -360,6 +369,7 @@ class DailyHoldReportor(RoboReportor):
         return '每日持仓信息'
 
     def load_report(self, max_date=prev_workday(dt.today()), min_date=None) -> List[dict]:
+
         holds = pd.DataFrame(rhp.get_list(max_date=max_date, min_date=min_date))
         holds = holds[holds['date'].dt.date == max_date.date()]
         if not holds.empty:
@@ -379,6 +389,7 @@ class DailyHoldReportor(RoboReportor):
             holds['asset_ids'] = holds.apply(lambda row: datums.loc[int(row['portfolios'][0])]['ftTicker'], axis=1)
             holds['name'] = holds.apply(lambda row: datums.loc[int(row['portfolios'][0])]['chineseName'], axis=1)
             holds['lipper_id'] = holds.apply(lambda row: datums.loc[int(row['portfolios'][0])]['lipperKey'], axis=1)
-            holds = holds[['lipper_id', 'asset_ids', 'name', 'weight', 'risk', 'date', 'rebalance_type', 'rebalance_date']]
+            holds = holds[
+                ['lipper_id', 'asset_ids', 'name', 'weight', 'risk', 'date', 'rebalance_type', 'rebalance_date']]
             return holds.to_dict('records')
         return []

rebalance/base_signal.py

+import json
+from abc import ABC
+from datetime import datetime as dt
+from datetime import timedelta
+from functools import reduce
+from typing import List
+
+import pandas as pd
+from py_jftech import component, autowired, get_config, prev_workday
+from py_jftech import is_workday
+
+from api import PortfoliosBuilder
+from api import (
+    PortfoliosRisk, RebalanceSignal, SignalType, PortfoliosType, PortfoliosHolder,
+    RoboReportor, Datum, DatumType
+)
+from rebalance.dao import robo_rebalance_signal as rrs
+
+
+@component(bean_name='base-signal')
+class BaseRebalanceSignal(RebalanceSignal, ABC):
+
+    @autowired
+    def __init__(self, builder: PortfoliosBuilder = None):
+        self._builder = builder
+
+    def get_signal(self, day, risk: PortfoliosRisk):
+        signal = rrs.get_one(type=self.signal_type, risk=risk, date=day)
+        if signal:
+            return signal
+        trigger = self.need_rebalance(day, risk)
+        if trigger:
+            portfolio_type = self.portfolio_type
+            portfolio = self._builder.get_portfolios(day, risk, portfolio_type)
+            id = rrs.insert({
+                'date': day,
+                'type': self.signal_type,
+                'risk': risk,
+                'portfolio_type': portfolio_type,
+                'portfolio': portfolio
+            })
+            return rrs.get_by_id(id)
+        return None
+
+    def need_rebalance(self, day, risk: PortfoliosRisk) -> bool:
+        # 若记录为空则，将传入日期作为初始日期，进行build
+        signal = rrs.get_last_one(day, risk, SignalType.NORMAL, effective=None)
+        if signal:
+            frequency = get_config('portfolios')['holder']['warehouse-frequency']
+            date = pd.to_datetime(day.replace(day=1)) + pd.DateOffset(months=frequency)
+            date = date - timedelta(days=1)
+            # 指定周期末的工作日
+            date = date if is_workday(date) else prev_workday(date)
+            if date == day:
+                return True
+            elif signal['date'] == day:
+                return True
+            else:
+                return False
+        else:
+            return True
+
+    @property
+    def portfolio_type(self):
+        return self.signal_type.p_type
+
+    @property
+    def signal_type(self) -> SignalType:
+        return SignalType.NORMAL
+
+    def get_last_signal(self, day, risk: PortfoliosRisk):
+        last_re = rrs.get_last_one(max_date=day, risk=risk, effective=True)
+        return last_re
+
+
+@component(bean_name='signal-report')
+class SignalReportor(RoboReportor):
+
+    @autowired
+    def __init__(self, hold: PortfoliosHolder = None, datum: Datum = None):
+        self._hold = hold
+        self._datum = datum
+
+    @property
+    def report_name(self) -> str:
+        return '调仓信号'
+
+    def load_report(self, max_date=dt.today(), min_date=None) -> List[dict]:
+        result = []
+        datums = {str(x['id']): x for x in self._datum.get_datums(type=DatumType.FUND, exclude=False)}
+        for signal in rrs.get_list(max_date=max_date, min_date=min_date, effective=True):
+            rebalance_date = self._hold.get_rebalance_date_by_signal(signal['id'])
+            for fund_id, weight in json.loads(signal['portfolio']).items():
+                result.append({
+                    'risk': PortfoliosRisk(signal['risk']).name,
+                    'type': SignalType(signal['type']).name,
+                    'signal_date': signal['date'],
+                    'rebalance_date': rebalance_date,
+                    'portfolio_type': PortfoliosType(signal['portfolio_type']).name,
+                    'ft_ticker': datums[fund_id]['ftTicker'],
+                    'blooberg_ticker': datums[fund_id]['bloombergTicker'],
+                    'fund_name': datums[fund_id]['chineseName'],
+                    'weight': weight
+                })
+        return result
+
+
+@component(bean_name='daily-signal-report')
+class DailySignalReportor(RoboReportor):
+
+    @autowired
+    def __init__(self, hold: PortfoliosHolder = None, datum: Datum = None):
+        self._hold = hold
+        self._datum = datum
+
+    @property
+    def report_name(self) -> str:
+        return '每日调仓信号'
+
+    def load_report(self, max_date=prev_workday(dt.today()), min_date=None) -> List[dict]:
+        signals = pd.DataFrame(rrs.get_list(max_date=max_date, min_date=min_date))
+        signals = signals[(signals['date'].dt.date == max_date.date())]
+        if not signals.empty:
+            datum_ids = reduce(lambda x, y: x | y, signals['portfolio'].apply(lambda x: set(json.loads(x).keys())))
+            datums = pd.DataFrame(self._datum.get_datums(type=DatumType.FUND, datum_ids=datum_ids))
+            datums.set_index('id', inplace=True)
+
+            signals['risk'] = signals.apply(lambda row: PortfoliosRisk(row['risk']).name, axis=1)
+            signals['rebalance_type'] = signals.apply(lambda row: SignalType(row['type']).name, axis=1)
+            signals['portfolio_type'] = signals.apply(lambda row: PortfoliosType(row['portfolio_type']).name, axis=1)
+            signals['portfolio'] = signals.apply(lambda row: [x for x in json.loads(row['portfolio']).items()], axis=1)
+            signals = signals.explode('portfolio', ignore_index=True)
+
+            signals['weight'] = signals.apply(lambda row: format(row['portfolio'][1], '.0%'), axis=1)
+            signals['asset_ids'] = signals.apply(lambda row: datums.loc[int(row['portfolio'][0])]['ftTicker'], axis=1)
+            signals['name'] = signals.apply(lambda row: datums.loc[int(row['portfolio'][0])]['chineseName'], axis=1)
+            signals['lipper_id'] = signals.apply(lambda row: datums.loc[int(row['portfolio'][0])]['lipperKey'], axis=1)
+
+            signals = signals[['lipper_id', 'asset_ids', 'name', 'weight', 'risk', 'date', 'rebalance_type']]
+            return signals.to_dict('records')
+
+        return []

rebalance/dao/mysql.sql

+CREATE TABLE IF NOT EXISTS robo_rebalance_signal
+(
+    rrs_id          BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
+    rrs_date        DATETIME        NOT NULL COMMENT '信号日期',
+    rrs_type        TINYINT         NOT NULL COMMENT '信号类型',
+    rrs_risk        TINYINT         NOT NULL COMMENT '风险等级',
+    rrs_p_type      VARCHAR(255)             DEFAULT NULL COMMENT '投组类型',
+    rrs_p_weight    JSON                     DEFAULT NULL COMMENT '投组信息',
+    rrs_effective   TINYINT         NOT NULL DEFAULT 0 COMMENT '是否生效',
+    rrs_create_time DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    rrs_update_time DATETIME                 DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
+    PRIMARY KEY (rrs_id),
+    INDEX (rrs_date),
+    INDEX (rrs_type),
+    INDEX (rrs_risk)
+) ENGINE = InnoDB
+  AUTO_INCREMENT = 0
+  DEFAULT CHARSET = utf8mb4 COMMENT '再平衡信号表';
+
+

rebalance/dao/robo_rebalance_signal.py

+from py_jftech import read, write, where, format_date, mapper_columns, to_tuple
+
+from api import SignalType, PortfoliosRisk
+
+__COLUMNS__ = {
+    'rrs_id': 'id',
+    'rrs_date': 'date',
+    'rrs_type': 'type',
+    'rrs_risk': 'risk',
+    'rrs_p_type': 'portfolio_type',
+    'rrs_p_weight': 'portfolio',
+    'rrs_effective': 'effective',
+}
+
+
+@read
+def get_list(min_date=None, max_date=None, risk: PortfoliosRisk = None, type: SignalType = None, effective: bool = None):
+    sqls = []
+    if min_date:
+        sqls.append(f"rrs_date >= '{format_date(min_date)}'")
+    if max_date:
+        sqls.append(f"rrs_date <= '{format_date(max_date)}'")
+    return f'''
+    select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_rebalance_signal
+    {where(*sqls, rrs_risk=risk, rrs_type=type, rrs_effective=effective)} order by rrs_risk, rrs_date
+    '''
+
+
+@read
+def get_by_ids(ids):
+    return f'''select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_rebalance_signal {where(rrs_id=to_tuple(ids))}'''
+
+
+@read(one=True)
+def get_by_id(id):
+    return f'''select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_rebalance_signal {where(rrs_id=id)}'''
+
+
+@read(one=True)
+def get_one(type: SignalType, risk: PortfoliosRisk, date):
+    return f'''
+    select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_rebalance_signal
+    {where(rrs_date=date, rrs_type=type, rrs_risk=risk)}
+    '''
+
+
+@read(one=True)
+def get_first_after(type: SignalType, risk: PortfoliosRisk, min_date, effective=None):
+    return f'''
+    select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_rebalance_signal
+    {where(f"rrs_date >= '{format_date(min_date)}'", rrs_type=type, rrs_risk=risk, rrs_effective=effective)} order by rrs_date limit 1
+    '''
+
+
+@read(one=True)
+def get_last_one(max_date, risk: PortfoliosRisk, type: SignalType = None, effective=None):
+    return f'''
+    select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_rebalance_signal
+    {where(f"rrs_date <= '{format_date(max_date)}'", rrs_type=type, rrs_risk=risk, rrs_effective=effective)} order by rrs_date desc limit 1
+    '''
+
+
+def get_count(risk: PortfoliosRisk = None, day=None, effective=None):
+    @read(one=True)
+    def exec():
+        return f"select count(*) as `count` from robo_rebalance_signal {where(rrs_risk=risk, rrs_date=day, rrs_effective=effective)}"
+
+    result = exec()
+    return result['count']
+
+
+@write
+def insert(datas):
+    datas = mapper_columns(datas=datas, columns=__COLUMNS__)
+    return f'''
+    insert into robo_rebalance_signal({','.join([x for x in datas.keys()])})
+    values ({','.join([f"'{x[1]}'" for x in datas.items()])})
+    '''
+
+
+@write
+def update(id, datas):
+    datas = mapper_columns(datas=datas, columns=__COLUMNS__)
+    return f'''
+    update robo_rebalance_signal 
+    set {','.join([f"{x[0]} = '{x[1]}'" for x in datas.items()])}
+    where rrs_id = {id}
+    '''
+
+
+@write
+def delete_by_id(id):
+    return f"delete from robo_rebalance_signal where rrs_id = {id}"
+
+
+@write
+def delete(min_date=None, risk: PortfoliosRisk = None):
+    if min_date is None and risk is None:
+        return 'truncate table robo_rebalance_signal'
+    else:
+        sql = f"rrs_date >= '{format_date(min_date)}'" if min_date else None
+        return f"delete from robo_rebalance_signal {where(sql, rrs_risk=risk)}"

requirements.txt

@@ -9,7 +9,7 @@ openpyxl==3.0.10
 pandas==1.5.1
 pandas-datareader==0.10.0
 ply==3.11
-PyJFTech==1.2.0
+PyJFTech==1.3.0
 PyMySQL==1.0.2
 Pyomo==6.4.3
 python-dateutil==2.8.2
@@ -21,4 +21,4 @@ scipy==1.9.3
 six==1.16.0
 urllib3==1.26.12
 fastapi==0.100.0
-uvicorn==0.23.1
\ No newline at end of file
+uvicorn==0.23.1

robo_executor.py

@@ -12,7 +12,7 @@ from py_jftech import (
 
 from api import (
     RoboExecutor, Datum, AssetPool, PortfoliosBuilder,
-    PortfoliosRisk, PortfoliosHolder, DataSync, RoboExportor, BacktestStep
+    PortfoliosRisk, PortfoliosHolder, DataSync, RoboExportor, BacktestStep, RebalanceSignal
 )
 
 logger = logging.getLogger(__name__)
@@ -135,7 +135,8 @@ class RealExecutor(RoboExecutor):
 
     @autowired(names={'daily_export': 'daily-real-export', 'monitor_export': 'daily-monitor-export'})
     def __init__(self, builder: PortfoliosBuilder = None, hold: PortfoliosHolder = None, syncs: List[DataSync] = None,
-                 daily_export: RoboExportor = None, monitor_export: RoboExportor = None, pool: AssetPool = None, ):
+                 daily_export: RoboExportor = None, monitor_export: RoboExportor = None, pool: AssetPool = None,
+                 signal: RebalanceSignal = None):
         self._builder = builder
         self._pool = pool
         self._hold = hold
@@ -143,6 +144,8 @@ class RealExecutor(RoboExecutor):
         self._daily_export = daily_export
         self._monitor_export = monitor_export
         self._config = get_config(__name__)['real']
+        self._signal = signal
+
 
     @property
     def start_date(self):
@@ -179,6 +182,9 @@ class RealExecutor(RoboExecutor):
             for risk in PortfoliosRisk:
                 logger.info(f"start to build risk[{risk.name}] real for date[{format_date(date)}]".center(50, '-'))
                 now = dt.now()
+                # 因为每天都必须有NORMAL最优投组，不管用不用
+                self._builder.get_portfolios(date, risk)
+                self._signal.get_signal(date, risk)
                 # 更新持仓
                 self._hold.build_hold_portfolio(date, risk)
                 logger.info(

web/robo_controller.py

@@ -21,4 +21,5 @@ async def root():
 
 
 if __name__ == "__main__":
+    # 开启定时任务，执行实盘
     uvicorn.run("robo_controller:app", reload=True, port=8080)