依赖注入实现中

api.py

+from framework import parse_date, get_quarter_start, get_config
 from abc import ABC, abstractmethod
 from enum import Enum, unique
+from dateutil.relativedelta import relativedelta
 
 
 @unique
@@ -7,6 +9,19 @@ class DatumType(Enum):
     FUND = 'FUND'
 
 
+@unique
+class AssetRiskDateType(Enum):
+    START_DATE = 1
+    STOP_DATE = 2
+
+
+@unique
+class AssetPoolType(Enum):
+    OPTIMIZE = 1
+    RISK = 2
+
+
+
 class BusinessException(Exception):
     def __init__(self, msg):
         self.__msg = msg
@@ -97,3 +112,17 @@ class AssetRisk(ABC):
         :return: 如果处于风控状态则返回True，否则返回False
         '''
         pass
+
+
+class AssetPool(ABC):
+    '''
+    资产池相关服务
+    '''
+    @abstractmethod
+    def get_pool(self, day):
+        '''
+        返回指定日期的可用资产池
+        :param day: 指定日期
+        :return: 资产id列表
+        '''
+        pass

asset_pool/asset_optimize.py

@@ -3,9 +3,10 @@ import pandas as pd
 from abc import ABC, abstractmethod
 from dateutil.relativedelta import relativedelta
 from empyrical import sortino_ratio
-from framework import filter_weekend, dict_remove, get_config, component, autowired
-from api import AssetOptimize, Navs, BusinessException, Datum
-from asset_pool.dao import robo_optimize_pool as rop
+from framework import filter_weekend, dict_remove, get_config, component, autowired, get_quarter_start
+from api import AssetOptimize, Navs, BusinessException, Datum, AssetPoolType
+from asset_pool.dao import robo_assets_pool as rop
+from datetime import datetime as dt
 
 
 class SortinoAssetOptimize(AssetOptimize, ABC):
@@ -15,7 +16,7 @@ class SortinoAssetOptimize(AssetOptimize, ABC):
         self._config = [{
             **x,
             'name': [f"sortino_{y[1]}_{y[0]}" for y in x.items() if y[0] != 'weight'][0]
-        } for x in optimize_config['sortino_weight']] if 'sortino_weight' in optimize_config else []
+        } for x in optimize_config['sortino-weight']] if 'sortino-weight' in optimize_config else []
 
     def find_optimize(self, fund_ids, day):
         if not self._config:
@@ -34,17 +35,18 @@ class SortinoAssetOptimize(AssetOptimize, ABC):
         return pct_change.columns[sortino.index[0]]
 
     def get_optimize_pool(self, day):
-        last_one = rop.get_last_one(day)
-        if not last_one:
+        last_one = rop.get_last_one(day=day, type=AssetPoolType.OPTIMIZE)
+        start = get_quarter_start(day or dt.today())
+        if not last_one or start > last_one['date']:
             pool = []
             for fund_group in self.get_groups():
                 if len(fund_group) > 1:
                     pool.append(self.find_optimize(fund_group, day))
                 else:
                     pool.append(fund_group[0])
-            rop.insert(day, sorted(pool))
-            last_one = rop.get_last_one(day)
-        return json.loads(last_one['fund_ids'])
+            rop.insert(day, AssetPoolType.OPTIMIZE, sorted(pool))
+            last_one = rop.get_last_one(day=day, type=AssetPoolType.OPTIMIZE)
+        return json.loads(last_one['asset_ids'])
 
     @abstractmethod
     def get_groups(self):

asset_pool/asset_pool.py

+from framework import component, autowired
+from api import AssetPool, AssetOptimize, AssetRisk
+from datetime import datetime as dt
+
+
+@component
+class FundAssetPool(AssetPool):
+
+    @autowired
+    def __init__(self, optimize: AssetOptimize = None, risk: AssetRisk = None):
+        self._optimize = optimize
+        self._risk = risk
+
+    def get_pool(self, day=dt.today()):
+        opti_pool = self._optimize.get_optimize_pool(day)
+        risk_pool = self._risk.get_risk_pool(day)
+        return [x for x in opti_pool if x not in risk_pool]
\ No newline at end of file

asset_pool/asset_risk.py

+import time
+import traceback
 import pandas as pd
-from framework import component, autowired, get_config
-from api import AssetRisk, Navs
+import json
+from scipy.stats import norm
+from datetime import datetime as dt
+from dateutil.relativedelta import relativedelta
+from framework import component, autowired, get_config, log, format_date
+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
+
+
+def get_risk_start_date():
+    config = get_config("main")
+    return config['start-date'] - relativedelta(months=3)
+
+
+def test(*args, **kwargs):
+    print(args, kwargs)
 
 
 @component
 class CvarEwmaAssetRisk(AssetRisk):
     '''
