回测报告增加基金池

asset_pool/dao/robo_assets_pool.py

@@ -12,6 +12,19 @@ __COLUMNS__ = {
 }
 
 
+@read
+def get_list(max_date=None, min_date=None, type: AssetPoolType = None):
+    sqls = []
+    if max_date:
+        sqls.append(f"rap_date <= '{format_date(max_date)}'")
+    if min_date:
+        sqls.append(f"rap_date >= '{format_date(min_date)}'")
+    return f'''
+    select {','.join([f"`{x[0]}` as `{x[1]}`" for x in __COLUMNS__.items()])} from robo_assets_pool
+    {where(*sqls, rap_type=type)} order by rap_type, rap_date
+    '''
+
+
 @read(one=True)
 def get_one(day, type: AssetPoolType):
     return f'''select {','.join([f"`{x[0]}` as `{x[1]}`" for x in __COLUMNS__.items()])} from robo_assets_pool {where(rap_date=day, rap_type=type)}'''

asset_pool/pool.py

 from datetime import datetime as dt
+from typing import List
 
-from py_jftech import component, autowired
+import pandas as pd
+from py_jftech import component, autowired, workday_range, parse_date
 
-from api import AssetPool, AssetOptimize, AssetRisk
+from api import AssetPool, AssetOptimize, AssetRisk, RoboReportor, Datum, DatumType, RoboExecutor
 from asset_pool.dao import robo_assets_pool as rap
 
 
@@ -21,3 +23,31 @@ class FundAssetPool(AssetPool):
 
     def clear(self, day=None):
         rap.delete(day)
+
+
+@component(bean_name='asset-pool-report')
+class AssetPoolReportor(RoboReportor):
+
+    @autowired
+    def __init__(self, optimize: AssetOptimize = None, risk: AssetRisk = None, datum: Datum = None, executor: RoboExecutor = None):
+        self._optimize = optimize
+        self._risk = risk
+        self._datum = datum
+        self._executor = executor
+
+    @property
+    def report_name(self) -> str:
+        return '基金池'
+
+    def load_report(self, max_date=dt.today(), min_date=None) -> List[dict]:
+        datums = self._datum.get_datums(type=DatumType.FUND)
+        result = []
+        for date in workday_range(min_date or self._executor.start_date, max_date):
+            opts = self._optimize.get_optimize_pool(date)
+            risk = self._risk.get_risk_pool(date)
+            datas = {x['bloombergTicker']: 0 if x['id'] in risk else 1 if x['id'] in opts else -1 for x in datums}
+            result.append({'date': date, **datas})
+        return result
+
+
+

asset_pool/test_case.py

 import logging
 import unittest
 
-from py_jftech import autowired
+from py_jftech import autowired, parse_date, to_str
 
-from api import AssetPool
+from api import AssetPool, RoboReportor
+
+logger = logging.getLogger(__name__)
 
 
 class AssetPoolTest(unittest.TestCase):
-    logger = logging.getLogger(__name__)
 
     @autowired
-    def test_something(self, pool: AssetPool = None):
-        self.logger.info(pool)
+    def test_asset_pool(self, pool: AssetPool = None):
+        result = pool.get_pool(parse_date('2008-11-05'))
+
+    @autowired(names={'report': 'asset-pool-report'})
+    def test_pool_report(self, report: RoboReportor = None):
+        result = report.load_report(max_date=parse_date('2009-12-31'))
+        logger.info(to_str(result))
 
 
 if __name__ == '__main__':

config.yml

@@ -200,12 +200,13 @@ reports: # 报告模块相关
     exist-build: on # 如果报告文件存在，是否重新构建文件
     save-path: ${EXPORT_PATH:excels} # 导出报告文件存放路径，如果以./或者../开头，则会以执行python文件为根目录，如果以/开头，则为系统绝对路径，否则，以项目目录为根目录
     include-report: # 需要导出的报告类型列表，下面的顺序，也代表了excel中sheet的顺序
-      - hold-report # 持仓报告
-      - signal-report # 信号报告
-      - benckmark-report # benckmark报告
-      - indicators-report # 各种特殊指标报告
-      - fixed-range-report # 固定区间收益报告
-      - relative-range-report # 相对区间收益报告
+#      - hold-report # 持仓报告
+#      - signal-report # 信号报告
+      - asset-pool-report # 基金池报告
+#      - benckmark-report # benckmark报告
+#      - indicators-report # 各种特殊指标报告
+#      - fixed-range-report # 固定区间收益报告
+#      - relative-range-report # 相对区间收益报告
 robo-executor: # 执行器相关
   use: ${ROBO_EXECUTOR:backtest} #执行哪个执行器，优先取系统环境变量ROBO_EXECUTOR的值，默认backtest
   sync-data: ${SYNC_DATA:on}

reports/test_case.py

@@ -31,7 +31,7 @@ class ReportTest(unittest.TestCase):
 
     @autowired(names={'reportor': 'backtest-export'})
     def test_backtest_export(self, exportor: RoboExportor = None):
-        path = exportor.export(max_date=parse_date('2022-11-01'))
+        path = exportor.export(max_date=parse_date('2022-06-01'))
         logger.info(path)