导出模块完毕

api.py

@@ -98,26 +98,14 @@ class Datum(ABC):
 
     @abstractmethod
     def get_datums(self, type: DatumType = None, crncy=None, risk=None, datum_ids=None, ticker=None):
-        pass
-
-    @abstractmethod
-    def get_fund_datums(self, crncy=None, risk=None, fund_ids=None):
-        '''
-        获取基金资料数据
-        :param crncy: 货币类型
-        :param risk: 风险等级，e.g: 1-5
-        :param fund_ids: 基金ID列表
-        :return:基金资料信息
-        '''
-        pass
-
-    @abstractmethod
-    def get_index_datums(self, ticker=None, index_ids=None):
         '''
-        获取指标资料数据
-        :param ticker: 指标的彭博ticker
-        :param index_ids: 指标id列表
-        :return: 即表资料信息
+        获取资料信息，当id和ticker都有时，取二者并集
+        :param type: 资料类型
+        :param crncy: 货币类型，仅对基金资料有效
+        :param risk: 风险等级，仅对基金资料有效
+        :param datum_ids: 资料ID列表
+        :param ticker: 资料ticker列表
+        :return: 资料信息数据
         '''
         pass
 
@@ -181,11 +169,29 @@ class Navs(ABC):
         pass
 
     @abstractmethod
-    def get_eco_values(self, datum_ids=None, min_date=None, max_date=None, ticker=None):
+    def get_eco_values(self, datum_ids=None, min_date=None, max_date=None, ticker=None, by_release_date=False):
+        '''
+        获取经济指标数据，若同时给出ID，和ticker，则取二者并集
+        :param datum_ids: 经济指标id
+        :param min_date: 起始日期
+        :param max_date: 截止日期
+        :param ticker: 经济指标ticker
+        :param by_release_date: 如果为True，则使用公告日期查询，否则使用抓取日期
+        :return: 经济指标的值，包括查询日期，指标和公告日期
+        '''
         pass
 
     @abstractmethod
-    def get_last_eco_values(self, max_date, datum_id=None, ticker=None, count=1):
+    def get_last_eco_values(self, max_date, datum_id=None, ticker=None, count=1, by_release_date=False):
+        '''
+        获取指定资料或ticker，指定日期之前最后count个指标数据，当指定datum_id后，ticker参数无效
+        :param max_date: 指定日期
+        :param datum_id: 指标id，只能指定一个
+        :param ticker: 指标ticker，只能指定一个，当指标id有值后，该参数无效
+        :param count: 指定要返回数据的个数
+        :param by_release_date: 如果为True，则使用公告日期查询，否则使用抓取日期
+        :return: 如果存在，则返回指定日期最后count个指标项（查询日期，指标，公告日期），否则返回None
+        '''
         pass
 
 

basic/dao/robo_eco_datas.py

@@ -29,8 +29,9 @@ def get_list(eco_ids=None, min_date=None, max_date=None):
 
 
 @read(one=True)
-def get_last_one(eco_id, max_date=None):
-    sql = f"red_date <= '{format_date(max_date)}'" if max_date else None
+def get_last_one(eco_id, max_date=None, by_release_date=False):
+    date_field = 'red_release_date' if by_release_date else 'red_date'
+    sql = f"{date_field} <= '{format_date(max_date)}'" if max_date else None
     return f'''
     select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_eco_datas 
     {where(sql, red_eco_id=eco_id)} order by red_date desc limit 1
@@ -38,8 +39,9 @@ def get_last_one(eco_id, max_date=None):
 
 
 @read
-def get_last(eco_id, max_date=None, count=1):
-    sql = f"red_date <= '{format_date(max_date)}'" if max_date else None
+def get_last(eco_id, max_date=None, count=1, by_release_date=False):
+    date_field = 'red_release_date' if by_release_date else 'red_date'
+    sql = f"{date_field} <= '{format_date(max_date)}'" if max_date else None
     return f'''
     select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_eco_datas 
     {where(sql, red_eco_id=eco_id)} order by red_date desc limit {count}

basic/datum.py

 import json
 
-from py_jftech import component, parse_date, get_config
+from py_jftech import component, parse_date, get_config, to_tuple
 
 from api import DatumType, Datum, PortfoliosRisk
 from basic.dao import robo_base_datum as rbd
@@ -25,20 +25,14 @@ class DefaultDatum(Datum):
         return datum
 
     def get_datums(self, type: DatumType = None, crncy=None, risk=None, datum_ids=None, ticker=None):
-        result = rbd.get_base_datums(type=type, crncy=crncy, risk=risk, datum_ids=datum_ids, ticker=ticker)
+        datum_ids = to_tuple(datum_ids)
+        if ticker:
+            datums = rbd.get_base_datums(type=type, ticker=ticker)
+            datum_ids = tuple(set(list(datum_ids or []) | {x['id'] for x in datums}))
+        result = rbd.get_base_datums(type=type, crncy=crncy, risk=risk, datum_ids=datum_ids)
         result = [{**json.loads(x['datas']), 'id': x['id']} for x in result]
         return [self.format_datum(x) for x in result if DatumType(x['type']) is not DatumType.FUND or x['bloombergTicker'] not in self.excludes]
 
