依赖注入实现中

api.py

@@ -55,6 +55,24 @@ class SignalType(Enum):
     DRIFT_BUY = 7
 
 
+# 信号处理优先级
+SignalType.CRISIS_ONE.level = 1
+SignalType.CRISIS_TWO.level = 2
+SignalType.MARKET_RIGHT.level = 3
+SignalType.HIGH_BUY.level = 4
+SignalType.LOW_BUY.level = 5
+SignalType.DRIFT_BUY.level = 5
+SignalType.INIT.level = 6
+# 对应需要再平衡的投组类型
+SignalType.CRISIS_ONE.p_type = PortfoliosType.CRISIS_1
+SignalType.CRISIS_TWO.p_type = PortfoliosType.CRISIS_2
+SignalType.MARKET_RIGHT.p_type = PortfoliosType.RIGHT_SIDE
+SignalType.HIGH_BUY.p_type = PortfoliosType.NORMAL
+SignalType.LOW_BUY.p_type = PortfoliosType.NORMAL
+SignalType.DRIFT_BUY.p_type = PortfoliosType.NORMAL
+SignalType.INIT.p_type = PortfoliosType.RIGHT_SIDE
+
+
 class Datum(ABC):
     '''
     基础资料服务，基金资料数据，各种指数，指标资料数据
@@ -287,6 +305,7 @@ class SolverFactory(ABC):
     '''
     解算器工厂
     '''
+
     @abstractmethod
     def create_solver(self, risk: PortfoliosRisk, type: PortfoliosType = PortfoliosType.NORMAL) -> Solver:
         '''
@@ -302,6 +321,7 @@ class PortfoliosHolder(ABC):
     '''
     投资组合持仓器
     '''
