依赖注入实现中

api.py

@@ -52,6 +52,7 @@ class SignalType(Enum):
     MARKET_RIGHT = 4
     HIGH_BUY = 5
     LOW_BUY = 6
+    DRIFT_BUY = 7
 
 
 class Datum(ABC):
@@ -227,18 +228,22 @@ class Solver(ABC):
     '''
     解算器
     '''
+
+    @abstractmethod
     def solve_max_rtn(self):
         '''
         :return: max_rtn, max_var, minCVaR_whenMaxR
         '''
         pass
 
+    @abstractmethod
     def solve_min_rtn(self):
         '''
         :return: min_rtn, min_var, maxCVaR_whenMinR
         '''
         pass
 
+    @abstractmethod
     def solve_mpt(self, min_rtn, max_rtn):
         '''
         常规mpt计算
@@ -248,6 +253,7 @@ class Solver(ABC):
         '''
         pass
 
+    @abstractmethod
     def solve_poem(self, min_rtn, max_rtn, base_cvar, max_cvar):
         '''
         poem方式的mpt计算
@@ -259,6 +265,7 @@ class Solver(ABC):
         '''
         pass
 
+    @abstractmethod
     def reset_navs(self, day):
         '''
         根据指定的日期，重置当前解算器，其他计算，全部依赖这里重置后的基金净值数据
@@ -271,7 +278,7 @@ class Solver(ABC):
     @abstractmethod
     def navs(self):
         '''
-        :return: 当前解算器需要的基金净值
+        :return: 当前解算器使用的基金净值
         '''
         pass
 
@@ -282,6 +289,35 @@ class SolverFactory(ABC):
     '''
     @abstractmethod
     def create_solver(self, risk: PortfoliosRisk, type: PortfoliosType = PortfoliosType.NORMAL) -> Solver:
+        '''
+        根据指定的投组风险等级，以及投组类型，创建解算器
+        :param risk: 投组风险等级
+        :param type: 投组类型
+        :return: 解算器
+        '''
+        pass
+
+
+class PortfoliosHolder(ABC):
+    '''
+    投资组合持仓器
+    '''
+    @abstractmethod
+    def get_portfolios_weight(self, day, risk: PortfoliosRisk):
+        '''
+        获取指定日期指定风险等级的持仓投组比重
+        :param day: 指定日期
+        :param risk: 指定风险等级
+        :return: 持仓投组占比
+        '''
+        pass
+
+    def has_hold(self, risk: PortfoliosRisk) -> bool:
+        '''
+        是否存在指定分线等级的投组持仓
+        :param risk: 指定风险等级
+        :return: 如果已经存在持仓，则返回True, 否则返回False
+        '''
         pass
 
 

basic/datum.py

@@ -18,5 +18,13 @@ class DefaultDatum(Datum):
         return [{**json.loads(x['datas']), 'id': x['id']} for x in result]
 
     def get_high_risk_datums(self, risk: PortfoliosRisk):
-        pass
+        risk3 = self.get_fund_datums(risk=3)
+        if risk is PortfoliosRisk.FT3:
+            return risk3
+        risk3 = [x for x in risk3 if x['assetType'] in ['STOCK', 'BALANCED', 'COMMODITY']]
+        if risk is PortfoliosRisk.FT6:
+            return risk3 + self.get_fund_datums(risk=4)
+        if risk is PortfoliosRisk.FT9:
+            return risk3 + self.get_fund_datums(risk=(4, 5))
+        return None
 

basic/test_case.py

 import unittest
 
-from api import Navs
+from api import Navs, Datum, PortfoliosRisk
 from framework import autowired, parse_date, get_logger
 
 
@@ -12,6 +12,11 @@ class BasicTest(unittest.TestCase):
         closes = navs.get_index_close(ticker='SPX Index', min_date=parse_date('2022-11-01'), datum_ids=67)
         self.logger.info(closes)
 
+    @autowired
+    def test_get_high_risk_datums(self, datum: Datum = None):
+        datums = datum.get_high_risk_datums(PortfoliosRisk.FT9)
+        self.logger.info(datums)
+
 
 if __name__ == '__main__':
     unittest.main()

config.yml

@@ -74,6 +74,8 @@ asset-pool:
       threshold: -0.03
       coef: 0.95
 portfolios:
+  holder:
+    init-nav: 100
   solver:
     tol: 1E-10
     navs:
@@ -111,6 +113,8 @@ portfolios:
       mpt:
         quantile: 0.1
 rebalance:
+  init-signal:
+    date: 2022-09-01
   crisis-signal:
     exp-years: 3
     exp-init: 2022-03-04
