调整信号规则逻辑

api.py

@@ -50,6 +50,7 @@ class SolveType(Enum):
 
 @unique
 class SignalType(Enum):
+    NONE = -1
     INIT = 0
     CRISIS_EXP = 1
     CRISIS_ONE = 2
@@ -62,7 +63,7 @@ class SignalType(Enum):
 
 # 信号处理优先级
 SignalType.CRISIS_ONE.level = 1
-SignalType.CRISIS_TWO.level = 2
+SignalType.CRISIS_TWO.level = 1
 SignalType.MARKET_RIGHT.level = 3
 SignalType.HIGH_BUY.level = 4
 SignalType.LOW_BUY.level = 5
@@ -519,11 +520,12 @@ class RebalanceRuler(ABC):
     '''
 
     @abstractmethod
-    def take_next_signal(self, day, risk: PortfoliosRisk):
+    def take_next_signal(self, day, risk: PortfoliosRisk, only_today=True):
         '''
         取出指定日期，指定风险等级的再平衡信号数据，注意取出消费后，无法退回，非幂等函数
         :param day: 指定日期
         :param risk: 指定风险等级
+        :param only_today: 只获取今天是否有信号
         :return: 如果存在，则返回取出的再平衡信号信息，否则返回None
         '''
         pass

portfolios/holder.py

@@ -59,8 +59,8 @@ class NextReblanceHolder(PortfoliosHolder):
                 logger.info(f"start to build hold portfolio[{risk.name}] for date[{format_date(start)}]")
                 signal = None
                 if last_nav:
-                    last_re = rhp.get_last_one(max_date=start, risk=risk, rebalance=True)
-                    if len(workday_range(last_re['date'], start)) > self.interval_days:
+                    last_re_date = self.get_last_rebalance_date(risk=risk, max_date=start)
+                    if len(workday_range(last_re_date, start)) > self.interval_days:
                         signal = self._rule.take_next_signal(prev_workday(start), risk)
                 else:
                     signal = self._rule.take_next_signal(prev_workday(start), risk)

portfolios/test_case.py

@@ -29,8 +29,7 @@ class PortfoliosTest(unittest.TestCase):
 
     @autowired(names={'hold': 'next-re'})
     def test_build_hold(self, hold: PortfoliosHolder = None):
-        hold.build_hold_portfolio(parse_date('2016-01-01'), PortfoliosRisk.FT9)
-        pass
+        hold.build_hold_portfolio(parse_date('2022-11-01'), PortfoliosRisk.FT9)
 
     @autowired(names={'reportor': 'hold-report'})
     def test_hold_report(self, reportor: RoboReportor = None):

rebalance/dao/robo_rebalance_signal.py

@@ -88,6 +88,11 @@ def update(id, datas):
     '''
 
 
+@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:

rebalance/ruler.py

@@ -39,35 +39,49 @@ class LevelRebalanceRuler(RebalanceRuler):
         else:
             return {t: result for t in PortfoliosType}
 
-    def take_next_signal(self, day, risk: PortfoliosRisk):
+    def without_disable_period(self, day, risk: PortfoliosRisk) -> bool:
+        last_re = rrs.get_last_one(max_date=day, risk=risk, effective=True)
+        if SignalType(last_re['type']).p_type in self.disable_period:
+            return len(workday_range(last_re['date'], day)) > self.disable_period[SignalType(last_re['type']).p_type]
+        return False
+
+    def take_next_signal(self, day, risk: PortfoliosRisk, only_today=True):
         last_re = rrs.get_last_one(max_date=day, risk=risk, effective=True)
         if not last_re:
             builder = [x for x in self._signals if x.signal_type is SignalType.INIT][0]
             return builder.get_signal(day, risk)
 
