依赖注入实现中

api.py

@@ -21,6 +21,27 @@ class AssetPoolType(Enum):
     RISK = 2
 
 
+@unique
+class PortfoliosRisk(Enum):
+    FT3 = 3
+    FT6 = 6
+    FT9 = 9
+
+
+@unique
+class PortfoliosType(Enum):
+    CRISIS_1 = 'crisis_1'
+    CRISIS_2 = 'crisis_2'
+    RIGHT_SIDE = 'right_side'
+    NORMAL = 'normal'
+
+
+@unique
+class SolveType(Enum):
+    INFEASIBLE = 0
+    MPT = 1
+    POEM = 2
+
 
 class BusinessException(Exception):
     def __init__(self, msg):
@@ -118,6 +139,7 @@ class AssetPool(ABC):
     '''
     资产池相关服务
     '''
+
     @abstractmethod
     def get_pool(self, day):
         '''
@@ -126,3 +148,19 @@ class AssetPool(ABC):
         :return: 资产id列表
         '''
         pass
+
+
+class PortfoliosBuilder(ABC):
+    '''
+    投组组合构建器
+    '''
+
+    @abstractmethod
+    def get_portfolios(self, day, risk: PortfoliosRisk, type: PortfoliosType = PortfoliosType.NORMAL):
+        '''
+        获取指定日期，指定风险等级的投资组合
+        :param day: 指定日期
+        :param risk: 风险等级
+        :return: 资产组合字典{id: weight}
+        '''
+        pass

asset_pool/asset_risk.py

-import time
-import traceback
-import pandas as pd
 import json
-from scipy.stats import norm
 from datetime import datetime as dt
+
+import pandas as pd
 from dateutil.relativedelta import relativedelta
-from framework import component, autowired, get_config, log, format_date
+from scipy.stats import norm
+
 from api import AssetRisk, Navs, AssetRiskDateType as DateType, Datum, AssetPoolType
 from asset_pool.dao import asset_risk_dates as ard, asset_ewma_value as aev, robo_assets_pool as rap
-from concurrent.futures import ProcessPoolExecutor, wait, as_completed
+from framework import component, autowired, get_config, log, format_date, block_execute
 
 
 def get_risk_start_date():
@@ -16,10 +15,6 @@ def get_risk_start_date():
     return config['start-date'] - relativedelta(months=3)
 
 