-    CVAR方式决定风控开始
-    EWMA方式决定风控结束
+    CVAR方式决定风控开始。风控开始后，开始计算ewma寻找风控结束日期，也就是ewma的起始日期
+    EWMA方式决定风控结束：风控结束后，就可以找到风控期的最低点日期，该日期作为下一轮cvar计算的起始日期
     '''
+
     @autowired
-    def __init__(self, navs: Navs = None):
+    def __init__(self, navs: Navs = None, datum: Datum = None):
         self._navs = navs
+        self._datum = datum
         self._config = get_config(__name__)
 
     def get_risk_pool(self, day):
-        pass
+        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()]
+            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:
-        rtn = self.get_income_return(id, day)
-        print(rtn)
+        asset_pool = rap.get_one(day, AssetPoolType.RISK)
+        if asset_pool:
+            return id in json.loads(asset_pool['asset_ids'])
+        self.build_risk_date(id, day)
+        last = ard.get_last_one(id, day)
+        return DateType(last['type']) is DateType.START_DATE if last else False
 
-    def get_cvar_start_date(self):
-        pass
+    def build_risk_date(self, asset_id, day=None):
+        risk_date = "first"
+        try:
+            log.debug(f"start build risk date for asset[{asset_id}] to date[{format_date(day)}]")
+            while risk_date is not None:
+                risk_date = self.get_next_date(asset_id, day=day)
+        except Exception as e:
+            log.exception(f"build risk date for asset[{asset_id}] after date[{risk_date}] error", e)
 
-    def get_ewma_start_date(self):
-        pass
+    def get_next_date(self, asset_id, day=dt.today()):
+        last = ard.get_last_one(asset_id, day)
+        if not last or DateType(last['type']) is DateType.START_DATE:
+            start_date = last['date'] if last else get_risk_start_date()
+            ewma = pd.DataFrame(self.get_ewma_value(asset_id, min_date=start_date, max_date=day))
+            total = self._config['ewma']['condition-total']
+            meet = self._config['ewma']['condition-meet']
+            threshold = self._config['ewma']['threshold']
+            if len(ewma) < total:
+                return None
+            for index in range(total, len(ewma) - 1):
+                sub_ewma = ewma[index - total:index]
+                if len(sub_ewma[sub_ewma['ewma'] >= threshold]) >= meet:
+                    stop_date = sub_ewma.iloc[-1]['date']
+                    ard.insert(asset_id, DateType.STOP_DATE, stop_date)
+                    return {'date': stop_date, 'type': DateType.STOP_DATE}
+        elif DateType(last['type']) is DateType.STOP_DATE:
+            last_start = ard.get_last_one(asset_id, last['date'], type=DateType.START_DATE)
+            start_date = last_start['date'] if last_start else get_risk_start_date()
+            rtns = pd.DataFrame(self.get_income_return(asset_id, min_date=start_date, max_date=day))
+            risk_rtns = rtns[rtns.date <= last['date']]
+            cvar_start_date = risk_rtns.loc[risk_rtns.rtn.idxmin()].date
+            for index, row in rtns[rtns.date >= cvar_start_date].iterrows():
+                tigger = False
+                cvar_rtns = rtns[(rtns.date >= cvar_start_date) & (rtns.date <= row['date'])]
+                if row['rtn'] < rtns[rtns.date == cvar_start_date].iloc[0].rtn:
+                    # 当日回报率跌破最低点, 则直接触发
+                    tigger = True
+                elif row['rtn'] <= self._config['cvar']['threshold'] and len(cvar_rtns) >= self._config['cvar']['min-volume']:
+                    # 当日回报率小于等于阀值并且有足够cvar累计计算数据，则计算cvar判断
+                    alpha = 1 - self._config['cvar']['coef']
+                    mean = cvar_rtns['rtn'].mean()
+                    std = cvar_rtns['rtn'].std()
+                    cvar = mean - std * norm.pdf(norm.ppf(alpha)) / alpha
+                    tigger = row['rtn'] < cvar
+                if tigger:
+                    ard.insert(asset_id, DateType.START_DATE, row['date'])
+                    return {'date': row['date'], 'type': DateType.START_DATE}
+        return None
 
-    def get_income_return(self, id, day):
-        fund_navs = pd.DataFrame(self._navs.get_fund_navs(fund_ids=id, max_date=day))
-        fund_navs['rtn'] = fund_navs['nav_cal'] / fund_navs['nav_cal'].shift(self._config['rtn_day']) - 1
+    def get_ewma_value(self, id, min_date=get_risk_start_date(), max_date=None):
+        rtn = pd.DataFrame(self.get_income_return(id, min_date=min_date, max_date=max_date))
+        if rtn.empty:
+            return []
+        rtn.sort_values('date', inplace=True)
+
+        last_one = aev.get_last_one(id, max_date=max_date)
+        if not last_one:
+            aev.insert(asset_id=id, date=rtn.iloc[0].date, value=rtn.iloc[0].rtn)
+            last_one = aev.get_last_one(id, max_date=max_date)
+        last_day = last_one['date']
+        if last_day < max_date:
+            ewma = last_one['value']
+            factor = self._config['ewma']['factor']
+            for index, row in rtn[rtn['date'] > last_day].iterrows():
+                ewma = factor * row['rtn'] + (1 - factor) * ewma
+                aev.insert(id, row['date'], ewma)
+
+        result = aev.get_list(id, min_date=min_date, max_date=max_date)
+        return [{'date': x['date'], 'ewma': x['value']} for x in result]
+
+    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.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

asset_pool/dao/asset_ewma_value.py

+from framework import read, write, where, format_date
+
+
+__COLUMNS__ = {
+    'aev_id': 'id',
+    'aev_date': 'date',
+    'aev_asset_id': 'asset_id',
+    'aev_value': 'value',
+}
+
+
+@write
+def insert(asset_id, date, value):
+    return f'''
+    insert into asset_ewma_value(aev_date, aev_asset_id, aev_value)
+    values ('{format_date(date)}', {asset_id}, {value})
+    '''
+
+
+@read(one=True)
+def get_last_one(asset_id, max_date=None):
+    sqls = []
+    if max_date:
+        sqls.append(f"aev_date <= '{format_date(max_date)}'")
+    return f'''
+    select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from asset_ewma_value 
+    {where(*sqls, aev_asset_id=asset_id)} order by aev_date desc limit 1
+    '''
+
+
+@read
+def get_list(asset_id, min_date=None, max_date=None):
+    sqls =[]
+    if min_date:
+        sqls.append(f"aev_date >= '{format_date(min_date)}'")
+    if max_date:
+        sqls.append(f"aev_date <= '{format_date(max_date)}'")
+    return f'''select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from asset_ewma_value {where(*sqls, aev_asset_id=asset_id)}'''
+
+
+@read
+def get_last(asset_id, max_date=None, limit=1):
+    sqls = []
+    if max_date:
+        sqls.append(f"aev_date <= '{format_date(max_date)}'")
+    return f'''
+    select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from asset_ewma_value 
+    {where(*sqls, aev_asset_id=asset_id)} order by aev_date desc limit {limit}
+    '''

asset_pool/dao/asset_risk_dates.py

+from framework import read, write, where, format_date
+from api import AssetRiskDateType as DateType
+
+__COLUMNS__ = {
+    'ard_id': 'id',
+    'ard_date': 'date',
+    'ard_type': 'type',
+    'ard_asset_id': 'asset_id',
+}
+
+
+@write
+def insert(asset_id, type: DateType, date):
+    return f'''
+    insert into asset_risk_dates(ard_asset_id, ard_type, ard_date)
+    values ({asset_id}, {type.value}, '{format_date(date)}')
+    '''
+
+
+@read(one=True)
+def get_last_one(fund_id, date, type: DateType = None):
+    kwargs = {
+        'ard_asset_id': fund_id,
+        'ard_type': type.value if type is not None else None
+    }
+    sql = f"ard_date <= '{format_date(date)}'" if date else None
+    return f'''
+    select {','.join([f"`{x[0]}` as `{x[1]}`" for x in __COLUMNS__.items()])} 
+    from asset_risk_dates {where(sql, **kwargs)} order by ard_date desc, ard_type asc limit 1
+    '''
+

asset_pool/dao/mysql.sql

-CREATE TABLE IF NOT EXISTS robo_optimize_pool
+CREATE TABLE IF NOT EXISTS robo_assets_pool
 (
-    rop_id          BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
-    rop_date        DATETIME        NOT NULL COMMENT '数据日期',
-    rop_type        TINYINT         NOT NULL COMMENT '资产池类别',
-    rop_asset_ids   JSON                     DEFAULT NULL COMMENT '基金ID',
-    rop_create_time DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    rop_update_time DATETIME                 DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
-    PRIMARY KEY (rop_id),
-    INDEX (rop_date),
-    INDEX (rop_type)
+    rap_id          BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
+    rap_date        DATETIME        NOT NULL COMMENT '数据日期',
+    rap_type        TINYINT         NOT NULL COMMENT '资产池类别',
+    rap_asset_ids   JSON                     DEFAULT NULL COMMENT '基金ID',
+    rap_create_time DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    rap_update_time DATETIME                 DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
+    PRIMARY KEY (rap_id),
+    UNIQUE INDEX (rap_date, rap_type),
+    INDEX (rap_type)
 ) ENGINE = InnoDB
   AUTO_INCREMENT = 0
-  DEFAULT CHARSET = utf8mb4 COMMENT '优选基金池';
+  DEFAULT CHARSET = utf8mb4 COMMENT '资产池';
 
 
 CREATE TABLE IF NOT EXISTS asset_risk_dates
@@ -28,4 +28,20 @@ CREATE TABLE IF NOT EXISTS asset_risk_dates
     INDEX (ard_asset_id)
 ) ENGINE = InnoDB
   AUTO_INCREMENT = 0
-  DEFAULT CHARSET = utf8mb4 COMMENT '资产风控日期表';
+  DEFAULT CHARSET = utf8mb4 COMMENT '资产风控日期表';
\ No newline at end of file
+
+
+CREATE TABLE IF NOT EXISTS asset_ewma_value
+(
+    aev_id          BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
+    aev_date        DATETIME        NOT NULL COMMENT '日期',
+    aev_asset_id    BIGINT UNSIGNED NOT NULL COMMENT '资产ID',
+    aev_value   DOUBLE NOT NULL COMMENT 'ewma值',
+    aev_create_time DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    aev_update_time DATETIME                 DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
+    PRIMARY KEY (aev_id),
+    UNIQUE INDEX (aev_asset_id, aev_date),
+    INDEX (aev_date)
+) ENGINE = InnoDB
+  AUTO_INCREMENT = 0
+  DEFAULT CHARSET = utf8mb4 COMMENT '资产EWMA数据';
\ No newline at end of file

asset_pool/dao/robo_assets_pool.py

+import json
+from framework import read, write, where, format_date
+from api import AssetPoolType
+
+__COLUMNS__ = {
+    'rap_id': 'id',
+    'rap_date': 'date',
+    'rap_type': 'type',
+    'rap_asset_ids': 'asset_ids',
+}
+
+
+@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)}'''
+
+
+@read(one=True)
+def get_last_one(type: AssetPoolType = None, day=None):
+    sql = f"rap_date <= '{format_date(day)}'" if day else None
+    return f'''select {','.join([f"`{x[0]}` as `{x[1]}`" for x in __COLUMNS__.items()])} from robo_assets_pool {where(sql, rap_type=type)} order by rap_date desc limit 1'''
+
+
+@write
+def insert(day, type: AssetPoolType, pool: list):
+    return f'''
+        insert into robo_assets_pool(rap_date, rap_type, rap_asset_ids)
+        values ('{format_date(day)}', {type.value},'{json.dumps(pool)}')
+    '''