+
     @abstractmethod
     def get_portfolios_weight(self, day, risk: PortfoliosRisk):
         '''
@@ -329,3 +349,28 @@ class SignalBuilder(ABC):
     @abstractmethod
     def get_signal(self, day, risk: PortfoliosRisk):
         pass
+
+    @property
+    @abstractmethod
+    def signal_type(self) -> SignalType:
+        '''
+        返回信号类型
+        :return: 信号类型
+        '''
+        pass
+
+
+class RebalanceBuilder(ABC):
+    '''
+    再平衡构建器
+    '''
+
+    @abstractmethod
+    def get_rebalance(self, day, risk: PortfoliosRisk):
+        '''
+        获取指定日期，指定风险等级的再平衡数据
+        :param day: 指定日期
+        :param risk: 指定风险等级
+        :return: 再平衡数据
+        '''
+        pass

config.yml

@@ -96,7 +96,7 @@ portfolios:
       high-weight: [ 1, 0.6, 0.35 ]
     poem:
       cvar-scale-factor: 0.1
-    right-side:
+    right_side:
       navs:
         risk: [1, 2]
         exclude-asset-type: ['STOCK', 'BALANCED']
@@ -113,30 +113,37 @@ portfolios:
       mpt:
         quantile: 0.1
 rebalance:
-  init-signal:
-    date: 2022-09-01
-  crisis-signal:
-    exp-years: 3
-    exp-init: 2022-03-04
-    inversion-years: 1
-    inversion-threshold: 0.3
-    crisis-1:
-      mean-count: 850
-      consecut-days: 5
-    crisis-2:
-      negative-growth: 1
-      fed-months: 3
-      fed-threshold: 0.75
-  right-side:
-    rtn-days: 5
-    min-threshold: -0.05
-    coef: 0.95
-  curve-drift:
-    diff-threshold: 0.4
-    init-factor: 0.000000002
-  high-low-buy:
-    drift-coef: 0.2
-    threshold: [0.5, 0.8]
+  builder:
+    disable-period: #自然日
+      normal: 10
+      crisis_1: 15
+      crisis_2: 15
+      right_side: 15
+  signals:
+    init-signal:
+      date: 2022-09-01
+    crisis-signal:
+      exp-years: 3
+      exp-init: 2022-03-04
+      inversion-years: 1
+      inversion-threshold: 0.3
+      crisis-1:
+        mean-count: 850
+        consecut-days: 5
+      crisis-2:
+        negative-growth: 1
+        fed-months: 3
+        fed-threshold: 0.75
+    right-side:
+      rtn-days: 5
+      min-threshold: -0.05
+      coef: 0.95
+    curve-drift:
+      diff-threshold: 0.4
+      init-factor: 0.000000002
+    high-low-buy:
+      drift-coef: 0.2
+      threshold: [ 0.5, 0.8 ]
 
 
 

portfolios/holder.py

 from framework import component, autowired
-from api import PortfoliosHolder, PortfoliosRisk
+from api import PortfoliosHolder, PortfoliosRisk, RebalanceBuilder
 from portfolios.dao import robo_hold_portfolios as rhp
 import json
 
@@ -7,6 +7,10 @@ import json
 @component(bean_name='next-re')
 class NextReblanceHolder(PortfoliosHolder):
 
+    @autowired
+    def __init__(self, rebalance: RebalanceBuilder):
+        self._rebalance = rebalance
+
     def get_portfolios_weight(self, day, risk: PortfoliosRisk):
         hold = rhp.get_one(day, risk)
         if hold:
@@ -15,4 +19,8 @@ class NextReblanceHolder(PortfoliosHolder):
         return None
 
     def has_hold(self, risk: PortfoliosRisk) -> bool:
-        return rhp.get_count(risk=risk) > 0
+        return rhp.get_count(risk=risk) > 0
\ No newline at end of file
+
+    def build_hold_portfolio(self, day, risk: PortfoliosRisk):
+
+        pass
\ No newline at end of file

rebalance/init_signal.py → rebalance/base_signal.py

-from api import SignalBuilder, PortfoliosRisk, SignalType, PortfoliosBuilder
-from framework import component, autowired
+from abc import ABC, abstractmethod
+
+from api import SignalBuilder, PortfoliosBuilder, PortfoliosRisk
+from framework import autowired
 from rebalance.dao import robo_rebalance_signal as rrs
 
 
-@component(bean_name='init')
-class InitSignalBuilder(SignalBuilder):
+class BaseSignalBuilder(SignalBuilder, ABC):
 
     @autowired
     def __init__(self, builder: PortfoliosBuilder = None):
         self._builder = builder
 
     def get_signal(self, day, risk: PortfoliosRisk):
-        if rrs.get_count(risk) == 0:
-            portfolio = self._builder.get_portfolios(day, risk, PortfoliosType.RIGHT_SIDE)
+        trigger = self.is_trigger(day, risk)
+        if trigger:
+            portfolio = self._builder.get_portfolios(day, risk, self.signal_type.p_type)
             id = rrs.insert({
                 'date': day,
-                'type': SignalType.INIT,
+                'type': self.signal_type,
                 'risk': risk,
-                'portfolio_type': PortfoliosType.RIGHT_SIDE,
+                'portfolio_type': self.signal_type.p_type,
                 'portfolio': portfolio
             })
             return rrs.get_by_id(id)
         return None
+
+    @abstractmethod
+    def is_trigger(self, day, risk: PortfoliosRisk) -> bool:
+        pass

rebalance/builder.py

+from framework import component, autowired, get_config
+from api import RebalanceBuilder, PortfoliosRisk, SignalBuilder, SignalType, PortfoliosType
+from typing import List
+from rebalance.dao import robo_rebalance_signal as rrs
+
+
+@component
+class LevelRebalanceBuilder(RebalanceBuilder):
+
+    @autowired
+    def __init__(self, signals: List[SignalBuilder] = None):
+        self._signals = signals
+        self._config = get_config(__name__)
+
+    @property
+    def disable_period(self):
+        result = self._config['disable-period']
+        return {PortfoliosType(x[0]): x[1] for x in result.items()}
+
+    def get_rebalance(self, day, risk: PortfoliosRisk):
+        last_re = rrs.get_last_one(max_date=day, risk=risk, effective=True)
+        if last_re and last_re['date'] == day:
+            return last_re
+        if last_re:
+            disable_period = self.disable_period[PortfoliosType(last_re['p_type'])]
+        signals = [x.get_signal(day, risk) for x in self._signals]
+        signals = sorted([x for x in signals if x is not None], key=lambda x: SignalType(x['type']).level)
+        if signals:
+            use_signal = signals[0]
+            rrs.update(use_signal['id'], {'effective': True})
+            return use_signal
+        return None
+
+

rebalance/dao/robo_rebalance_signal.py

@@ -37,24 +37,39 @@ def get_last_one(max_date, risk: PortfoliosRisk, type: SignalType = None, effect
     '''
 
 