-def test(*args, **kwargs):
-    print(args, kwargs)
-
-
 @component
 class CvarEwmaAssetRisk(AssetRisk):
     '''
@@ -36,14 +31,14 @@ class CvarEwmaAssetRisk(AssetRisk):
     def get_risk_pool(self, day):
         asset_pool = rap.get_one(day, AssetPoolType.RISK)
         if not asset_pool:
-            with ProcessPoolExecutor(max_workers=4) as process:
-                futures = {process.submit(self.is_risk, x['id'], day): x for x in self._datum.get_fund_datums()}
-                risk_ids = [futures[x]['id'] for x in as_completed(futures) if x.result()]
+            result = block_execute(self.is_risk, {x['id']: (x['id'], day) for x in self._datum.get_fund_datums()})
+            risk_ids = [x[0] for x in result.items() if x[1]]
             rap.insert(day, AssetPoolType.RISK, risk_ids)
             asset_pool = rap.get_one(day, AssetPoolType.RISK)
         return json.loads(asset_pool['asset_ids'])
 
     def is_risk(self, id, day) -> bool:
+        print(id, day)
         asset_pool = rap.get_one(day, AssetPoolType.RISK)
         if asset_pool:
             return id in json.loads(asset_pool['asset_ids'])
@@ -51,8 +46,8 @@ class CvarEwmaAssetRisk(AssetRisk):
         last = ard.get_last_one(id, day)
         return DateType(last['type']) is DateType.START_DATE if last else False
 
-    def build_risk_date(self, asset_id, day=None):
-        risk_date = "first"
+    def build_risk_date(self, asset_id, day=dt.today()):
+        risk_date = not None
         try:
             log.debug(f"start build risk date for asset[{asset_id}] to date[{format_date(day)}]")
             while risk_date is not None:
@@ -123,15 +118,10 @@ class CvarEwmaAssetRisk(AssetRisk):
 
     def get_income_return(self, asset_id, min_date=None, max_date=None):
         fund_navs = pd.DataFrame(self._navs.get_fund_navs(fund_ids=asset_id, max_date=max_date))
-        fund_navs['rtn'] = fund_navs['nav_cal'] / fund_navs['nav_cal'].shift(self._config['rtn-day']) - 1
+        fund_navs['rtn'] = fund_navs['nav_cal'] / fund_navs['nav_cal'].shift(self._config['rtn-days']) - 1
         fund_navs.dropna(inplace=True)
         if min_date:
             fund_navs = fund_navs[fund_navs.nav_date >= pd.to_datetime(min_date)]
         fund_navs.rename(columns={'nav_date': 'date'}, inplace=True)
         fund_navs = fund_navs[['date', 'rtn']]
         return fund_navs.to_dict('records')
-
-
-@autowired
-def build_date_task(id, day=dt.today(), asset_risk: CvarEwmaAssetRisk = None):
-    asset_risk.build_risk_date(id, day)
\ No newline at end of file

config.yml

@@ -32,7 +32,7 @@ framework:
         level: INFO
         stream: ext://sys.stdout
       file:
-        class:  logging.handlers.TimedRotatingFileHandler
+        class: logging.handlers.TimedRotatingFileHandler
         level: INFO
         formatter: brief
         filename: logs/info.log
@@ -42,12 +42,12 @@ framework:
         when: D
     loggers:
       prod:
-        handlers: [console,file]
+        handlers: [ console,file ]
         level: INFO
         propagate: 0
     root:
       level: INFO
-      handlers: [console]
+      handlers: [ console ]
 main:
   start-date: 2022-09-01
 basic:
@@ -61,12 +61,12 @@ asset-pool:
       - months: 3
         weight: 0.5
       - months: 6
-        weight: 0.2
+        weight: 0.3
       - years: 1
         weight: 0.2
   asset-risk:
     advance-months: 3
-    rtn-day: 5
+    rtn-days: 5
     ewma:
       condition-total: 6
       condition-meet: 4
@@ -76,5 +76,43 @@ asset-pool:
       min-volume: 30
       threshold: -0.03
       coef: 0.95
+portfolios:
+  builder:
+    tol: 1E-10
+    navs:
+      months: 3
+      max-nan:
+        asset: 8
+        day: 0.5
+      risk:
+        ft3: [ 2, 3 ]
+        ft6: [ 2, 3, 4 ]
+        ft9: [ 2, 3, 4, 5 ]
+    matrix-rtn-days: 21
+    asset-count: 5 # count or [min, max]
+    mpt:
+      cvar-beta: 0.2
+      quantile: 0.9
+      low-weight: 0.05
+      high-weight: [ 1, 0.6, 0.35 ]
+    poem:
+      cvar-scale-factor: 0.1
+    right-side:
+      navs:
+        risk: [1, 2]
+        exclude-asset-type: ['STOCK', 'BALANCED']
+      mpt:
+        quantile: 0.3
+    crisis_1:
+      navs:
+        risk: [1, 2]
+      mpt:
+        quantile: 0.1
+    crisis_2:
+      navs:
+        risk: [ 1, 2 ]
+      mpt:
+        quantile: 0.1
+
 
 

framework/__init__.py

@@ -4,7 +4,7 @@ from .database import read, write, transaction, where, to_columns
 from .env_config import config, get_config
 from .logger import build_logger, logger as log
 from .injectable import component, autowired, get_instance, init_injectable as _init_injectable
-from .mulit_process import process_pool, process
+from .mulit_process import process_pool, create_process_pool, block_execute
 
 _init_injectable()
 del injectable, logger, env_config, database, base, date_utils, _init_injectable, mulit_process

framework/mulit_process.py

-from concurrent.futures import ProcessPoolExecutor
+from concurrent.futures import ProcessPoolExecutor, as_completed
 from framework.env_config import get_config
 from functools import partial, wraps
 
-
 config = get_config(__name__)
 process_pool = ProcessPoolExecutor(max_workers=config['max-workers'] or 2)
 
 
-def process(func=None):
-    if func is None:
-        return partial(process)
+def create_process_pool(max_workers=None):
+    return ProcessPoolExecutor(max_workers=max_workers or config['max-workers'])
 
-    @wraps(func)
-    def wrap(*args, **kwargs):
-        return process_pool.submit(func, *args, **kwargs)
 
-    return wrap
+def block_execute(func, params: dict, isolate=False) -> dict:
+    if isolate:
+        with create_process_pool() as process:
+            futures = {process.submit(func, *x[1]): x[0] for x in params.items()}
+            return {futures[x]: x.result() for x in as_completed(futures)}
+    else:
+        futures = {process_pool.submit(func, *x[1]): x[0] for x in params.items()}
+        return {futures[x]: x.result() for x in as_completed(futures)}

main.py

 from framework import autowired, parse_date, log
-from api import AssetPool
+from api import PortfoliosBuilder, PortfoliosRisk
 
 
 @autowired
-def start(pool: AssetPool = None):
+def start(builder: PortfoliosBuilder = None):
     day = parse_date('2022-11-07')
-    log.info(pool.get_pool(day))
+    log.info(builder.get_portfolios(day, PortfoliosRisk.FT3))
+
+
+def test(arg):
+    if arg:
+        return 1, 1
+    else:
+        return None
 
 
 if __name__ == '__main__':
     log.info("start")
-    start()
+    start()
\ No newline at end of file

portfolios/dao/mysql.sql

+CREATE TABLE IF NOT EXISTS robo_mpt_portfolios
+(
+    rmp_id          BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
+    rmp_date        DATETIME        NOT NULL COMMENT '日期',
+    rmp_risk        TINYINT         NOT NULL COMMENT '风险等级',
+    rmp_type        VARCHAR(255)    NOT NULL COMMENT '投组类型',
+    rmp_rolve       TINYINT         NOT NULL COMMENT '求解方式',
+    rmp_portfolio   JSON                     DEFAULT NULL COMMENT '投组权重',
+    rmp_cvar        DOUBLE                   DEFAULT NULL COMMENT '投组cvar',
+    rmp_create_time DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    rmp_update_time DATETIME                 DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
+    PRIMARY KEY (rmp_id),
+    UNIQUE INDEX (rmp_date, rmp_risk, rmp_type),
+    INDEX (rmp_risk),
+    INDEX (rmp_type)
+) ENGINE = InnoDB
+  AUTO_INCREMENT = 0
+  DEFAULT CHARSET = utf8mb4 COMMENT '最优投组表';
\ No newline at end of file

portfolios/dao/robo_mpt_portfolios.py

+from datetime import datetime
+from framework import read, write, where, format_date
+from enum import Enum
+from api import PortfoliosRisk, PortfoliosType
+
+__COLUMNS__ = {
+    'rmp_id': 'id',
+    'rmp_date': 'date',
+    'rmp_risk': 'risk',
+    'rmp_type': 'type',
+    'rmp_rolve': 'solve',
+    'rmp_portfolio': 'portfolio',
+    'rmp_cvar': 'cvar',
+}
+
+
+@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_mpt_portfolios({','.join([x for x in datas.keys()])})
+    values ({','.join([f"'{x[1]}'" for x in datas.items()])})
+    '''
+
+
+@read(one=True)
+def get_one(day, type: PortfoliosType, risk: PortfoliosRisk):
+    return f'''
+    select {','.join([x for x in __COLUMNS__.keys()])} from robo_mpt_portfolios
+    {where(rmp_date=day, rmp_risk=risk, rmp_type=type)} 
+    '''

requirements.txt

@@ -6,7 +6,9 @@ lxml==4.9.0
 numpy==1.23.4
 pandas==1.5.1
 pandas-datareader==0.10.0
+ply==3.11
 PyMySQL==1.0.2
+Pyomo @ git+https://github.com/Pyomo/pyomo.git@3cb35f91c7c1ab4e50871110450429ec29950273
 python-dateutil==2.8.2
 pytz==2022.6
 PyYAML==6.0