DolphinDB/modules/task_fundPerformance.dos

module fundit::task_fundPerformance

use fundit::sqlUtilities;
use fundit::dataPuller;
use fundit::dataSaver;
use fundit::returnCalculator;
use fundit::indicatorCalculator;
use fundit::fundCalculator;
use fundit::bfiMatcher;

/*
 *   按照 XXX_performance 表结构准备数据记录
 * 
 * 
 */
def generate_entity_performance(entity_info, indicators, isToMySQL, mutable entity_performance) {

    t = null;

    if(isToMySQL) {

       if(indicators['PBI-3M'].isVoid() || indicators['PBI-3M'].size() == 0) return;

        t = SELECT entity_id, end_date, price_date, nav AS cumulative_nav, ret AS ret_1m, ret AS ret_1m_a, trailing_ret AS ret_3m, trailing_ret_a AS ret_3m_a
            FROM indicators['PBI-3M'] AS ind
            INNER JOIN entity_info fi ON ind.entity_id = fi.entity_id
            WHERE ind.end_date >= fi.price_date.month(); // 过滤掉不必更新的旧记录
        
        UPDATE t
        SET ret_6m = trailing_ret, ret_6m_a = trailing_ret_a
        FROM ej(t, indicators['PBI-6M'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET ret_1y = trailing_ret, ret_1y_a = trailing_ret_a
        FROM ej(t, indicators['PBI-1Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET ret_2y = trailing_ret, ret_2y_a = trailing_ret_a
        FROM ej(t, indicators['PBI-2Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET ret_3y = trailing_ret, ret_3y_a = trailing_ret_a
        FROM ej(t, indicators['PBI-3Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET ret_4y = trailing_ret, ret_4y_a = trailing_ret_a
        FROM ej(t, indicators['PBI-4Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET ret_5y = trailing_ret, ret_5y_a = trailing_ret_a
        FROM ej(t, indicators['PBI-5Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET ret_10y = trailing_ret, ret_10y_a = trailing_ret_a
        FROM ej(t, indicators['PBI-10Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET ret_ytd = trailing_ret, ret_ytd_a = trailing_ret_a
        FROM ej(t, indicators['PBI-YTD'], ['entity_id', 'end_date']);
        
        // 取消了 ret_incep_a_all （没意义） 和 ret_incep_a_gips (ret_incep_a 与之相等）
        UPDATE t
        SET ret_incep = trailing_ret, ret_incep_a = trailing_ret_a
        FROM ej(t, indicators['PBI-INCEP'], ['entity_id', 'end_date']);

        INSERT INTO entity_performance SELECT * FROM t;

    } else {
        
    }
}

/*
 *   按照 XXX_risk_stats 表结构准备数据记录
 * 
 * 
 */
def generate_entity_risk_stats(entity_info, indicators, isToMySQL, mutable entity_risk_stats) {

    t = null;

   if(indicators['PBI-6M'].isVoid() || indicators['PBI-6M'].size() == 0) return;

    if(isToMySQL) {

        t = SELECT entity_id, end_date, std_dev_a AS stddev_6m, ds_dev_a AS downsidedev_6m, alpha_a AS alpha_6m, winrate AS winrate_6m, beta AS beta_6m,
                                        skewness AS skewness_6m, kurtosis AS kurtosis_6m, wrst_month AS worstmonth_6m, drawdown AS maxdrawdown_6m // mfdb中的真实字段名是 6m_maxdrawdown
            FROM indicators['PBI-6M'] AS ind
            INNER JOIN entity_info fi ON ind.entity_id = fi.entity_id
            WHERE ind.end_date >= fi.price_date.month(); // 过滤掉不必更新的旧记录
        
        UPDATE t
        SET stddev_1y = std_dev_a, downsidedev_1y = ds_dev_a, alpha_1y = alpha_a, winrate_1y = winrate, beta_1y = beta,
            skewness_1y = skewness, kurtosis_1y = kurtosis, worstmonth_1y = wrst_month, maxdrawdown_1y = drawdown
        FROM ej(t, indicators['PBI-1Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET stddev_2y = std_dev_a, downsidedev_2y = ds_dev_a, alpha_2y = alpha_a, winrate_2y = winrate, beta_2y = beta,
            skewness_2y = skewness, kurtosis_2y = kurtosis, worstmonth_2y = wrst_month, maxdrawdown_2y = drawdown
        FROM ej(t, indicators['PBI-2Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET stddev_3y = std_dev_a, downsidedev_3y = ds_dev_a, alpha_3y = alpha_a, winrate_3y = winrate, beta_3y = beta,
            skewness_3y = skewness, kurtosis_3y = kurtosis, worstmonth_3y = wrst_month, maxdrawdown_3y = drawdown
        FROM ej(t, indicators['PBI-3Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET stddev_4y = std_dev_a, downsidedev_4y = ds_dev_a, alpha_4y = alpha_a, winrate_4y = winrate, beta_4y = beta,
            skewness_4y = skewness, kurtosis_4y = kurtosis, worstmonth_4y = wrst_month, maxdrawdown_4y = drawdown
        FROM ej(t, indicators['PBI-4Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET stddev_5y = std_dev_a, downsidedev_5y = ds_dev_a, alpha_5y = alpha_a, winrate_5y = winrate, beta_5y = beta,
            skewness_5y = skewness, kurtosis_5y = kurtosis, worstmonth_5y = wrst_month, maxdrawdown_5y = drawdown
        FROM ej(t, indicators['PBI-5Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET stddev_10y = std_dev_a, downsidedev_10y = ds_dev_a, alpha_10y = alpha_a, winrate_10y = winrate, beta_10y = beta,
            skewness_10y = skewness, kurtosis_10y = kurtosis, worstmonth_10y = wrst_month, maxdrawdown_10y = drawdown
        FROM ej(t, indicators['PBI-10Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET stddev_ytd = std_dev_a, downsidedev_ytd = ds_dev_a, alpha_ytd = alpha_a, winrate_ytd = winrate, beta_ytd = beta,
            skewness_ytd = skewness, kurtosis_ytd = kurtosis, worstmonth_ytd = wrst_month, maxdrawdown_ytd = drawdown // mfdb中的真实字段名是 ytd_maxdrawdown
        FROM ej(t, indicators['PBI-YTD'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET stddev_incep = std_dev_a, downsidedev_incep = ds_dev_a, alpha_incep = alpha_a, winrate_incep = winrate, beta_incep = beta,
            skewness_incep = skewness, kurtosis_incep = kurtosis, worstmonth_incep = wrst_month, maxdrawdown_incep = drawdown
        FROM ej(t, indicators['PBI-INCEP'], ['entity_id', 'end_date']);
        
        INSERT INTO entity_risk_stats SELECT * FROM t;

    } else {
        
    }
}

/*
 *   按照 XXX_riskadjret_stats 表结构准备数据记录
 * 
 * 
 */
def generate_entity_riskadjret_stats(entity_info, indicators, isToMySQL, mutable entity_riskadjret_stats) {

    t = null;

   if(indicators['PBI-6M'].isVoid() || indicators['PBI-6M'].size() == 0) return;

    if(isToMySQL) {

        t = SELECT entity_id, end_date,
                   sharpe_a AS sharperatio_6m, sortino_a AS sortinoratio_6m, treynor AS treynorratio_6m, jensen_a AS jensen_6m,
                   calmar AS calmarratio_6m, omega AS omegaratio_6m, kappa AS kapparatio_6m
            FROM indicators['PBI-6M'] AS ind
            INNER JOIN entity_info fi ON ind.entity_id = fi.entity_id
            WHERE ind.end_date >= fi.price_date.month(); // 过滤掉不必更新的旧记录
        
        UPDATE t
        SET sharperatio_1y = sharpe_a, sortinoratio_1y = sortino_a, treynorratio_1y = treynor, jensen_1y = jensen_a,
            calmarratio_1y = calmar, omegaratio_1y = omega, kapparatio_1y = kappa
        FROM ej(t, indicators['PBI-1Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET sharperatio_2y = sharpe_a, sortinoratio_2y = sortino_a, treynorratio_2y = treynor, jensen_2y = jensen_a,
            calmarratio_2y = calmar, omegaratio_2y = omega, kapparatio_2y = kappa
        FROM ej(t, indicators['PBI-2Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET sharperatio_3y = sharpe_a, sortinoratio_3y = sortino_a, treynorratio_3y = treynor, jensen_3y = jensen_a,
            calmarratio_3y = calmar, omegaratio_3y = omega, kapparatio_3y = kappa
        FROM ej(t, indicators['PBI-3Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET sharperatio_4y = sharpe_a, sortinoratio_4y = sortino_a, treynorratio_4y = treynor, jensen_4y = jensen_a,
            calmarratio_4y = calmar, omegaratio_4y = omega, kapparatio_4y = kappa
        FROM ej(t, indicators['PBI-4Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET sharperatio_5y = sharpe_a, sortinoratio_5y = sortino_a, treynorratio_5y = treynor, jensen_5y = jensen_a,
            calmarratio_5y = calmar, omegaratio_5y = omega, kapparatio_5y = kappa
        FROM ej(t, indicators['PBI-5Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET sharperatio_10y = sharpe_a, sortinoratio_10y = sortino_a, treynorratio_10y = treynor, jensen_10y = jensen_a,
            calmarratio_10y = calmar, omegaratio_10y = omega, kapparatio_10y = kappa
        FROM ej(t, indicators['PBI-10Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET sharperatio_ytd = sharpe_a, sortinoratio_ytd = sortino_a, treynorratio_ytd = treynor, jensen_ytd = jensen_a,
            calmarratio_ytd = calmar, omegaratio_ytd = omega, kapparatio_ytd = kappa
        FROM ej(t, indicators['PBI-YTD'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET sharperatio_incep = sharpe_a, sortinoratio_incep = sortino_a, treynorratio_incep = treynor, jensen_incep = jensen_a,
            calmarratio_incep = calmar, omegaratio_incep = omega, kapparatio_incep = kappa
        FROM ej(t, indicators['PBI-INCEP'], ['entity_id', 'end_date']);
        
        INSERT INTO entity_riskadjret_stats SELECT * FROM t;

    } else {
        
    }
}

/*
 *   按照 XXX_indicator 表结构准备数据记录
 * 
 * 
 */
def generate_entity_indicator(entity_info, indicators, isToMySQL, mutable entity_indicator) {

    t = null;

   if(indicators['PBI-6M'].isVoid() || indicators['PBI-6M'].size() == 0) return;

    if(isToMySQL) {

        t = SELECT entity_id, end_date, info_a AS info_ratio_6m, m2_a AS m2_6m, track_error_a AS tracking_error_6m
            FROM indicators['PBI-6M'] AS ind
            INNER JOIN entity_info fi ON ind.entity_id = fi.entity_id
            WHERE ind.end_date >= fi.price_date.month(); // 过滤掉不必更新的旧记录
      
        UPDATE t
        SET info_ratio_1y = info_a, m2_1y = m2_a, tracking_error_1y = track_error_a
        FROM ej(t, indicators['PBI-1Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET info_ratio_2y = info_a, m2_2y = m2_a, tracking_error_2y = track_error_a, var_2y = var, cvar_2y = cvar
        FROM ej(t, indicators['PBI-2Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET info_ratio_3y = info_a, m2_3y = m2_a, tracking_error_3y = track_error_a, var_3y = var, cvar_3y = cvar
        FROM ej(t, indicators['PBI-3Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET info_ratio_4y = info_a, m2_4y = m2_a, tracking_error_4y = track_error_a, var_4y = var, cvar_4y = cvar
        FROM ej(t, indicators['PBI-4Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET info_ratio_5y = info_a, m2_5y = m2_a, tracking_error_5y = track_error_a, var_5y = var, cvar_5y = cvar
        FROM ej(t, indicators['PBI-5Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET info_ratio_10y = info_a, m2_10y = m2_a, tracking_error_10y = track_error_a, var_10y = var, cvar_10y = cvar
        FROM ej(t, indicators['PBI-10Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET info_ratio_ytd = info_a, m2_ytd = m2_a, tracking_error_ytd = track_error_a
        FROM ej(t, indicators['PBI-YTD'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET info_ratio_incep = info_a, m2_incep = m2_a, tracking_error_incep = track_error_a, var_incep = var, cvar_incep = cvar
        FROM ej(t, indicators['PBI-INCEP'], ['entity_id', 'end_date']);
        
        INSERT INTO entity_indicator SELECT * FROM t;

    } else {
        
    }
}

/*
 *   按照 XXX_style_stats 表结构准备数据记录
 * 
 * 
 */
def generate_entity_style_stats(entity_info, indicators, isToMySQL, mutable entity_style_stats) {

    t = null;

   if(indicators['PBI-6M'].isVoid() || indicators['PBI-6M'].size() == 0) return;

    if(isToMySQL) {

        t = SELECT entity_id, end_date, upside_capture_ret AS upsidecapture_ret_6m, downside_capture_ret AS downsidecapture_ret_6m,
                                        upside_capture_ratio AS upsidecapture_ratio_6m, downside_capture_ratio AS downsidecapture_ratio_6m
            FROM indicators['PBI-6M'] AS ind
            INNER JOIN entity_info fi ON ind.entity_id = fi.entity_id
            WHERE ind.end_date >= fi.price_date.month(); // 过滤掉不必更新的旧记录
        
        UPDATE t
        SET upsidecapture_ret_1y = upside_capture_ret, downsidecapture_ret_1y = downside_capture_ret,
            upsidecapture_ratio_1y = upside_capture_ratio, downsidecapture_ratio_1y = downside_capture_ratio
        FROM ej(t, indicators['PBI-1Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET upsidecapture_ret_2y = upside_capture_ret, downsidecapture_ret_2y = downside_capture_ret,
            upsidecapture_ratio_2y = upside_capture_ratio, downsidecapture_ratio_2y = downside_capture_ratio
        FROM ej(t, indicators['PBI-2Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET upsidecapture_ret_3y = upside_capture_ret, downsidecapture_ret_3y = downside_capture_ret,
            upsidecapture_ratio_3y = upside_capture_ratio, downsidecapture_ratio_3y = downside_capture_ratio
        FROM ej(t, indicators['PBI-3Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET upsidecapture_ret_4y = upside_capture_ret, downsidecapture_ret_4y = downside_capture_ret,
            upsidecapture_ratio_4y = upside_capture_ratio, downsidecapture_ratio_4y = downside_capture_ratio
        FROM ej(t, indicators['PBI-4Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET upsidecapture_ret_5y = upside_capture_ret, downsidecapture_ret_5y = downside_capture_ret,
            upsidecapture_ratio_5y = upside_capture_ratio, downsidecapture_ratio_5y = downside_capture_ratio
        FROM ej(t, indicators['PBI-5Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET upsidecapture_ret_10y = upside_capture_ret, downsidecapture_ret_10y = downside_capture_ret,
            upsidecapture_ratio_10y = upside_capture_ratio, downsidecapture_ratio_10y = downside_capture_ratio
        FROM ej(t, indicators['PBI-10Y'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET upsidecapture_ret_ytd = upside_capture_ret, downsidecapture_ret_ytd = downside_capture_ret,
            upsidecapture_ratio_ytd = upside_capture_ratio, downsidecapture_ratio_ytd = downside_capture_ratio
        FROM ej(t, indicators['PBI-YTD'], ['entity_id', 'end_date']);
        
        UPDATE t
        SET upsidecapture_ret_incep = upside_capture_ret, downsidecapture_ret_incep = downside_capture_ret,
            upsidecapture_ratio_incep = upside_capture_ratio, downsidecapture_ratio_incep = downside_capture_ratio
        FROM ej(t, indicators['PBI-INCEP'], ['entity_id', 'end_date']);
        
        INSERT INTO entity_style_stats SELECT * FROM t;

    } else {
        
    }
}


/*
 *   按照 XXX_performance_weekly 表结构准备数据记录
 * 
 * 
 */
def generate_entity_performance_weekly(entity_info, ret_w, isToMySQL, mutable entity_performance_weekly) {

    t = null;

   if(ret_w.isVoid() || ret_w.size() == 0) return;

    if(isToMySQL) {

        t = SELECT entity_id, year_week, year_week.left(4)$INT AS end_year, year_week.right(2)$INT AS week_of_year, price_date,
                   cumulative_nav, ret_1w
            FROM ret_w r
            INNER JOIN entity_info fi ON r.entity_id = fi.entity_id
            WHERE r.price_date >= fi.price_date; // 过滤掉不必更新的旧记录
        
        INSERT INTO entity_performance_weekly SELECT * FROM t;

    } else {
        
    }
}

/*
 *   按照 XXX_latest_performance 表结构准备数据记录
 * 
 * 
 */
def generate_entity_latest_performance(entity_info, perf_latest, isToMySQL, mutable entity_latest_performance) {

    t = null;

   if(perf_latest.isVoid() || perf_latest.size() == 0) return;

    if(isToMySQL) {

        t = SELECT r.*
            FROM perf_latest r
            INNER JOIN entity_info fi ON r.entity_id = fi.entity_id
            WHERE r.price_date >= fi.price_date; // 过滤掉不必更新的旧记录
        
        INSERT INTO entity_latest_performance SELECT * FROM t;

    } else {
        
    }
}


/*
 *   定时任务：最新净值触发的业绩指标计算
 * 
 *   @param entityType <STRING>: 'MF', 'HF'...
 *   @param date <DATETIME>: 净值更新时间
 * 
 *   NOTE: 与Java不同的是当月indicator计算每日触发，不必等到Month-end production
 */
def calFundPerformance(entityType, date) {

    rt = '';

    very_old_date = 1990.01.01;

    if(find(['HF', 'MF'], entityType) < 0) return null;

    // 取有最新净值变动的基金列表 (1s)
    tb_cal_funds = get_entity_list_by_nav_updatetime(entityType, NULL, date, true);

    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)];
        //funds = SELECT * FROM tb_cal_funds WHERE entity_id in ('MF00003PXI', 'MF00003PXX')

        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_fund_info(funds.entity_id), 'entity_id', 'fund_id');

        // 计算月收益 (12s)
        rets = mix_monthly_returns(entityType, fund_info);

        if(!rets.isVoid() && rets.size() > 0) {

            // 计算月度指标 (56s)
            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);

        }
        
        // 计算周收益 (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);
        }

        // 计算最新收益 (69s)
        perf_latest = cal_latest_performance(entityType, 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  (13s)
        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;
	
}



/*
 *   【临时】用于数据初始化：只计算收益
 * 
 *   @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_createtime(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;
	
}

/*
 *   实验性质的API
 * 
 * 
 */
def calFundIndexCorrelation(entityType, date) {

    if(find(['HF', 'MF'], entityType) < 0) return null;

    // 取有最新净值变动的基金列表 (1s)
    tb_cal_funds = get_entity_list_by_nav_updatetime(entityType, NULL, date, true);

    if(tb_cal_funds.isVoid() || tb_cal_funds.size() == 0 ) return;

//    tb_fund_index_coe = create_entity_index_coe();
    // (7m)
    coe = cal_entity_index_coe(entityType, tb_cal_funds[0:1000]);

    return coe;
}