-def get_count(risk: PortfoliosRisk):
+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)}"
+        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):
+def format_datas(datas):
     datas = {x[0]: datas[x[1]] for x in __COLUMNS__.items() if x[1] in datas and datas[x[1]] is not None}
-    datas = {
+    return {
         **datas,
         **{x[0]: format_date(x[1]) for x in datas.items() if isinstance(x[1], datetime)},
         **{x[0]: x[1].value for x in datas.items() if isinstance(x[1], Enum)},
         **{x[0]: json.dumps(x[1]) for x in datas.items() if isinstance(x[1], dict)},
         **{x[0]: (1 if x[1] else 0) for x in datas.items() if isinstance(x[1], bool)}
     }
+
+
+@write
+def insert(datas):
+    datas = format_datas(datas)
     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, dates):
+    datas = format_datas(dates)
+    return f'''
+    update robo_rebalance_signal 
+    set {','.join([f"{x[0]} = '{x[1]}'" for x in datas.items()])}
+    where rrs_id = {id}
+    '''

rebalance/crisis_signal.py → rebalance/signals/crisis_signal.py

@@ -3,15 +3,17 @@ from abc import ABC
 import pandas as pd
 from dateutil.relativedelta import relativedelta
 
-from api import SignalBuilder, PortfoliosRisk, SignalType, Navs, PortfoliosBuilder, PortfoliosType
+from api import PortfoliosRisk, SignalType, Navs
 from framework import get_config, autowired, component
+from rebalance.base_signal import BaseSignalBuilder
 from rebalance.dao import robo_rebalance_signal as rrs
 
 
-class CrisisSignal(SignalBuilder, ABC):
+class CrisisSignal(BaseSignalBuilder, ABC):
 
     @autowired
     def __init__(self, navs: Navs = None):
+        super().__init__()
         self._navs = navs
         self._config = get_config(__name__)
 
@@ -56,12 +58,7 @@ class CrisisSignal(SignalBuilder, ABC):
 
 
 @component(bean_name='crisis_one')
-class CrisisOneSignal(CrisisSignal):
-
-    @autowired
-    def __init__(self, builder: PortfoliosBuilder = None):
-        super(CrisisOneSignal, self).__init__()
-        self._builder = builder
+class CrisisOneSignal(CrisisSignal, BaseSignalBuilder):
 
     @property
     def consecut_days(self):
@@ -71,33 +68,23 @@ class CrisisOneSignal(CrisisSignal):
     def mean_count(self):
         return self._config['crisis-1']['mean-count']
 
-    def get_signal(self, day, risk: PortfoliosRisk):
+    @property
+    def signal_type(self):
+        return SignalType.CRISIS_ONE
+
+    def is_trigger(self, day, risk: PortfoliosRisk) -> bool:
         exp_date = self.get_exp_start_date(day, risk)
         if exp_date:
             crisis_one = rrs.get_first_after(type=SignalType.CRISIS_ONE, risk=risk, min_date=exp_date)
             if not crisis_one:
                 spx = self._navs.get_last_index_close(max_date=day, ticker='SPX Index', count=self.mean_count)
                 spx_ma850 = pd.DataFrame(spx).close.mean()
-                if len([x for x in spx[0:5] if x['close'] > spx_ma850]) == 0:
-                    portfolio = self._builder.get_portfolios(day, risk, PortfoliosType.CRISIS_1)
-                    id = rrs.insert({
-                        'date': day,
-                        'type': SignalType.CRISIS_ONE,
-                        'risk': risk,
-                        'portfolio_type': PortfoliosType.CRISIS_1,
-                        'portfolio': portfolio
-                    })
-                    return rrs.get_by_id(id)
-        return None
+                return len([x for x in spx[0:5] if x['close'] > spx_ma850]) == 0
+        return False
 
 
 @component(bean_name='crisis_two')
