task_fundPerformance.dos 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331
  1. module fundit::task_fundPerformance
  2. use fundit::sqlUtilities;
  3. use fundit::operationDataPuller;
  4. use fundit::performanceDataPuller;
  5. use fundit::dataSaver;
  6. use fundit::returnCalculator;
  7. use fundit::indicatorCalculator;
  8. use fundit::ms_dataPuller;
  9. /*
  10. * [定时任务]: 同步公募,私募,市场指数的净值并存入本地数据库
  11. *
  12. * @param date <DATETIME>
  13. *
  14. * Example: GetEntityNavTask(2024.09.01); // 【初始化专用】
  15. */
  16. def GetEntityNavTask(date) {
  17. for(entity_type in ['HF', 'MI', 'MF']) {
  18. des = get_nav_table_description(entity_type)[0];
  19. t_local_table = load_table_from_local('fundit', des.table_name);
  20. entity_date = get_entity_list_by_nav_updatetime(entity_type, null, date, true);
  21. if(entity_date.isVoid() || entity_date.size() == 0) return;
  22. v_entity_id = entity_date.entity_id.distinct();
  23. i = 0;
  24. batch_size = 1000;
  25. max_cnt = v_entity_id.size();
  26. cnt_num = 0;
  27. do {
  28. tb_entity = SELECT * FROM entity_date WHERE entity_id IN v_entity_id[i : min(max_cnt, i+batch_size)];
  29. s_json = tb_entity.toStdJson();
  30. tb_nav = sync_entity_nav_by_date(entity_type, s_json);
  31. if(!(tb_nav.isVoid() || tb_nav.size() == 0)) {
  32. try {
  33. t_local_table.append!(tb_nav);
  34. } catch(ex) {
  35. writeLogLevel(ERROR, ex);
  36. }
  37. writeLogLevel(INFO, des.table_name + ': UPDATE ' + tb_nav.size()$STRING);
  38. }
  39. i += batch_size;
  40. } while (i < max_cnt);
  41. }
  42. }
  43. /*
  44. * [定时任务]:最新净值触发的业绩指标计算
  45. *
  46. * @param entityType <STRING>: 'MF', 'HF'...
  47. * @param date <DATETIME>: 净值更新时间, 为空时缺省为当前时间-1天;为1989.01.01或更早日期时代表初始化,指标会被存入本地数据库
  48. *
  49. * NOTE: 1) 与Java不同的是当月indicator计算每日触发,不必等到Month-end production
  50. * 2) latest performance 尝试用 dolphin 本地数据
  51. *
  52. * Example: calFundPerformanceTask('HF', 2020.01.23);
  53. * calFundPerformanceTask('MI', 2020.10.28);
  54. * calFundPerformanceTask('FI', 2024.10.28);
  55. * calFundPerformanceTask('MF', get_ini_data_const()['date']); -- 【初始化数据专用】(100min)
  56. */
  57. def calFundPerformanceTask(entityType, date) {
  58. rt = '';
  59. if(!(entityType IN ['MF', 'HF', 'MI', 'FI'])) return null;
  60. if(date.isNothing() || date.isNull())
  61. end_day = temporalAdd(now(), -1d);
  62. else
  63. end_day = date;
  64. // 取有最新净值变动的基金列表 (2.5 min)
  65. tb_cal_funds = get_entity_list_by_nav_updatetime(entityType, NULL, end_day, true);
  66. if(tb_cal_funds.isVoid() || tb_cal_funds.size() == 0 ) return;
  67. // 分批跑
  68. i = 0;
  69. batch_size = 200;
  70. do {
  71. // 按照 MySQL 建好各表
  72. tb_fund_performance = create_entity_performance();
  73. tb_fund_indicator = create_entity_indicator();
  74. tb_fund_risk_stats = create_entity_risk_stats();
  75. tb_fund_riskadjret_stats = create_entity_riskadjret_stats();
  76. tb_fund_style_stats = create_entity_style_stats();
  77. tb_fund_ms_stats = create_entity_ms_stats();
  78. tb_fund_performance_weekly = create_entity_performance_weekly();
  79. tb_fund_latest_performance = create_entity_latest_performance();
  80. funds = tb_cal_funds[i : min(tb_cal_funds.size(), i+batch_size)];
  81. // funds = SELECT * FROM tb_cal_funds WHERE entity_id = 'MF00004QR9'
  82. if(funds.isVoid() || funds.size() == 0) break;
  83. // 200ms
  84. fund_info = SELECT entity_id, price_date, inception_date, benchmark_id, ini_value
  85. FROM ej(funds, get_entity_info(entityType, funds.entity_id), 'entity_id');
  86. // 计算月收益 (17s)
  87. rets = mix_monthly_returns(entityType, fund_info);
  88. if(!rets.isVoid() && rets.size() > 0) {
  89. // 计算月度指标 (78s)
  90. rets.rename!('cumulative_nav', 'nav');
  91. indicators = cal_monthly_indicators(entityType, 'PBI', rets);
  92. // 仿照MySQL的表结构准备好记录 (1s)
  93. generate_entity_performance(fund_info, indicators, true, tb_fund_performance);
  94. generate_entity_indicator(fund_info, indicators, true, tb_fund_indicator);
  95. generate_entity_risk_stats(fund_info, indicators, true, tb_fund_risk_stats);
  96. generate_entity_riskadjret_stats(fund_info, indicators, true, tb_fund_riskadjret_stats);
  97. generate_entity_style_stats(fund_info, indicators, true, tb_fund_style_stats);
  98. generate_entity_ms_stats(fund_info, indicators, true, tb_fund_ms_stats);
  99. }
  100. // 计算周收益 (8s)
  101. rets_w = cal_weekly_returns(entityType, fund_info);
  102. if(! rets_w.isVoid() && rets_w.size() > 0) {
  103. generate_entity_performance_weekly(fund_info, rets_w, true, tb_fund_performance_weekly);
  104. }
  105. // 计算最新收益 (2s using local, 62s using mysql)
  106. perf_latest = cal_latest_performance(entityType, fund_info, false);
  107. if(! perf_latest.isVoid() && perf_latest.size() > 0) {
  108. generate_entity_latest_performance(fund_info, perf_latest, true, tb_fund_latest_performance);
  109. }
  110. if(! tb_fund_performance.isVoid() && tb_fund_performance.size() > 0) {
  111. // save data to MySQL (5s)
  112. try {
  113. chg_columns_for_mysql(tb_fund_performance, 'fund_id');
  114. save_and_sync(tb_fund_performance, 'raw_db.fund_performance', 'raw_db.fund_performance');
  115. chg_columns_for_mysql(tb_fund_indicator, 'fund_id');
  116. save_and_sync(tb_fund_indicator, 'raw_db.fund_indicator', 'raw_db.fund_indicator');
  117. chg_columns_for_mysql(tb_fund_risk_stats, 'fund_id');
  118. // mfdb.fund_performance 表中 maxdrawdown_6m & maxdrawdown_ytd 是虚拟列,这里用数据列顺序强行写入真实列 6m_maxdrawdown & ytd_maxdrawdown (DolphinDB 不允许字段名以数字开头)
  119. save_and_sync(tb_fund_risk_stats, 'raw_db.fund_risk_stats', 'raw_db.fund_risk_stats');
  120. chg_columns_for_mysql(tb_fund_riskadjret_stats, 'fund_id');
  121. save_and_sync(tb_fund_riskadjret_stats, 'raw_db.fund_riskadjret_stats', 'raw_db.fund_riskadjret_stats');
  122. chg_columns_for_mysql(tb_fund_style_stats, 'fund_id');
  123. save_and_sync(tb_fund_style_stats, 'raw_db.fund_style_stats', 'raw_db.fund_style_stats');
  124. chg_columns_for_mysql(tb_fund_ms_stats, 'fund_id');
  125. save_and_sync(tb_fund_ms_stats, 'raw_db.fund_ms_stats', 'mfdb.fund_ms_stats'); // new table, write into directly
  126. save_and_sync(tb_fund_performance_weekly, 'raw_db.fund_performance_weekly', 'raw_db.fund_performance_weekly');
  127. save_and_sync(tb_fund_latest_performance, 'raw_db.fund_latest_performance', 'raw_db.fund_latest_nav_performance');
  128. // 数据初始化时将指标存入本地
  129. if(end_day <= get_ini_data_const()['date']) {
  130. save_table(tb_fund_performance, 'mfdb.fund_performance', false);
  131. save_table(tb_fund_indicator, 'mfdb.fund_indicator', false);
  132. save_table(tb_fund_risk_stats, 'mfdb.fund_risk_stats', false);
  133. save_table(tb_fund_riskadjret_stats, 'mfdb.fund_riskadjret_stats', false);
  134. save_table(tb_fund_style_stats, 'mfdb.fund_style_stats', false);
  135. save_table(tb_fund_ms_stats, 'mfdb.fund_ms_stats', false);
  136. save_table(tb_fund_performance_weekly, 'mfdb.fund_performance_weekly', false);
  137. save_table(tb_fund_latest_performance, 'mfdb.fund_latest_performance', false);
  138. }
  139. } catch(ex) {
  140. //TODO: Log errors
  141. rt = ex;
  142. }
  143. }
  144. i += batch_size;
  145. } while (i <= tb_cal_funds.size());
  146. return rt;
  147. }
  148. /*
  149. * 【临时】用于数据初始化:只计算收益
  150. *
  151. * @param entityType <STRING>: 'MF', 'HF'...
  152. * @param date <DATETIME>: 净值更新时间
  153. *
  154. */
  155. def ms_calFundReturns() {
  156. rt = '';
  157. very_old_date = 1990.01.01;
  158. // 取基金列表 (27s)
  159. tb_cal_funds = ms_get_fund_list_by_nav_updatetime(NULL, very_old_date);
  160. if(tb_cal_funds.isVoid() || tb_cal_funds.size() == 0 ) return;
  161. tb_fund_performance = create_entity_performance();
  162. tb_fund_indicator = create_entity_indicator();
  163. tb_fund_risk_stats = create_entity_risk_stats();
  164. tb_fund_riskadjret_stats = create_entity_riskadjret_stats();
  165. tb_fund_style_stats = create_entity_style_stats();
  166. tb_fund_performance_weekly = create_entity_performance_weekly();
  167. tb_fund_latest_performance = create_entity_latest_performance();
  168. // 分批跑
  169. i = 0;
  170. batch_size = 1000;
  171. do {
  172. funds = tb_cal_funds[i : min(tb_cal_funds.size(), i+batch_size)];
  173. if(funds.isVoid() || funds.size() == 0) break;
  174. // 200ms
  175. fund_info = SELECT entity_id, price_date, inception_date, benchmark_id, ini_value
  176. FROM ej(funds, ms_get_fund_info(funds.entity_id), 'entity_id', 'fund_id');
  177. // 计算月收益 (19s)
  178. tb_nav = ms_get_fund_monthly_nav(fund_info.entity_id);
  179. rets = cal_monthly_returns_by_nav(fund_info, tb_nav);
  180. if(!rets.isVoid() && rets.size() > 0) {
  181. // 计算月度指标 (67s)
  182. rets.rename!('cumulative_nav', 'nav');
  183. indicators = cal_monthly_indicators('MF', 'PBI', rets);
  184. // 仿照MySQL的表结构准备好记录 (1s)
  185. generate_entity_performance(fund_info, indicators, true, tb_fund_performance);
  186. generate_entity_indicator(fund_info, indicators, true, tb_fund_indicator);
  187. generate_entity_risk_stats(fund_info, indicators, true, tb_fund_risk_stats);
  188. generate_entity_riskadjret_stats(fund_info, indicators, true, tb_fund_riskadjret_stats);
  189. generate_entity_style_stats(fund_info, indicators, true, tb_fund_style_stats);
  190. }
  191. // 计算周收益 (49s)
  192. rets_w = cal_weekly_returns('MF', fund_info);
  193. if(! rets_w.isVoid() && rets_w.size() > 0) {
  194. generate_entity_performance_weekly(fund_info, rets_w, true, tb_fund_performance_weekly);
  195. }
  196. // 计算最新收益 (23s)
  197. perf_latest = cal_latest_performance('MF', fund_info, true);
  198. if(! perf_latest.isVoid() && perf_latest.size() > 0) {
  199. generate_entity_latest_performance(fund_info, perf_latest, true, tb_fund_latest_performance);
  200. }
  201. i += batch_size;
  202. // } while (i < batch_size);
  203. } while (i <= tb_cal_funds.size());
  204. if(! tb_fund_performance.isVoid() && tb_fund_performance.size() > 0) {
  205. // save data to MySQL (26m)
  206. try {
  207. chg_columns_for_mysql(tb_fund_performance, 'fund_id');
  208. save_and_sync(tb_fund_performance, 'raw_db.fund_performance', 'raw_db.fund_performance');
  209. chg_columns_for_mysql(tb_fund_indicator, 'fund_id');
  210. save_and_sync(tb_fund_indicator, 'raw_db.fund_indicator', 'raw_db.fund_indicator');
  211. chg_columns_for_mysql(tb_fund_risk_stats, 'fund_id');
  212. // mfdb.fund_performance 表中 maxdrawdown_6m & maxdrawdown_ytd 是虚拟列,这里用数据列顺序强行写入真实列 6m_maxdrawdown & ytd_maxdrawdown (DolphinDB 不允许字段名以数字开头)
  213. save_and_sync(tb_fund_risk_stats, 'raw_db.fund_risk_stats', 'raw_db.fund_risk_stats');
  214. chg_columns_for_mysql(tb_fund_riskadjret_stats, 'fund_id');
  215. save_and_sync(tb_fund_riskadjret_stats, 'raw_db.fund_riskadjret_stats', 'raw_db.fund_riskadjret_stats');
  216. chg_columns_for_mysql(tb_fund_style_stats, 'fund_id');
  217. save_and_sync(tb_fund_style_stats, 'raw_db.fund_style_stats', 'raw_db.fund_style_stats');
  218. save_and_sync(tb_fund_performance_weekly, 'raw_db.fund_performance_weekly', 'raw_db.fund_performance_weekly');
  219. save_and_sync(tb_fund_latest_performance, 'raw_db.fund_latest_performance', 'raw_db.fund_latest_nav_performance');
  220. } catch(ex) {
  221. //TODO: Log errors
  222. rt = ex;
  223. }
  224. }
  225. return rt;
  226. }