Merge remote-tracking branch 'origin/dev-dividend' into dev-dividend

ai/EstimateMarketTrendV20.py

@@ -6,30 +6,29 @@ from py_jftech import autowired, parse_date, prev_workday, format_date
 from ai.dao.robo_datas import get_base_info, get_index_list, get_fund_list
 from ai.data_access import DataAccess
 from ai.model_trainer import ModelTrainer
+from ai.noticer import send
 from ai.training_data_builder import TrainingDataBuilder
 from api import DataSync
 
-
 # 截止日期
 # max_date = None
-# max_date = '2023-10-06'
-# max_date = '2023-10-13'
-# max_date = '2023-10-20'
-# max_date = '2023-10-27'
-# max_date = '2023-11-03'
-# max_date = '2023-11-10'
-# max_date = '2023-11-17'
-# max_date = '2023-11-24'
-# max_date = '2023-12-01'
-# max_date = '2023-12-08'
-max_date = '2023-12-15'
+max_date = '2024-03-01'
+# max_date = '2024-01-11'
+
+toForecast = True  # False means test, True means forecast
+syncData = True  # 开启会同步数据库指数及基金数据
+uploadData = False  # 开启会上传预测结果
+doReport = True  # 开启会生成Excel报告
 
 # 待预测指数
 # PREDICT_LIST = [67, 121, 122, 123]
-PREDICT_LIST = [67, 121, 122, 123, 155, 157, 158, 159, 160, 161, 162, 163, 164, 165, 166, 168, 169, 170, 171, 174, 175]
+PREDICT_LIST = [67, 121, 122, 123, 155, 156, 157, 158, 159, 160, 161, 162, 163, 164, 165, 166, 167, 168, 169, 170, 171,
+                174, 175, 177, 178]
 eco = [65, 66, 74, 134]
-index = [67, 68, 69, 70, 71, 72, 73, 75,76,77,105,106, 116, 117, 138, 139, 142, 143, 140, 141, 144, 145, 146]
-fund = [121, 122, 123, 155, 157, 158, 159, 160, 161, 162, 163, 164, 165, 166, 168, 169, 170, 171, 174, 175]
+index = [67, 68, 69, 70, 71, 72, 73, 75, 76, 77, 105, 106, 116, 117, 138, 139, 142, 143, 140, 141, 144, 145, 146]
+fund = [121, 122, 123, 155, 156, 157, 158, 159, 160, 161, 162, 163, 164, 165, 166, 167, 168, 169, 170, 171, 174, 175,
+        177,
+        178]
 
 
 @autowired