-class CrisisTwoSignal(CrisisSignal):
-
-    @autowired
-    def __init__(self, builder: PortfoliosBuilder = None):
-        super(CrisisSignal, self).__init__()
-        self._builder = builder
+class CrisisTwoSignal(CrisisSignal, BaseSignalBuilder):
 
     @property
     def negative_growth_years(self):
@@ -111,7 +98,11 @@ class CrisisTwoSignal(CrisisSignal):
     def fed_threshold(self):
         return self._config['crisis-2']['fed-threshold']
 
-    def get_signal(self, day, risk: PortfoliosRisk):
+    @property
+    def signal_type(self):
+        return SignalType.CRISIS_TWO
+
+    def is_trigger(self, day, risk: PortfoliosRisk) -> bool:
         exp_date = self.get_exp_start_date(day, risk)
         if exp_date:
             crisis_two = rrs.get_first_after(type=SignalType.CRISIS_TWO, risk=risk, min_date=exp_date)
@@ -124,21 +115,8 @@ class CrisisTwoSignal(CrisisSignal):
                 before = ten_before['close'] - cpi_before['close']
                 today = ten_today['close'] - cpi_today['close']
 
-                fed_today = self._navs.get_last_index_close(max_date=day, ticker='FDTR_Index')
-                fed_before = self._navs.get_last_index_close(max_date=day-relativedelta(months=self.fed_months), ticker='FDTR_Index')
-
-                if today <= before and fed_today['close'] - fed_before['close'] < -self.fed_threshold:
-                    portfolio = self._builder.get_portfolios(day, risk, PortfoliosType.CRISIS_2)
-                    id = rrs.insert({
-                        'date': day,
-                        'type': SignalType.CRISIS_TWO,
-                        'risk': risk,
-                        'portfolio_type': PortfoliosType.CRISIS_2,
-                        'portfolio': portfolio
-                    })
-                    return rrs.get_by_id(id)
-        return None
-
-
-
+                fed_today = self._navs.get_last_index_close(max_date=day, ticker='FDTR Index')
+                fed_before = self._navs.get_last_index_close(max_date=day - relativedelta(months=self.fed_months), ticker='FDTR Index')
 
+                return today <= before and fed_today['close'] - fed_before['close'] < -self.fed_threshold
+        return False

rebalance/curve_drift.py → rebalance/signals/curve_drift.py

+from api import PortfoliosRisk, SignalType, Datum, PortfoliosHolder
 from framework import component, autowired, get_config
-from api import SignalBuilder, PortfoliosRisk, SignalType, Datum, PortfoliosBuilder, PortfoliosHolder
+from rebalance.base_signal import BaseSignalBuilder
 from rebalance.dao import robo_rebalance_signal as rrs
 
 
 @component(bean_name='curve-drift')
-class CurveDrift(SignalBuilder):
+class CurveDrift(BaseSignalBuilder):
 
     @autowired
-    def __init__(self, datum: Datum = None, builder: PortfoliosBuilder = None, hold: PortfoliosHolder = None):
+    def __init__(self, datum: Datum = None, hold: PortfoliosHolder = None):
+        super().__init__()
         self._datum = datum
-        self._builder = builder
         self._hold = hold
         self._config = get_config(__name__)
 
@@ -23,7 +24,7 @@ class CurveDrift(SignalBuilder):
             SignalType.LOW_BUY
         ]
 
-    def get_signal(self, day, risk: PortfoliosRisk):
+    def is_trigger(self, day, risk: PortfoliosRisk) -> bool:
         last_re = rrs.get_last_one(max_date=day, risk=risk, effective=True)
         if last_re is None or SignalType(last_re['type']) in self.exclude_last_type:
             return None