asset_pool/dao/robo_optimize_pool.py

-import json
-from framework import read, write, where, format_date
-
-
-__COLUMNS__ = {
-    'rop_id': 'id',
-    'rop_date': 'date',
-    'rop_fund_ids': 'fund_ids',
-}
-
-
-@read
-def get_list(*args, **kwargs):
-    return f'''select {','.join([f"`{x[0]}` as `{x[1]}`" for x in __COLUMNS__.items()])} from robo_optimize_pool {where(*args, **kwargs)}'''
-
-
-@read(one=True)
-def get_one(*args, **kwargs):
-    return f'''select {','.join([f"`{x[0]}` as `{x[1]}`" for x in __COLUMNS__.items()])} from robo_optimize_pool {where(*args, **kwargs)}'''
-
-
-@read(one=True)
-def get_last_one(day=None):
-    sql = f"rop_date <= '{format_date(day)}'" if day else None
-    return f'''select {','.join([f"`{x[0]}` as `{x[1]}`" for x in __COLUMNS__.items()])} from robo_optimize_pool {where(sql)} order by rop_date desc limit 1'''
-
-
-@write
-def insert(day, pool):
-    return f'''
-        insert into robo_optimize_pool(rop_date, rop_fund_ids)
-        values ('{format_date(day)}', '{json.dumps(pool) if isinstance(pool, dict) else pool}')
-    '''