-        long_signals = [x for x in self._signals if
-                        x.signal_type.p_type is not PortfoliosType.NORMAL and x.signal_type.level < SignalType(last_re['type']).level]
-        for long_signal in sorted(long_signals, key=lambda x: x.signal_type.level):
-            workdays = workday_range(next_workday(last_re['date']), day)
-            if len(workdays) <= self._hold.interval_days:
-                for date in workdays:
-                    signal = long_signal.get_signal(date, risk)
-                    if signal:
-                        return signal
-            else:
-                signal = long_signal.get_signal(day, risk)
-                if signal:
-                    return signal
-        if SignalType(last_re['type']).p_type in self.disable_period:
-            re_date = self._hold.get_last_rebalance_date(risk=risk, max_date=day)
-            if re_date:
-                workdays = workday_range(re_date, day)
-                if len(workdays) < self.disable_period[SignalType(last_re['type']).p_type]:
-                    return None
-        for temp_signal in sorted([x for x in self._signals if x.signal_type.p_type is PortfoliosType.NORMAL], key=lambda x: x.signal_type.level):
-            signal = temp_signal.get_signal(day, risk)
-            if signal:
-                return signal
+        risk_signals = [x for x in self._signals if x.signal_type.p_type is not PortfoliosType.NORMAL]
+        buy_signals = [x for x in self._signals if x.signal_type.p_type is PortfoliosType.NORMAL]
+
+        last_signal = rrs.get_last_one(max_date=day, risk=risk)
+        start = next_workday(last_signal['date'])
+
+        while start <= day:
+            # 检查风控信号
+            signals = {x.signal_type: x.get_signal(start, risk) for x in risk_signals if x.signal_type.level <= SignalType(last_re['type']).level}
+            signals = {x[0]: x[1] for x in signals.items() if x[1] is not None}
+            # 上次实际调仓类型为危机信号，本次危机信号不调仓
+            if signals and SignalType(last_re['type']) in [SignalType.CRISIS_ONE, SignalType.CRISIS_TWO]:
+                signals = {x[0]: x[1] for x in signals.items() if x[0] not in [SignalType.CRISIS_ONE, SignalType.CRISIS_TWO]}
+            # 检查买入信号，只有当天需要检查
+            if not signals and start == day and self.without_disable_period(day, risk):
+                signals = {x.signal_type: x.get_signal(start, risk) for x in buy_signals}
+                signals = {x[0]: x[1] for x in signals.items() if x[1] is not None}
+            if signals:
+                if SignalType(last_signal['type']) is SignalType.NONE:
+                    rrs.delete_by_id(last_signal['id'])
+                return signals[sorted(signals.keys(), key=lambda x: x.level)[0]]
+            start = next_workday(start)
+
+        if SignalType(last_signal['type']) is SignalType.NONE:
+            rrs.update(last_signal['id'], {'date': day})
+        else:
+            rrs.insert({
+                'date': day,
+                'type': SignalType.NONE,
+                'risk': risk
+            })
         return None
 
     def get_signal_type(self, sign_id) -> SignalType | Dict[int, SignalType]:

rebalance/test_case.py

@@ -23,7 +23,7 @@ class RebalanceTest(unittest.TestCase):
 
     @autowired(names={'builder': 'crisis_two'})
     def test_crisis_two(self, builder: RebalanceSignal = None):
-        start = parse_date('2008-01-22')
+        start = parse_date('2020-04-29')
         end = start + relativedelta(years=3)
         while start < end:
             signal = builder.is_trigger(start, PortfoliosRisk.FT9)
@@ -47,7 +47,7 @@ class RebalanceTest(unittest.TestCase):
 
     @autowired
     def test_rebalance_builder(self, builder: RebalanceRuler = None):
-        builder.take_next_signal(parse_date('2022-09-01'), PortfoliosRisk.FT3)
+        builder.take_next_signal(parse_date('2020-04-29'), PortfoliosRisk.FT9)
 
     @autowired(names={'reportor': 'signal-report'})
     def test_signal_report(self, reportor: RoboReportor = None):