@@ -34,17 +35,7 @@ class CurveDrift(SignalBuilder):
         hold_portfolio = self._hold.get_portfolios_weight(day, risk)
         hold_weight = round(sum([x[1] for x in normal_portfolio.items() if x[0] in datum_ids]), 2)
         threshold = self.diff_threshold - self.init_factor * (day - last_re['date']).days ** 4
-        if normal_weight - hold_weight >= max(0, threshold):
-            portfolio = self._builder.get_portfolios(day, risk)
-            id = rrs.insert({
-                'date': day,
-                'type': SignalType.DRIFT_BUY,
-                'risk': risk,
-                'portfolio_type': PortfoliosType.NORMAL,
-                'portfolio': portfolio
-            })
-            return rrs.get_by_id(id)
-        return None
+        return normal_weight - hold_weight >= max(0, threshold)
 
     @property
     def diff_threshold(self):
@@ -54,5 +45,6 @@ class CurveDrift(SignalBuilder):
     def init_factor(self):
         return self._config['init-factor']
 
-
-
+    @property
+    def signal_type(self) -> SignalType:
+        return SignalType.DRIFT_BUY

rebalance/high_low_buy.py → rebalance/signals/high_low_buy.py

 import pandas as pd
 
-from framework import component, autowired, get_config, filter_weekend, next_workday, parse_date, is_workday
-from api import PortfoliosBuilder, SignalType, PortfoliosRisk, SignalBuilder, Datum
+from api import PortfoliosBuilder, SignalType, PortfoliosRisk, Datum
+from framework import component, autowired, get_config, filter_weekend, next_workday, is_workday
+from rebalance.base_signal import BaseSignalBuilder
 from rebalance.dao import robo_weight_drift as rwd, robo_rebalance_signal as rrs
-from datetime import timedelta, datetime as dt
 
 
 def get_start_date():
@@ -51,7 +51,7 @@ class DriftSupport:
 
 
 @component(bean_name='high-buy')
-class HighBuySignal(SignalBuilder, DriftSupport):
+class HighBuySignal(BaseSignalBuilder, DriftSupport):
 
     @property
     def include_last_type(self):
@@ -62,27 +62,21 @@ class HighBuySignal(SignalBuilder, DriftSupport):
             SignalType.LOW_BUY
         ]
 
-    def get_signal(self, day, risk: PortfoliosRisk):
+    @property
+    def signal_type(self) -> SignalType:
+        return SignalType.HIGH_BUY
+
+    def is_trigger(self, day, risk: PortfoliosRisk) -> bool:
         last_re = rrs.get_last_one(max_date=day, risk=risk, effective=True)
         if last_re is None or SignalType(last_re['type']) not in self.include_last_type:
-            return None
+            return False
         drift = self.get_drift(day, risk)
         threshold = self.get_threshold(risk)
-        if drift > threshold[1]:
-            portfolio = self._builder.get_portfolios(date, risk)
-            id = rrs.insert({
-                'date': day,
-                'type': SignalType.HIGH_BUY,
-                'risk': risk,
-                'portfolio_type': PortfoliosType.NORMAL,
-                'portfolio': portfolio
-            })
-            return rrs.get_by_id(id)
-        return None
+        return drift > threshold[1]
 
 
 @component(bean_name='low-buy')
-class LowBuySignal(SignalBuilder, DriftSupport):
+class LowBuySignal(BaseSignalBuilder, DriftSupport):
 
     @property
     def include_last_type(self):
@@ -92,20 +86,14 @@ class LowBuySignal(SignalBuilder, DriftSupport):
             SignalType.MARKET_RIGHT
         ]
 
-    def get_signal(self, day, risk: PortfoliosRisk):
+    @property
+    def signal_type(self) -> SignalType:
+        return SignalType.LOW_BUY
+
+    def is_trigger(self, day, risk: PortfoliosRisk) -> bool:
         last_re = rrs.get_last_one(max_date=day, risk=risk, effective=True)
         if last_re is None or SignalType(last_re['type']) not in self.include_last_type:
-            return None
+            return False
         drift = self.get_drift(day, risk)
         threshold = self.get_threshold(risk)