basic/dao/robo_base_datum.py

@@ -3,14 +3,11 @@ from framework import read, where, to_tuple
 
 
 @read
-def get_list(*args, **kwargs):
-    return f'''select rbd_id as id, rbd_datas as datas from robo_base_datum {where(*args, **kwargs)}'''
-
-
-@read(one=True)
-def get_one(*args, **kwargs):
-    return f'''select rbd_id as id, rbd_datas as datas from robo_base_datum {where(*args, **kwargs)}'''
-
-
 def get_base_datums(type: DatumType = None, crncy=None, risk=None, fund_ids=None):
-    return get_list(rbd_id=to_tuple(fund_ids), v_rbd_type=type, v_rbd_crncy=crncy, v_rbd_risk=risk)
+    kwargs = {
+        'rbd_id': to_tuple(fund_ids),
+        'v_rbd_type': type,
+        'v_rbd_crncy': crncy,
+        'v_rbd_risk': risk
+    }
+    return f'''select rbd_id as id, rbd_datas as datas from robo_base_datum {where(**kwargs)}'''

basic/dao/robo_exrate.py

 from framework import read, where, format_date
 
-
-@read
-def get_list(*args, **kwargs):
-    return f'''select re_id as id, re_ticker as ticker, re_date as date, re_close as close from robo_exrate {where(*args, **kwargs)}'''
-
-
-@read(one=True)
-def get_one(*args, **kwargs):
-    return f'''select re_id as id, re_ticker as ticker, re_date as date, re_close as close from robo_exrate {where(*args, **kwargs)}'''
+__COLUMNS__ = {
+    're_id': 'id',
+    're_ticker': 'ticker',
+    're_date': 'date',
+    're_close': 'close',
+}
 
 
+@read
 def get_exrates(ticker=None, min_date=None, max_date=None):
     sqls = []
     if min_date:
         sqls.append(f"re_date >= '{format_date(min_date)}'")
     if max_date:
         sqls.append(f"re_date <= '{format_date(max_date)}'")
