module fundit::task_fundPerformance
use fundit::sqlUtilities;
use fundit::operationDataPuller;
use fundit::performanceDataPuller;
use fundit::dataSaver;
use fundit::returnCalculator;
use fundit::indicatorCalculator;
use fundit::ms_dataPuller;
/*
* [定时任务]: 同步公募，私募，市场指数的净值并存入本地数据库
*
* @param date <DATETIME>
*
* Example: GetEntityNavTask(2024.09.01); // 【初始化专用】
*/
def GetEntityNavTask(date) {
for(entity_type in ['HF', 'MI', 'MF']) {
des = get_nav_table_description(entity_type)[0];
t_local_table = load_table_from_local('fundit', des.table_name);
entity_date = get_entity_list_by_nav_updatetime(entity_type, null, date, true);
if(entity_date.isVoid() || entity_date.size() == 0) return;
v_entity_id = entity_date.entity_id.distinct();
i = 0;
batch_size = 1000;
max_cnt = v_entity_id.size();
cnt_num = 0;
do {
tb_entity = SELECT * FROM entity_date WHERE entity_id IN v_entity_id[i : min(max_cnt, i+batch_size)];
s_json = tb_entity.toStdJson();
tb_nav = sync_entity_nav_by_date(entity_type, s_json);
if(!(tb_nav.isVoid() || tb_nav.size() == 0)) {
try {
t_local_table.append!(tb_nav);
} catch(ex) {
writeLogLevel(ERROR, ex);
}
writeLogLevel(INFO, des.table_name + ': UPDATE ' + tb_nav.size()$STRING);
}
i += batch_size;
} while (i < max_cnt);
}
}
/*
* [定时任务]：最新净值触发的业绩指标计算
*
* @param entityType <STRING>: 'MF', 'HF'...
* @param date <DATETIME>: 净值更新时间, 为空时缺省为当前时间-1天；为1989.01.01或更早日期时代表初始化，指标会被存入本地数据库
*
* NOTE: 1) 与Java不同的是当月indicator计算每日触发，不必等到Month-end production
* 2) latest performance 尝试用 dolphin 本地数据
*
* Example: calFundPerformanceTask('HF', 2020.01.23);
* calFundPerformanceTask('MI', 2020.10.28);
* calFundPerformanceTask('FI', 2024.10.28);
* calFundPerformanceTask('MF', get_ini_data_const()['date']); -- 【初始化数据专用】(100min)
*/
def calFundPerformanceTask(entityType, date) {
rt = '';
if(!(entityType IN ['MF', 'HF', 'MI', 'FI'])) return null;
if(date.isNothing() || date.isNull())
end_day = temporalAdd(now(), -1d);
else
end_day = date;
// 取有最新净值变动的基金列表 (2.5 min)
tb_cal_funds = get_entity_list_by_nav_updatetime(entityType, NULL, end_day, true);
if(tb_cal_funds.isVoid() || tb_cal_funds.size() == 0 ) return;
// 分批跑
i = 0;
batch_size = 200;
do {
// 按照 MySQL 建好各表
tb_fund_performance = create_entity_performance();
tb_fund_indicator = create_entity_indicator();
tb_fund_risk_stats = create_entity_risk_stats();
tb_fund_riskadjret_stats = create_entity_riskadjret_stats();
tb_fund_style_stats = create_entity_style_stats();
tb_fund_ms_stats = create_entity_ms_stats();
tb_fund_performance_weekly = create_entity_performance_weekly();
tb_fund_latest_performance = create_entity_latest_performance();
funds = tb_cal_funds[i : min(tb_cal_funds.size(), i+batch_size)];
// funds = SELECT * FROM tb_cal_funds WHERE entity_id = 'MF00004QR9'
if(funds.isVoid() || funds.size() == 0) break;
// 200ms
fund_info = SELECT entity_id, price_date, inception_date, benchmark_id, ini_value
FROM ej(funds, get_entity_info(entityType, funds.entity_id), 'entity_id');
// 计算月收益 (17s)
rets = mix_monthly_returns(entityType, fund_info);
if(!rets.isVoid() && rets.size() > 0) {
// 计算月度指标 (78s)
rets.rename!('cumulative_nav', 'nav');
indicators = cal_monthly_indicators(entityType, 'PBI', rets);
// 仿照MySQL的表结构准备好记录 （1s)
generate_entity_performance(fund_info, indicators, true, tb_fund_performance);
generate_entity_indicator(fund_info, indicators, true, tb_fund_indicator);
generate_entity_risk_stats(fund_info, indicators, true, tb_fund_risk_stats);
generate_entity_riskadjret_stats(fund_info, indicators, true, tb_fund_riskadjret_stats);
generate_entity_style_stats(fund_info, indicators, true, tb_fund_style_stats);
generate_entity_ms_stats(fund_info, indicators, true, tb_fund_ms_stats);
}
// 计算周收益 (8s)
rets_w = cal_weekly_returns(entityType, fund_info);
if(! rets_w.isVoid() && rets_w.size() > 0) {
generate_entity_performance_weekly(fund_info, rets_w, true, tb_fund_performance_weekly);
}
// 计算最新收益 (2s using local, 62s using mysql)
perf_latest = cal_latest_performance(entityType, fund_info, false);
if(! perf_latest.isVoid() && perf_latest.size() > 0) {
generate_entity_latest_performance(fund_info, perf_latest, true, tb_fund_latest_performance);
}
if(! tb_fund_performance.isVoid() && tb_fund_performance.size() > 0) {
// save data to MySQL (5s)
try {
chg_columns_for_mysql(tb_fund_performance, 'fund_id');
save_and_sync(tb_fund_performance, 'raw_db.fund_performance', 'raw_db.fund_performance');
chg_columns_for_mysql(tb_fund_indicator, 'fund_id');
save_and_sync(tb_fund_indicator, 'raw_db.fund_indicator', 'raw_db.fund_indicator');
chg_columns_for_mysql(tb_fund_risk_stats, 'fund_id');
// mfdb.fund_performance 表中 maxdrawdown_6m & maxdrawdown_ytd 是虚拟列，这里用数据列顺序强行写入真实列 6m_maxdrawdown & ytd_maxdrawdown (DolphinDB 不允许字段名以数字开头)
save_and_sync(tb_fund_risk_stats, 'raw_db.fund_risk_stats', 'raw_db.fund_risk_stats');
chg_columns_for_mysql(tb_fund_riskadjret_stats, 'fund_id');
save_and_sync(tb_fund_riskadjret_stats, 'raw_db.fund_riskadjret_stats', 'raw_db.fund_riskadjret_stats');
chg_columns_for_mysql(tb_fund_style_stats, 'fund_id');
save_and_sync(tb_fund_style_stats, 'raw_db.fund_style_stats', 'raw_db.fund_style_stats');
chg_columns_for_mysql(tb_fund_ms_stats, 'fund_id');
save_and_sync(tb_fund_ms_stats, 'raw_db.fund_ms_stats', 'mfdb.fund_ms_stats'); // new table, write into directly
save_and_sync(tb_fund_performance_weekly, 'raw_db.fund_performance_weekly', 'raw_db.fund_performance_weekly');
save_and_sync(tb_fund_latest_performance, 'raw_db.fund_latest_performance', 'raw_db.fund_latest_nav_performance');
// 数据初始化时将指标存入本地
if(end_day <= get_ini_data_const()['date']) {
save_table(tb_fund_performance, 'mfdb.fund_performance', false);
save_table(tb_fund_indicator, 'mfdb.fund_indicator', false);
save_table(tb_fund_risk_stats, 'mfdb.fund_risk_stats', false);
save_table(tb_fund_riskadjret_stats, 'mfdb.fund_riskadjret_stats', false);
save_table(tb_fund_style_stats, 'mfdb.fund_style_stats', false);
save_table(tb_fund_ms_stats, 'mfdb.fund_ms_stats', false);
save_table(tb_fund_performance_weekly, 'mfdb.fund_performance_weekly', false);
save_table(tb_fund_latest_performance, 'mfdb.fund_latest_performance', false);
}
} catch(ex) {
//TODO: Log errors
rt = ex;
}
}
i += batch_size;
} while (i <= tb_cal_funds.size());
return rt;
}
/*
* 【临时】用于数据初始化：只计算收益
*
* @param entityType <STRING>: 'MF', 'HF'...
* @param date <DATETIME>: 净值更新时间
*
*/
def ms_calFundReturns() {
rt = '';
very_old_date = 1990.01.01;
// 取基金列表 (27s)
tb_cal_funds = ms_get_fund_list_by_nav_updatetime(NULL, very_old_date);
if(tb_cal_funds.isVoid() || tb_cal_funds.size() == 0 ) return;
tb_fund_performance = create_entity_performance();
tb_fund_indicator = create_entity_indicator();
tb_fund_risk_stats = create_entity_risk_stats();
tb_fund_riskadjret_stats = create_entity_riskadjret_stats();
tb_fund_style_stats = create_entity_style_stats();
tb_fund_performance_weekly = create_entity_performance_weekly();
tb_fund_latest_performance = create_entity_latest_performance();
// 分批跑
i = 0;
batch_size = 1000;
do {
funds = tb_cal_funds[i : min(tb_cal_funds.size(), i+batch_size)];
if(funds.isVoid() || funds.size() == 0) break;
// 200ms
fund_info = SELECT entity_id, price_date, inception_date, benchmark_id, ini_value
FROM ej(funds, ms_get_fund_info(funds.entity_id), 'entity_id', 'fund_id');
// 计算月收益 (19s)
tb_nav = ms_get_fund_monthly_nav(fund_info.entity_id);
rets = cal_monthly_returns_by_nav(fund_info, tb_nav);
if(!rets.isVoid() && rets.size() > 0) {
// 计算月度指标 (67s)
rets.rename!('cumulative_nav', 'nav');
indicators = cal_monthly_indicators('MF', 'PBI', rets);
// 仿照MySQL的表结构准备好记录 （1s)
generate_entity_performance(fund_info, indicators, true, tb_fund_performance);
generate_entity_indicator(fund_info, indicators, true, tb_fund_indicator);
generate_entity_risk_stats(fund_info, indicators, true, tb_fund_risk_stats);
generate_entity_riskadjret_stats(fund_info, indicators, true, tb_fund_riskadjret_stats);
generate_entity_style_stats(fund_info, indicators, true, tb_fund_style_stats);
}
// 计算周收益 (49s)
rets_w = cal_weekly_returns('MF', fund_info);
if(! rets_w.isVoid() && rets_w.size() > 0) {
generate_entity_performance_weekly(fund_info, rets_w, true, tb_fund_performance_weekly);
}
// 计算最新收益 (23s)
perf_latest = cal_latest_performance('MF', fund_info, true);
if(! perf_latest.isVoid() && perf_latest.size() > 0) {
generate_entity_latest_performance(fund_info, perf_latest, true, tb_fund_latest_performance);
}
i += batch_size;
// } while (i < batch_size);
} while (i <= tb_cal_funds.size());
if(! tb_fund_performance.isVoid() && tb_fund_performance.size() > 0) {
// save data to MySQL (26m)
try {
chg_columns_for_mysql(tb_fund_performance, 'fund_id');
save_and_sync(tb_fund_performance, 'raw_db.fund_performance', 'raw_db.fund_performance');
chg_columns_for_mysql(tb_fund_indicator, 'fund_id');
save_and_sync(tb_fund_indicator, 'raw_db.fund_indicator', 'raw_db.fund_indicator');
chg_columns_for_mysql(tb_fund_risk_stats, 'fund_id');
// mfdb.fund_performance 表中 maxdrawdown_6m & maxdrawdown_ytd 是虚拟列，这里用数据列顺序强行写入真实列 6m_maxdrawdown & ytd_maxdrawdown (DolphinDB 不允许字段名以数字开头)
save_and_sync(tb_fund_risk_stats, 'raw_db.fund_risk_stats', 'raw_db.fund_risk_stats');
chg_columns_for_mysql(tb_fund_riskadjret_stats, 'fund_id');
save_and_sync(tb_fund_riskadjret_stats, 'raw_db.fund_riskadjret_stats', 'raw_db.fund_riskadjret_stats');
chg_columns_for_mysql(tb_fund_style_stats, 'fund_id');
save_and_sync(tb_fund_style_stats, 'raw_db.fund_style_stats', 'raw_db.fund_style_stats');
save_and_sync(tb_fund_performance_weekly, 'raw_db.fund_performance_weekly', 'raw_db.fund_performance_weekly');
save_and_sync(tb_fund_latest_performance, 'raw_db.fund_latest_performance', 'raw_db.fund_latest_nav_performance');
} catch(ex) {
//TODO: Log errors
rt = ex;
}
}
return rt;
}