-        if  threshold[0] < drift < threshold[1]:
-            portfolio = self._builder.get_portfolios(date, risk)
-            id = rrs.insert({
-                'date': day,
-                'type': SignalType.LOW_BUY,
-                'risk': risk,
-                'portfolio_type': PortfoliosType.NORMAL,
-                'portfolio': portfolio
-            })
-            return rrs.get_by_id(id)
-        return None
+        return threshold[0] < drift < threshold[1]

rebalance/signals/init_signal.py

+from api import PortfoliosRisk, SignalType
+from framework import component
+from rebalance.base_signal import BaseSignalBuilder
+from rebalance.dao import robo_rebalance_signal as rrs
+
+
+@component(bean_name='init')
+class InitSignalBuilder(BaseSignalBuilder):
+
+    @property
+    def signal_type(self) -> SignalType:
+        return SignalType.INIT
+
+    def is_trigger(self, day, risk: PortfoliosRisk) -> bool:
+        return rrs.get_count(risk=risk) == 0

rebalance/right_side.py → rebalance/signals/right_side.py

 import pandas as pd
+from scipy.stats import norm
 
+from api import SignalType, PortfoliosRisk, Navs
 from framework import component, autowired, get_config
-from api import SignalBuilder, SignalType, PortfoliosRisk, Navs, PortfoliosBuilder, PortfoliosType
+from rebalance.base_signal import BaseSignalBuilder
 from rebalance.dao import robo_rebalance_signal as rrs
-from scipy.stats import norm
 
 
 @component(bean_name='market-right')
-class MarketRight(SignalBuilder):
+class MarketRight(BaseSignalBuilder):
 
     @autowired
-    def __init__(self, navs: Navs = None, builder: PortfoliosBuilder = None):
+    def __init__(self, navs: Navs = None):
+        super().__init__()
         self._navs = navs
-        self._builder = builder
         self._config = get_config(__name__)
 
     @property
@@ -27,25 +28,19 @@ class MarketRight(SignalBuilder):
     def coef(self):
         return self._config['coef']
 
-    def get_signal(self, day, risk: PortfoliosRisk):
+    @property
+    def signal_type(self) -> SignalType:
+        return SignalType.MARKET_RIGHT
+
+    def is_trigger(self, day, risk: PortfoliosRisk) -> bool:
         last_re = rrs.get_last_one(risk=risk, max_date=day, effective=True)
         if last_re is not None and SignalType(last_re['type']) in [SignalType.CRISIS_ONE, SignalType.CRISIS_TWO, SignalType.MARKET_RIGHT]:
-            return None
+            return False
         spx = self.load_spx_close_rtns(day)
         if spx[-1]['rtn'] > self.min_threshold:
-            return None
+            return False
         cvar = self.get_cvar(day, risk, spx=spx)
-        if cvar is not None and spx[-1]['rtn'] < cvar:
-            portfolio = self._builder.get_portfolios(day, risk, PortfoliosType.RIGHT_SIDE)
-            id = rrs.insert({
-                'date': day,
-                'type': SignalType.MARKET_RIGHT,
-                'risk': risk,
-                'portfolio_type': PortfoliosType.RIGHT_SIDE,
-                'portfolio': portfolio
-            })
-            return rrs.get_by_id(id)
-        return None
+        return cvar is not None and spx[-1]['rtn'] < cvar
 
     def get_cvar(self, day, risk: PortfoliosRisk, spx = None):
         if spx is None:

rebalance/test_case.py

 import unittest
 
-from api import SignalBuilder, PortfoliosRisk
+from api import SignalBuilder, PortfoliosRisk, RebalanceBuilder
 from framework import autowired, parse_date, get_logger
 
 
@@ -27,6 +27,10 @@ class RebalanceTest(unittest.TestCase):
     def test_high_buy(self, builder: SignalBuilder = None):
         builder.get_signal(parse_date('2022-09-10'), PortfoliosRisk.FT3)
 
+    @autowired
+    def test_rebalance_builder(self, builder: RebalanceBuilder = None):
+        builder.get_rebalance(parse_date('2022-09-01'), PortfoliosRisk.FT3)
+
 
 if __name__ == '__main__':
     unittest.main()