@@ -127,6 +131,12 @@ rebalance:
     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/builder.py

@@ -29,7 +29,10 @@ class MptPortfoliosBuilder(PortfoliosBuilder):
                     'date': day
                 })
             portfolio = rmp.get_one(day, type, risk)
-        return json.loads(portfolio['portfolio']) if SolveType(portfolio['solve']) is not SolveType.INFEASIBLE else None
+        if SolveType(portfolio['solve']) is not SolveType.INFEASIBLE:
+            result = json.loads(portfolio['portfolio'])
+            return {int(x[0]): x[1] for x in result.items()}
+        return None
 
     def build_portfolio(self, day, type: PortfoliosType):
         result = {}

portfolios/dao/mysql.sql

@@ -15,4 +15,23 @@ CREATE TABLE IF NOT EXISTS robo_mpt_portfolios
     INDEX (rmp_type)
 ) ENGINE = InnoDB
   AUTO_INCREMENT = 0
-  DEFAULT CHARSET = utf8mb4 COMMENT '最优投组表';
\ No newline at end of file
+  DEFAULT CHARSET = utf8mb4 COMMENT '最优投组表';
+
+
+CREATE TABLE IF NOT EXISTS robo_hold_portfolios
+(
+    rhp_id          BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
+    rhp_date        DATETIME        NOT NULL COMMENT '日期',
+    rhp_risk        TINYINT         NOT NULL COMMENT '风险等级',
+    rhp_rrs_id      BIGINT UNSIGNED          DEFAULT NULL COMMENT '调仓信号id',
+    rhp_rebalance   TINYINT         NOT NULL DEFAULT 0 COMMENT '是否调仓',
+    rhp_portfolios  JSON            NOT NULL COMMENT '投组信息',
+    rhp_nav         DOUBLE          NOT NULL COMMENT '资产值',
+    rhp_create_time DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    rhp_update_time DATETIME                 DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
+    PRIMARY KEY (rhp_id),
+    UNIQUE INDEX (rhp_date, rhp_risk),
+    INDEX (rhp_risk)
+) ENGINE = InnoDB
+  AUTO_INCREMENT = 0
+  DEFAULT CHARSET = utf8mb4 COMMENT '持仓投组表';
\ No newline at end of file

portfolios/dao/robo_hold_portfolios.py

+from framework import read, where, write
+from api import PortfoliosRisk
+
+__COLUMNS__ = {
+    'rhp_id': 'id',
+    'rhp_date': 'date',
+    'rhp_risk': 'risk',
+    'rhp_rrs_id': 'signal_id',
+    'rhp_rebalance': 'rebalance',
+    'rhp_portfolios': 'portfolios',
+    'rhp_nav': 'nav',
+}
+
+
+@read(one=True)
+def get_one(day, risk: PortfoliosRisk):
+    return f'''select {','.join([f'{x[0]} as {x[1]}' for x in __COLUMNS__.items()])} from robo_hold_portfolios {where(rhp_date=day, rhp_risk=risk)}'''
+
+
+def get_count(risk: PortfoliosRisk = None):
+    @read(one=True)
+    def exec():
+        return f'''select count(*) as `count` from robo_hold_portfolios {where(rhp_risk=risk)}'''
+    result = exec()
+    return result['count']

portfolios/holder.py

+from framework import component, autowired
+from api import PortfoliosHolder, PortfoliosRisk
+from portfolios.dao import robo_hold_portfolios as rhp
+import json
+
+
+@component(bean_name='next-re')
+class NextReblanceHolder(PortfoliosHolder):
+
+    def get_portfolios_weight(self, day, risk: PortfoliosRisk):
+        hold = rhp.get_one(day, risk)
+        if hold:
+            result = json.loads(hold['portfolios']['weight'])
+            return {int(x[0]): x[1] for x in result.items()}
+        return None
+
+    def has_hold(self, risk: PortfoliosRisk) -> bool:
+        return rhp.get_count(risk=risk) > 0
\ No newline at end of file

portfolios/test_case.py

 import unittest
 from framework import autowired, parse_date, get_logger
-from api import PortfoliosBuilder, PortfoliosType, PortfoliosRisk
+from api import PortfoliosBuilder, PortfoliosType, PortfoliosRisk, PortfoliosHolder
 
 
 class PortfoliosTest(unittest.TestCase):
@@ -21,6 +21,10 @@ class PortfoliosTest(unittest.TestCase):
         portfolio = builder.get_portfolios(parse_date('2022-11-07'), PortfoliosRisk.FT9)
         self.logger.info(portfolio)
 