-    def get_fund_datums(self, crncy=None, risk=None, fund_ids=None):
-        result = rbd.get_base_datums(type=DatumType.FUND, crncy=crncy, risk=risk, datum_ids=fund_ids)
-        result = [{**json.loads(x['datas']), 'id': x['id']} for x in result]
-        return [{**x, 'inceptDate': parse_date(x['inceptDate'])} for x in result if x['bloombergTicker'] not in self.excludes]
-
-    def get_index_datums(self, ticker=None, index_ids=None):
-        result = rbd.get_base_datums(type=DatumType.INDEX, ticker=ticker, datum_ids=index_ids)
-        return [{**json.loads(x['datas']), 'id': x['id']} for x in result]
-
-
     def get_high_risk_datums(self, risk: PortfoliosRisk):
         risk3 = self.get_datums(type=DatumType.FUND, risk=3)
         if risk is PortfoliosRisk.FT3:

basic/navs.py

@@ -2,7 +2,7 @@ import pandas as pd
 from py_jftech import get_config, component, autowired, to_tuple
 
 from api import Navs, Datum, DatumType
-from basic.dao import robo_exrate as re, robo_fund_navs as rfn, robo_index_datas as rid
+from basic.dao import robo_exrate as re, robo_fund_navs as rfn, robo_index_datas as rid, robo_eco_datas as red
 
 
 @component
@@ -66,8 +66,23 @@ class DefaultNavs(Navs):
                 'close': x['close']
             } for x in last] if last else None
 
-    def get_eco_values(self, datum_ids=None, min_date=None, max_date=None, ticker=None):
-        pass
+    def get_eco_values(self, datum_ids=None, min_date=None, max_date=None, ticker=None, by_release_date=False):
+        datum_ids = to_tuple(datum_ids)
+        if ticker:
+            datums = self._datum.get_datums(type=DatumType.INDEX, ticker=ticker)
+            datum_ids = tuple(set(list(datum_ids or []) | {x['id'] for x in datums}))
+        return red.get_list(eco_ids=datum_ids, min_date=min_date, max_date=max_date, by_release_date=by_release_date)
+
+    def get_last_eco_values(self, max_date, datum_id=None, ticker=None, count=1, by_release_date=False):
+        if not datum_id:
+            assert ticker, "get last eco close, datum_id and ticker give at least one"
+            datum = self._datum.get_datums(type=DatumType.ECO, ticker=ticker)
+            datum_id = datum[0]['id'] if datum else None
+        assert datum_id, "get last eco close, datum id is not found"
+        assert max_date, "get last eco close, start_date is not found"
+        if count == 1:
+            return red.get_last_one(eco_id=datum_id, max_date=max_date, by_release_date=by_release_date)
+        else:
+            return red.get_last(eco_id=datum_id, max_date=max_date, count=count, by_release_date=by_release_date)
+
 
-    def get_last_eco_values(self, max_date, datum_id=None, ticker=None, count=1):
-        pass

basic/sync.py

@@ -91,6 +91,7 @@ class IndexSync(JDCDataSync):
             'pb': x['pbRatio'] if 'pbRatio' in x else None,
             'volume': x['volume'] if 'volume' in x else None,
         } for x in datas if is_workday(dt.fromtimestamp(x['date'] / 1000)) and 'close' in x]
+        if save_datas:
             rid.batch_insert(save_datas)
 
 
@@ -115,6 +116,7 @@ class EcoSync(JDCDataSync):
             'indicator': x['close'],
             'release_date': dt.fromtimestamp(x['releaseDate'] / 1000),
         } for x in datas if 'releaseDate' in x]
+        if save_datas:
             red.batch_insert(save_datas)
 
 
@@ -160,4 +162,5 @@ class FundNavSync(JDCDataSync):
             'div_p': x['postDividend'] if 'postDividend' in x else 0,
             'nav_cal': x['calibrateValue']
         } for x in datas if is_workday(dt.fromtimestamp(x['date'] / 1000))]
+        if save_navs:
             rfn.batch_insert(save_navs)

config.yml

@@ -212,8 +212,8 @@ robo-executor: # 执行器相关
   backtest: # 回测执行器相关
     start-date: 2008-01-02 # 回测起始日期
     end-date: 2022-11-01 # 回测截止日期
-    start-step: 3 # 回测从哪一步开始执行 1:计算资产ewma；2:计算资产池；3:计算最优投组：4:计算再平衡信号以及持仓投组
-    clean-up: off
+    start-step: 4 # 回测从哪一步开始执行 1:计算资产ewma；2:计算资产池；3:计算最优投组：4:计算再平衡信号以及持仓投组
+    clean-up: on
   real: # 实盘执行器
     start-date: 2022-11-01 # 实盘开始时间
 

portfolios/solver.py

@@ -210,7 +210,7 @@ class DefaultSolver(Solver):
     def reset_navs(self, day):
         asset_ids = self._assets.get_pool(day)
         asset_risk = self.get_config('navs.risk')