@@ -47,10 +46,15 @@ def predictionFromMoel(the_model, scaledX_forecast, predict_item, indexDict: dic
     content = f"""\n On day {forecastDay.strftime("%m/%d/%Y")}, the model predicts {predict_item} to be {predictionStr} in {str(numForecastDays)} business days. \n"""
     print(content)
     # 上传预测结果
-    # key = [k for k, v in indexDict.items() if v == predict_item]
-    # index_info = get_base_info(key)[0]
-    # upload_predict(index_info['ticker'], forecastDay, predictionStr)
-    # send(content)
+    key = [k for k, v in indexDict.items() if v == predict_item]
+    index_info = get_base_info(key)[0]
+    if uploadData:
+        from ai.noticer import upload_predict
+        upload_predict(index_info['ticker'], forecastDay, predictionStr)
+    if doReport:
+        from ai.reporter import do_reporter
+        do_reporter()
+    send(content)
     return prediction
 
 
@@ -86,8 +90,8 @@ def judgement(id, type, predict):
 
 ########################################
 if __name__ == '__main__':
-    sync()
-    toForecast = True  # False means test, True means forecast
+    if syncData:
+        sync()
     # define some parameters
     win1W = 5  # 1 week
     win1M = 21  # 1 Month

ai/dao/robo_datas.py

@@ -44,6 +44,6 @@ def get_fund_list(fund_ids=None, min_date=None, max_date=None, limit=None):
 def get_base_info(ids=None):
     sqls = []
     return f"""
-    SELECT rbd_id id,v_rbd_bloomberg_ticker ticker,v_rbd_type type FROM `robo_base_datum`
+    SELECT rbd_id id,v_rbd_bloomberg_ticker ticker,v_rbd_type type, rbd_datas datas FROM `robo_base_datum`
     {where(*sqls,rbd_id=to_tuple(ids))}
     """
\ No newline at end of file

ai/reporter.py

+import datetime
+import json
+
+import pandas as pd
+import requests
+
+symbols = ['ACWI', 'EWJ', 'MCHI', 'EEM', 'BKF', 'INDA', 'AAXJ', 'VGK', 'QQQ', 'SPY', 'SPX', 'IWN',
+           'IUSG', 'IWD', 'DON', 'GDX', 'TOLZ', 'XLU', 'XBI', 'ESGD', 'IGE', 'EMLC', 'IGAA',
+           'LQD', 'HYG', 'SHY', 'IEI', 'IEF', 'GLD', 'IYR', 'UUP', 'CEW', 'TLT']
+
+
+def do_reporter(start='2023-10-01', end=datetime.date.today()):
+    url = f"https://jrp.jfquant.com/api/v1.0/ai/predict?startTime={start}&endTime={end.strftime('%Y-%m-%d')}"
+    resp = requests.get(url)
+    datas = []
+    symbol_index_dict = {symbol: index for index, symbol in enumerate(symbols)}
+    for value in resp.json()['body'].values():
+        for item in value:
+            data = {
+                'Forcast On Date': item['aiPredict']['predictDate'],
+                'Ticker': item['bloombergTicker'].replace(' Index', '').replace(' Equity', ''),
+                'In 21 business days': 'UP' if item['aiPredict']['predict'] == 1 else 'DOWN',
+                'Ticker Name': item['indexName'],
+            }
+            datas.append(data)
+    sorted_data = sorted(datas, key=lambda x: symbol_index_dict[x['Ticker'].split(' ')[0]])
+    print(json.dumps(sorted_data, ensure_ascii=False))
+    pf = pd.DataFrame(sorted_data)
+    pf.to_excel("Forcast_Report.xlsx", index=False)
+
+
+if __name__ == '__main__':
+    do_reporter()

ai/training_data_builder.py

@@ -146,11 +146,9 @@ class TrainingDataBuilder(ABC):
         for col in ['NAPMPMI']:
             DataAll[col].bfill(inplace=True)
             DataAll[col].ffill(inplace=True)
-
-        ###### clean NaN
-        DataAll.ffill(inplace=True)
-        DataAll.dropna(inplace=True)
-        DataAll.reset_index(inplace=True, drop=True)
+        for col in DataAll.columns:
+            if col not in ['CPI_YOY', 'CPURNSA', 'CPI_MOM', 'CPI_MOM_Diff', 'futureR', 'yLabel']:
+                DataAll[col].ffill(inplace=True)
 
         if (self._toForecast):
             # 处理CPI_YOY:美国城镇消费物价指数同比未经季 CPURNSA:美国消费者物价指数未经季调
@@ -166,6 +164,10 @@ class TrainingDataBuilder(ABC):
             X_forecast = forecastData.to_numpy()
             del DataAllCopy
 
+        ###### clean NaN
+        DataAll.dropna(inplace=True)
+        DataAll.reset_index(inplace=True, drop=True)
+
         ###### get X and y
         y = DataAll['yLabel'].to_numpy(copy=True)
 

asset_pool/asset_optimize.py

@@ -4,12 +4,12 @@ from sys import exception
 
 import pandas as pd
 from dateutil.relativedelta import relativedelta
-from empyrical import sortino_ratio
+from empyrical import sortino_ratio, annual_volatility
 from py_jftech import filter_weekend, dict_remove, get_config, component, autowired, next_workday, \
     is_workday
 
 from api import AssetOptimize, Navs, Datum, AssetPoolType, DatumType
-from asset_pool.dao import robo_assets_pool as rop
+from asset_pool.dao import robo_assets_pool as rop, robo_indicator
 
 
 class SortinoAssetOptimize(AssetOptimize, ABC):
@@ -78,6 +78,14 @@ class FundDividendSortinoAssetOptimize(SortinoAssetOptimize):
         self._datum = datum
         self._conf = get_config(__name__)
 
+    @property
+    def annual_volatility_section(self):
+        return self._conf['annual-volatility-section']
+
+    @property
+    def annual_volatility_filter(self):
+        return self._conf['annual-volatility-filter']
+
     @property
     def asset_include(self):
         return self._conf['asset-include']
@@ -109,8 +117,15 @@ class FundDividendSortinoAssetOptimize(SortinoAssetOptimize):
         sortino = sortino.T
         sortino['score'] = sortino.apply(lambda r: sum([x['weight'] * r[x['name']] for x in self._config]), axis=1)
         sortino.sort_values('score', ascending=False, inplace=True)
+        records = sortino.to_dict(orient='records')
+        data = {key: record for record in records for key in fund_ids}
+        self.save_sortino(day, data)
         # 取得分数高的前optimize_count个
-        return pct_change.columns[sortino.index[0:self.optimize_count]].values,sortino['score']
+        return pct_change.columns[sortino.index[0:self.optimize_count]].values, sortino['score']
+
+    def save_sortino(self, day, datas):
+        for key, record in datas.items():
+            robo_indicator.update_sortino(key, day, json.dumps(record))
 
     def get_optimize_pool(self, day):
         opt_pool = rop.get_one(day=day, type=AssetPoolType.OPTIMIZE)
@@ -132,6 +147,45 @@ class FundDividendSortinoAssetOptimize(SortinoAssetOptimize):
             last_one = rop.get_last_one(day=day, type=AssetPoolType.OPTIMIZE)
         return json.loads(last_one['asset_ids'])
 
+    def do_annual_volatility_filter(self, day, funds):
+        """
+        年化波动率过滤器
+        @return:
+        """
+        filtered = []
+        fund_ids = [fund['id'] for fund in funds]
+        pct_change = pd.DataFrame(self.get_pct_change(fund_ids, day))
+        pct_change.set_index('date', inplace=True)
+        ratio = annual_volatility(
+            pct_change.truncate(before=(day - relativedelta(**self.annual_volatility_section[0]))))
+        ratio = pd.Series(ratio).to_dict()
+        annual = {fund_id: value for fund_id in fund_ids for value in ratio.values()}
+        self.save_annual(day, annual)
+        filters = self.annual_volatility_filter
+        for f in filters:
+            customType = f.get('customType')
+            exclude = f.get('exclude')
+            volatility = f.get('volatility')
+            records = [fund for fund in funds if fund['customType'] == customType and fund['id'] in annual.keys()]
+            if exclude:
+                exclude = exclude if len(records) > exclude else len(records)
+                records = records[0:-exclude]
+            if volatility:
+                records = [record for record in records if annual.get(record['id']) > volatility]
+            filtered.extend(records)
+        return funds
+
+    def save_annual(self, day, annual):
+        datas = []
+        for key, record in annual.items():
+            data = {
+                "id": key,
+                "date": day,
+                "annual": record,
+            }
+            datas.append(data)
+        robo_indicator.insert(datas)
+
     def get_filtered_funds(self, day):
         funds = self._datum.get_datums(type=DatumType.FUND)
         if get_config('portfolios.checker.month-fund-filter'):
@@ -149,6 +203,7 @@ class FundDividendSortinoAssetOptimize(SortinoAssetOptimize):
                 if fund[filters] in self.asset_filter[filters]:
                     funds_in.append(fund)
             return funds_in
+        funds = self.do_annual_volatility_filter(day, funds)
         return funds
 
     def get_groups(self, day=None):
@@ -167,8 +222,9 @@ class FundDividendSortinoAssetOptimize(SortinoAssetOptimize):
     def get_pct_change(self, fund_ids, day):
         if not self._config:
             raise exception(f"find optimize, but not found sortino config.")
-        start = filter_weekend(
-            sorted([day - relativedelta(days=1, **dict_remove(x, ('weight', 'name'))) for x in self._config])[0])
+        days = [day - relativedelta(days=1, **dict_remove(x, ('weight', 'name'))) for x in self._config]
+        days.append(day - relativedelta(days=1, **self.annual_volatility_section[0]))
+        start = filter_weekend(sorted(days)[0])
         fund_navs = pd.DataFrame(self._navs.get_fund_navs(fund_ids=tuple(fund_ids), min_date=start, max_date=day))
         if not fund_navs.empty:
             fund_navs.sort_values('nav_date', inplace=True)

asset_pool/dao/mysql.sql

@@ -13,3 +13,13 @@ CREATE TABLE IF NOT EXISTS robo_assets_pool
   AUTO_INCREMENT = 0
   DEFAULT CHARSET = utf8mb4 COMMENT '资产池';
 
+CREATE TABLE IF NOT EXISTS robo_indicator
+(
+  `ri_rbd_id` bigint(20) NOT NULL,
+  `ri_date` datetime NOT NULL,
+  `ri_annual` double NOT NULL,
+  `ri_sortino` json NULL,
+  `ri_create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  `ri_update_time` datetime NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
+  UNIQUE INDEX `ri_rbd_id`(`ri_rbd_id`, `ri_date`) USING BTREE
+) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;
\ No newline at end of file

asset_pool/dao/robo_indicator.py

+from py_jftech import write, mapper_columns
+
+__COLUMNS__ = {
+    'ri_rbd_id': 'id',
+    'ri_date': 'date',
+    'ri_annual': 'annual',
+    'ri_sortino': 'sortino',
+}
+
+
+@write
+def insert(datas):
+    datas = [mapper_columns(datas=x, columns=__COLUMNS__, ignore_none=False) for x in datas]
+    values = ','.join(
+        [f'''({','.join([(f"'{x[j]}'" if j in x and x[j] is not None else 'null') for j in __COLUMNS__.keys()])})''' for
+         x in datas])
+    return f'''insert into robo_indicator({','.join(__COLUMNS__.keys())}) values {values}'''
+
+
+@write
+def update_sortino(id, date, sortino):
+    return f'''update robo_indicator set ri_sortino='{sortino}' where ri_rbd_id={id} and ri_date='{date}' '''
+
+
+@write
+def clear():
+    return 'TRUNCATE robo_indicator'

asset_pool/pool.py

@@ -3,7 +3,7 @@ from datetime import datetime as dt
 from py_jftech import component, autowired
 
 from api import AssetPool, AssetOptimize
-from asset_pool.dao import robo_assets_pool as rap
+from asset_pool.dao import robo_assets_pool as rap, robo_indicator
 
 
 @component
@@ -18,3 +18,4 @@ class FundAssetPool(AssetPool):
 
     def clear(self, day=None):
         rap.delete(day)
+        robo_indicator.clear()

config-svrobo_Mdiv_PRR3.yml

@@ -78,6 +78,15 @@ asset-pool: # 资产池模块
         weight: 0.2
     asset-include: {'customType':[1,2,3,4]}
     optimize-count: 3 #基金优选个数
+    annual-volatility-filter: #1各资产年化波动率末exclude位 2各资产年化波动率大于volatility
+      - customType: 1
+        exclude: 2
+        volatility: 0
+      - customType: 2
+        exclude: 1
+        volatility: 0
+    annual-volatility-section:  # 波动率时间区间
+      - years: 3
 portfolios: # 投组模块
   holder: # 持仓投组相关
     init-nav: 100 # 初始金额
@@ -243,7 +252,7 @@ robo-executor: # 执行器相关
     start-date: 2023-01-02 # 回测起始日期
     end-date: 2023-10-31 # 回测截止日期
     sealing-period: 10 #调仓封闭期
-    start-step: ${BACKTEST_START_STEP:2} # 回测从哪一步开始执行 1:计算资产池；2:计算最优投组：3:计算再平衡信号以及持仓投组
+    start-step: ${BACKTEST_START_STEP:1} # 回测从哪一步开始执行 1:计算资产池；2:计算最优投组：3:计算再平衡信号以及持仓投组
     end-step: ${BACKTEST_END_STEP:3} # 回测从哪一步执行完成后结束执行 1:计算资产池；2:计算最优投组：3:计算再平衡信号以及持仓投组
     clean-up: on
   real: # 实盘执行器

portfolios/solver.py

@@ -195,7 +195,8 @@ class DefaultSolver(Solver):
     def solve_risk_parity(self):
         model = self.create_model()
         model.objective = Objective(expr=sum(
-            [(model.z[i] * model.w[i] * (self.risk_parity_sigma.iloc[i] @ model.w) - model.z[j] * model.w[j] * (self.risk_parity_sigma.iloc[j] @ model.w)) ** 2
+            [(model.z[i] * model.w[i] * (self.risk_parity_sigma.iloc[i] @ model.w) - model.z[j] * model.w[j] * (
+                        self.risk_parity_sigma.iloc[j] @ model.w)) ** 2
              for i in model.indices for j in model.indices]), sense=minimize)
         self._solver.solve(model)
         return self.calc_port_weight(model)
@@ -451,7 +452,8 @@ class PRRSolver(ARCSolver):
 
         model.cons_RR_LE_TRR = Constraint(
             expr=sum([model.w[i] * RR_LE_TRR[i] for i in model.indices]) >= minRRweightWithinTRR)
-
+        # todo trr<指定值
+        # model.cons_LE_TRR = Constraint(expr=sum([model.w[i] * RR[i] for i in model.indices]) <= TRR)
         if TRR < 5:
             model.cons_RR_in_1_5 = Constraint(
                 expr=sum([model.z[i] * (RR_in_1_5[i] * self.max_count - RR_EQ_5[i]) for i in model.indices]) >= 0)

portfolios/utils.py

 import pandas as pd
+from py_jftech import autowired
 
+from api import DatumType, Datum
 
-def format_weight(weight: dict, to=1) -> dict:
+
+@autowired
+def format_weight(weight: dict, to=1, datum: Datum = None) -> dict:
     """
     对权重的小数点进行截取，到指定权重
+    @param datum:
     @param weight:
     @param to: 指定权重
     @return:
     """
+    # funds = datum.get_datums(type=DatumType.FUND)
+    # risk_dict = {fund['id']: fund['risk'] for fund in funds}
+    # risk = 0
+    # for k, v in weight.items():
+    #     risk += risk_dict.get(int(k)) * v
+    # print(risk)
     weight_series = pd.Series(weight)
     weight_series = weight_series.fillna(0)
-    minidx = weight_series[weight_series > 0].idxmin()
-    maxidx = weight_series.idxmax()
     weight_series = weight_series.apply(lambda x: round(x, 2))
     if weight_series.sum() == to:
         return dict(weight_series)
-    elif weight_series.sum() < to:
+    funds = datum.get_datums(type=DatumType.FUND)
+    risk_dict = {fund['id']: fund['risk'] for fund in funds}
+    id_sort = sorted(weight_series.to_dict().keys(), key=lambda x: risk_dict.get(int(x)))
+    # 低风险
+    minidx = id_sort[0]
+    # 高风险
+    maxidx = id_sort[-1]
+    if weight_series.sum() < to:
         weight_series[minidx] += to - weight_series.sum()
     elif weight_series.sum() > to:
         weight_series[maxidx] += to - weight_series.sum()
@@ -23,4 +39,4 @@ def format_weight(weight: dict, to=1) -> dict:
 
 
 if __name__ == '__main__':
-    print(format_weight({19: 0.13, 27: 0.17, 56: 0.36}))
+    format_weight({"5": 0.35, "6": 0.35, "10": 0.09, "11": 0.16, "22": 0.05})