+    @autowired(names={'hold': 'next-re'})
+    def test_has_hold(self, hold: PortfoliosHolder = None):
+        self.logger.info(hold.has_hold(PortfoliosRisk.FT3))
+
 
 if __name__ == '__main__':
     unittest.main()

rebalance/curve_drift.py

-from framework import component, autowired
-from api import SignalBuilder, PortfoliosRisk, SignalType
+from framework import component, autowired, get_config
+from api import SignalBuilder, PortfoliosRisk, SignalType, Datum, PortfoliosBuilder, PortfoliosHolder
 from rebalance.dao import robo_rebalance_signal as rrs
 
 
 @component(bean_name='curve-drift')
 class CurveDrift(SignalBuilder):
 
+    @autowired
+    def __init__(self, datum: Datum = None, builder: PortfoliosBuilder = None, hold: PortfoliosHolder = None):
+        self._datum = datum
+        self._builder = builder
+        self._hold = hold
+        self._config = get_config(__name__)
+
     @property
     def exclude_last_type(self):
         return [
@@ -20,4 +27,32 @@ class CurveDrift(SignalBuilder):
         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
+        hr_datums = self._datum.get_high_risk_datums(risk)
+        datum_ids = [x['id'] for x in hr_datums]
+        normal_portfolio = self._builder.get_portfolios(day, risk)
+        normal_weight = round(sum([x[1] for x in normal_portfolio.items() if x[0] in datum_ids]), 2)
+        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
+
+    @property
+    def diff_threshold(self):
+        return self._config['diff-threshold']
+
+    @property
+    def init_factor(self):
+        return self._config['init-factor']
+
+
 

rebalance/dao/mysql.sql

@@ -16,3 +16,22 @@ CREATE TABLE IF NOT EXISTS robo_rebalance_signal
 ) ENGINE = InnoDB
   AUTO_INCREMENT = 0
   DEFAULT CHARSET = utf8mb4 COMMENT '再平衡信号表';
+
+
+
+CREATE TABLE IF NOT EXISTS robo_weight_drift
+(
+    rwd_id          BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
+    rwd_date        DATETIME        NOT NULL COMMENT '日期',
+    rwd_risk        TINYINT         NOT NULL COMMENT '风险等级',
+    rwd_weight      DOUBLE          NOT NULL COMMENT '高风险资产权重',
+    rwd_drift       DOUBLE          NOT NULL COMMENT '资产权重漂移计算值',
+    rwd_create_time DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    rwd_update_time DATETIME                 DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
+    PRIMARY KEY (rwd_id),
+    UNIQUE INDEX (rwd_date, rwd_risk),
+    INDEX (rwd_risk)
+) ENGINE = InnoDB
+  AUTO_INCREMENT = 0
+  DEFAULT CHARSET = utf8mb4 COMMENT '高风险资产权重漂移表';
+

rebalance/dao/robo_rebalance_signal.py