-        datum = self._datum.get_datums(type=DatumType.FUND, fund_ids=asset_ids, risk=asset_risk)
+        datum = self._datum.get_datums(type=DatumType.FUND, datum_ids=asset_ids, risk=asset_risk)
         exclude = self.get_config('navs.exclude-asset-type') or []
         asset_ids = list(set(asset_ids) & set([x['id'] for x in datum if x['assetType'] not in exclude]))
 

rebalance/dao/robo_rebalance_signal.py

@@ -48,7 +48,7 @@ def get_one(type: SignalType, risk: PortfoliosRisk, date):
 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
+    {where(f"rrs_date >= '{format_date(min_date)}'", rrs_type=type, rrs_risk=risk, rrs_effective=effective)} order by rrs_date limit 1
     '''
 
 

rebalance/signals/crisis_signal.py

@@ -109,14 +109,14 @@ class CrisisTwoSignal(CrisisSignal, BaseRebalanceSignal):
             if not crisis_two:
                 ng_date = day - relativedelta(years=self.negative_growth_years)
                 ten_today = self._navs.get_last_index_close(max_date=day, ticker='USGG10YR Index')
-                cpi_today = self._navs.get_last_index_close(max_date=day, ticker='CPI YOY Index')
+                cpi_today = self._navs.get_last_eco_values(max_date=day, ticker='CPI YOY Index', by_release_date=True)
                 ten_before = self._navs.get_last_index_close(max_date=ng_date, ticker='USGG10YR Index')
-                cpi_before = self._navs.get_last_index_close(max_date=ng_date, ticker='CPI YOY Index')
-                before = ten_before['close'] - cpi_before['close']
-                today = ten_today['close'] - cpi_today['close']
+                cpi_before = self._navs.get_last_eco_values(max_date=ng_date, ticker='CPI YOY Index', by_release_date=True)
+                before = ten_before['close'] - cpi_before['indicator']
+                today = ten_today['close'] - cpi_today['indicator']
 
-                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')
+                fed_today = self._navs.get_last_eco_values(max_date=day, ticker='FDTR Index', by_release_date=True)
+                fed_before = self._navs.get_last_eco_values(max_date=day - relativedelta(months=self.fed_months), ticker='FDTR Index', by_release_date=True)
 
-                return today <= before and fed_today['close'] - fed_before['close'] < self.fed_threshold
+                return today <= before and fed_today['indicator'] - fed_before['indicator'] < self.fed_threshold
         return False

rebalance/test_case.py

 import logging
 import unittest
+from dateutil.relativedelta import relativedelta
 
-from py_jftech import autowired, parse_date, to_str
+from py_jftech import autowired, parse_date, to_str, next_workday
 
 from api import RebalanceSignal, PortfoliosRisk, RebalanceRuler, RoboReportor
 
+logger = logging.getLogger(__name__)
 
-class RebalanceTest(unittest.TestCase):
 
-    logger = logging.getLogger(__name__)
+class RebalanceTest(unittest.TestCase):
 
     @autowired(names={'builder': 'crisis_one'})
     def test_crisis_one(self, builder: RebalanceSignal = None):
-        signal = builder.get_signal(parse_date('2022-10-13'), PortfoliosRisk.FT9)
-        self.logger.info(signal)
+        start = parse_date('2018-07-06')
+        end = start + relativedelta(years=3)
+        while start < end:
+            signal = builder.is_trigger(start, PortfoliosRisk.FT9)
+            if signal:
+                logger.info(start)
+            start = next_workday(start)
+
+    @autowired(names={'builder': 'crisis_two'})
+    def test_crisis_two(self, builder: RebalanceSignal = None):
+        start = parse_date('2018-07-06')
+        end = start + relativedelta(years=3)
+        while start < end:
+            signal = builder.is_trigger(start, PortfoliosRisk.FT9)
+            if signal:
+                logger.info(start)
+            start = next_workday(start)
 
     @autowired(names={'builder': 'market-right'})
     def test_market_right(self, builder: RebalanceSignal = None):
         signal = builder.get_signal(parse_date('2022-10-13'), PortfoliosRisk.FT9)
-        self.logger.info(signal)
+        logger.info(signal)
 
     @autowired(names={'builder': 'curve-drift'})
     def test_curve_drift(self, builder: RebalanceSignal = None):
         signal = builder.get_signal(parse_date('2022-11-07'), PortfoliosRisk.FT3)
-        self.logger.info(signal)
+        logger.info(signal)
 
     @autowired(names={'builder': 'high-buy'})
     def test_high_buy(self, builder: RebalanceSignal = None):
@@ -36,7 +52,7 @@ class RebalanceTest(unittest.TestCase):
     @autowired(names={'reportor': 'signal-report'})
     def test_signal_report(self, reportor: RoboReportor = None):
         result = reportor.load_report()
-        self.logger.info(to_str(result, show_line=10))
+        logger.info(to_str(result, show_line=10))
 
 
 if __name__ == '__main__':