-    return get_list(*sqls, re_ticker=ticker)
+    return f'''select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_exrate {where(*sqls, re_ticker=ticker)}'''
 
 
+@read(one=True)
 def get_exrate(ticker, date):
-    return get_one(re_ticker=ticker, re_date=date)
+    return f'''select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_exrate {where(re_ticker=ticker, re_date=date)}'''

basic/dao/robo_fund_navs.py

 from framework import read, where, format_date, to_tuple
 
 
-@read
-def get_list(*args, **kwargs):
-    return f'''select rfn_fund_id as fund_id, rfn_date as nav_date, rfn_nav_cal as nav_cal from robo_fund_navs {where(*args, **kwargs)}'''
-
-
-@read(one=True)
-def get_one(*args, **kwargs):
-    return f'''select rfn_fund_id as fund_id, rfn_date as nav_date, rfn_nav_cal as nav_cal from robo_fund_navs {where(*args, **kwargs)}'''
+__COLUMNS__ = {
+    'rfn_fund_id': 'fund_id',
+    'rfn_date': 'nav_date',
+    'rfn_nav_cal': 'nav_cal',
+}
 
 
+@read
 def get_navs(fund_id=None, min_date=None, max_date=None):
     sqls = []
     if min_date:
         sqls.append(f"rfn_date >= '{format_date(min_date)}'")
     if max_date:
         sqls.append(f"rfn_date <= '{format_date(max_date)}'")
-    return get_list(*sqls, rfn_fund_id=to_tuple(fund_id))
+    return f'''select {','.join([f"`{x[0]}` as `{x[1]}`" for x in __COLUMNS__.items()])} from robo_fund_navs {where(*sqls, rfn_fund_id=to_tuple(fund_id))}'''

config.yml