@@ -37,6 +37,13 @@ def get_last_one(max_date, risk: PortfoliosRisk, type: SignalType = None, effect
     '''
 
 
+def get_count(risk: PortfoliosRisk):
+    def exec():
+        return f"select count(*) as `count` from robo_rebalance_signal {where(rrs_risk=risk)}"
+    result = exec()
+    return result['count']
+
+
 @write
 def insert(datas):
     datas = {x[0]: datas[x[1]] for x in __COLUMNS__.items() if x[1] in datas and datas[x[1]] is not None}

rebalance/dao/robo_weight_drift.py

+from framework import read, write, where, format_date
+from api import PortfoliosRisk
+
+__COLUMNS__ = {
+    'rwd_id': 'id',
+    'rwd_date': 'date',
+    'rwd_risk': 'risk',
+    'rwd_weight': 'weight',
+    'rwd_drift': 'drift',
+}
+
+
+@read(one=True)
+def get_one(day, risk: PortfoliosRisk):
+    return f"select {','.join([f'{x[0]} as {x[1]}' for x in __COLUMNS__.items()])} from robo_weight_drift {where(rwd_date=day, rwd_risk=risk)}"
+
+
+@read(one=True)
+def get_last_one(max_date, risk: PortfoliosRisk):
+    return f'''
+    select {','.join([f'{x[0]} as {x[1]}' for x in __COLUMNS__.items()])} from robo_weight_drift 
+    {where(f"rwd_date <= '{format_date(max_date)}'", rwd_risk=risk)} order by rwd_date desc limit 1
+    '''
+
+
+
+
+
+@write
+def insert(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 = {
+        **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)}
+    }
+    return f'''
+    insert into robo_weight_drift({','.join([x for x in datas.keys()])})
+    values ({','.join([f"'{x[1]}'" for x in datas.items()])})
+    '''

rebalance/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 rebalance.dao import robo_weight_drift as rwd, robo_rebalance_signal as rrs
+from datetime import timedelta, datetime as dt
+
+
+def get_start_date():
+    config = get_config('main')
+    return filter_weekend(config['start-date'])
+
+
+class DriftSupport:
+
+    @autowired
+    def __init__(self, builder: PortfoliosBuilder = None, datum: Datum = None):
+        self._builder = builder
+        self._datum = datum
+        self._config = get_config(__name__)
+
+    @property
+    def drift_coef(self):
+        return self._config['drift-coef']
+
+    def get_threshold(self, risk: PortfoliosRisk):
+        threshold = self._config['threshold']
+        if isinstance(threshold, dict):
+            threshold = threshold[f'ft{risk.value}']
+        return threshold
+
+    def get_drift(self, day, risk: PortfoliosRisk):
+        drift = rwd.get_one(day, risk)
+        if not drift:
+            datum_ids = [x['id'] for x in self._datum.get_high_risk_datums(risk)]
+            last_one = rwd.get_last_one(max_date=day, risk=risk)
+            start = (next_workday(last_one['date'])) if last_one else get_start_date()
+            last_drift = last_one['drift'] if last_one else 0
+            for date in [x for x in pd.date_range(start, day, freq='D') if is_workday(x)]:
+                portfolio = self._builder.get_portfolios(date, risk)
+                weight = round(sum([x[1] for x in portfolio.items() if x[0] in datum_ids]), 2)
+                last_drift = (weight * self.drift_coef + (1 - self.drift_coef) * last_drift) if last_drift else weight
+                rwd.insert({
+                    'date': date,
+                    'risk': risk,
+                    'weight': weight,
+                    'drift': last_drift,
+                })
+            drift = rwd.get_last_one(day, risk)
+        return drift['drift']
+
+
+@component(bean_name='high-buy')
+class HighBuySignal(SignalBuilder, DriftSupport):
+
+    @property
+    def include_last_type(self):
+        return [
+            SignalType.CRISIS_ONE,
+            SignalType.CRISIS_TWO,
+            SignalType.MARKET_RIGHT,
+            SignalType.LOW_BUY
+        ]
+
+    def get_signal(self, day, risk: PortfoliosRisk):
+        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
+        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
+
+
+@component(bean_name='low-buy')
+class LowBuySignal(SignalBuilder, DriftSupport):
+
+    @property
+    def include_last_type(self):
+        return [
+            SignalType.CRISIS_ONE,
+            SignalType.CRISIS_TWO,
+            SignalType.MARKET_RIGHT
+        ]
+
+    def get_signal(self, day, risk: PortfoliosRisk):
+        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
+        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

rebalance/init_signal.py

+from api import SignalBuilder, PortfoliosRisk, SignalType, PortfoliosBuilder
+from framework import component, autowired
+from rebalance.dao import robo_rebalance_signal as rrs
+
+
+@component(bean_name='init')
+class InitSignalBuilder(SignalBuilder):
+
+    @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)
+            id = rrs.insert({
+                'date': day,
+                'type': SignalType.INIT,
+                'risk': risk,
+                'portfolio_type': PortfoliosType.RIGHT_SIDE,
+                'portfolio': portfolio
+            })
+            return rrs.get_by_id(id)
+        return None

rebalance/right_side.py

@@ -35,7 +35,7 @@ class MarketRight(SignalBuilder):
         if spx[-1]['rtn'] > self.min_threshold:
             return None
         cvar = self.get_cvar(day, risk, spx=spx)
-        if spx[-1]['rtn'] < cvar:
+        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,

rebalance/test_case.py

@@ -18,6 +18,15 @@ class RebalanceTest(unittest.TestCase):
         signal = builder.get_signal(parse_date('2022-10-13'), PortfoliosRisk.FT9)
         self.logger.info(signal)
 
+    @autowired(names={'builder': 'curve-drift'})
+    def test_curve_drift(self, builder: SignalBuilder = None):
+        signal = builder.get_signal(parse_date('2022-11-07'), PortfoliosRisk.FT3)
+        self.logger.info(signal)
+
+    @autowired(names={'builder': 'high-buy'})
+    def test_high_buy(self, builder: SignalBuilder = None):
+        builder.get_signal(parse_date('2022-09-10'), PortfoliosRisk.FT3)
+
 
 if __name__ == '__main__':
     unittest.main()