@@ -15,6 +15,8 @@ framework:
     server: smtphz.qiye.163.com
     user: jft-ra@thizgroup.com
     password: 5dbb#30ec6d3
+  mulit-process:
+    max-workers: 4
   logger:
     version: 1
     use: ${LOG_NAME:root}
@@ -46,21 +48,33 @@ framework:
     root:
       level: INFO
       handlers: [console]
+main:
+  start-date: 2022-09-01
 basic:
   navs:
     exrate:
       - from: EUR
         ticker: EURUSD BGN Curncy
-asset_pool:
-  asset_optimize:
-    sortino_weight:
+asset-pool:
+  asset-optimize:
+    sortino-weight:
       - months: 3
         weight: 0.5
       - months: 6
         weight: 0.2
       - years: 1
         weight: 0.2
-  asset_risk:
-    rtn_day: 5
+  asset-risk:
+    advance-months: 3
+    rtn-day: 5
+    ewma:
+      condition-total: 6
+      condition-meet: 4
+      factor: 0.3
+      threshold: 0
+    cvar:
+      min-volume: 30
+      threshold: -0.03
+      coef: 0.95
 
 

framework/__init__.py

 from .date_utils import *
 from .base import *
-from .database import read, write, transaction, where
+from .database import read, write, transaction, where, to_columns
 from .env_config import config, get_config
-from .logger import build_logger, _log
+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
 
 _init_injectable()
-del injectable, logger, env_config, database, base, date_utils, _init_injectable
-logger = _log
-del _log
+del injectable, logger, env_config, database, base, date_utils, _init_injectable, mulit_process

framework/database.py

@@ -5,8 +5,8 @@ from enum import Enum
 import pymysql
 from pymysql.cursors import DictCursor
 
-from .date_utils import format_date, datetime
-from .env_config import get_config
+from framework.date_utils import format_date, datetime
+from framework.env_config import get_config
 
 
 class DatabaseError(Exception):
@@ -141,3 +141,7 @@ def where(*args, **kwargs) -> str:
     if args:
         result.extend([x for x in args if x])
     return f"where {' and '.join(result)}" if result else ''
+
+
+def to_columns(columns: dict, datas: dict) -> dict:
+    return dict([(x[0], datas[x[1]]) for x in columns.items()])

framework/date_utils.py

 import calendar
-from datetime import timedelta, datetime
+from datetime import timedelta, datetime, date
 
 
 def filter_weekend(day):
@@ -27,5 +27,10 @@ def parse_date(date, has_time=False):
     return datetime.strptime(date, '%Y-%m-%d %H:%M:%S' if has_time else '%Y-%m-%d')
 
 
+def get_quarter_start(today=datetime.today()):
+    result = date(today.year, today.month - (today.month - 1) % 3, 1)
+    return datetime.combine(result, datetime.min.time())
+
+
 if __name__ == '__main__':
-    print(parse_date('2022-10-01 10:24:24', has_time=True))
+    print(get_quarter_start())

framework/env_config.py

@@ -92,6 +92,6 @@ config = build_config()
 def get_config(module: str = None):
     result = config
     if module:
-        for name in module.split('.'):
+        for name in [x.replace('_', '-') for x in module.split('.')]:
             result = result[name] if name in result else {}
     return result

framework/logger.py

@@ -15,4 +15,4 @@ def build_logger(config, name='root'):
 
 
 config = get_config(__name__)
-_log = build_logger(config, name=config['use'] if 'use' in config else None) if config else None
+logger = build_logger(config, name=config['use'] if 'use' in config else None) if config else None

framework/mulit_process.py

+from concurrent.futures import ProcessPoolExecutor
+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)
+
+    @wraps(func)
+    def wrap(*args, **kwargs):
+        return process_pool.submit(func, *args, **kwargs)
+
+    return wrap

main.py

-from framework import autowired, parse_date, logger
-from api import AssetOptimize, AssetRisk
+from framework import autowired, parse_date, log
+from api import AssetPool
 
 
 @autowired
-def start(optimize: AssetOptimize = None, risk: AssetRisk = None):
-    pool = optimize.get_optimize_pool(parse_date('2022-11-07'))
-    logger.info(pool)
-    risk.is_risk(1, parse_date('2022-11-07'))
+def start(pool: AssetPool = None):
+    day = parse_date('2022-11-07')
+    log.info(pool.get_pool(day))
 
 
 if __name__ == '__main__':
+    log.info("start")
     start()