DolphinDB/modules/dataPuller.dos

module fundit::dataPuller

use fundit::sqlUtilities



/*
 *  取所有一级策略
 * 
 */
def get_strategy_list() {

	s_query = "SELECT strategy_id, strategy, rasie_type AS raise_type FROM mfdb.d_strategy WHERE isvalid = 1";

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t
}


/*
 *  取所有二级策略
 * 
 */
def get_substrategy_list() {

	s_query = "SELECT substrategy_id, substrategy, raise_type FROM mfdb.d_substrategy WHERE isvalid = 1";

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t
}

/*
 *  取指数周收益
 *
 *  Example: get_index_weekly_rets("'FA00000WKG','FA00000WKH','IN0000007G'", 1990.01.01, today());
 *           get_index_weekly_rets("'IN0000000M'", 1990.01.01, 2024.10.31);
 */
def get_index_weekly_rets(index_ids, start_date, end_date) {

    s_start_date = iif(start_date.isNull(), "", " AND price_date >= '" + start_date$STRING + "'")
    s_end_date = iif(end_date.isNull(), "", " AND price_date <= '" + end_date$STRING + "'")

    s_query = "SELECT factor_id AS index_id, year_week, price_date, factor_value AS cumulative_nav, ret_1w
               FROM pfdb.cm_factor_performance_weekly
               WHERE isvalid = 1
                 AND factor_id IN (" + index_ids + ")" +
                 s_start_date +
                 s_end_date + "
                 AND ret_1w IS NOT NULL
               UNION
               SELECT fund_id AS index_id, year_week, price_date, cumulative_nav, ret_1w
               FROM mfdb.fund_performance_weekly
               WHERE isvalid = 1
                 AND fund_id IN (" + index_ids + ")" +
                 s_start_date +
                 s_end_date + "
                 AND ret_1w IS NOT NULL
               ORDER BY year_week"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t
}


/*
 *  取基金周收益
 *
 *
 *  get_fund_weekly_rets("'MF00003TMH','MF00003UQM'", 1990.01.01, null, true)
 */
def get_fund_weekly_rets(fund_ids, start_date, end_date, isFromMySQL) {

    s_fund_id = iif(fund_ids.isNull(), "", " AND fund_id IN (" + fund_ids + ")")
    s_start_date = iif(start_date.isNull(), "", " AND price_date >= '" + start_date$STRING + "'")
    s_end_date = iif(end_date.isNull(), "", " AND price_date <= '" + end_date$STRING + "'")
    
    s_query = "SELECT fund_id, year_week, price_date, cumulative_nav, ret_1w
               FROM mfdb.fund_performance_weekly
               WHERE isvalid = 1 " + 
                 s_fund_id + 
                 s_start_date +
                 s_end_date + "
                 AND ret_1w IS NOT NULL
               ORDER BY fund_id, year_week"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t
}

/*
 *   通用取周收益
 * 
 */
def get_entity_weekly_rets(entity_type, entity_info) {

    rets = null;

    if(entity_info.isVoid() || entity_info.size() == 0) return rets;

    very_old_date = '1990.01.01';

    // 简单起见，取整个数据集的最新日期（month-end production时取上月最后一天即可
    end_day = entity_info.price_date.max();

    s_entity_ids = ids_to_string(entity_info.entity_id);

    if(entity_type == 'HF' || entity_type == 'MF') {

    	rets = get_fund_weekly_rets(s_entity_ids, very_old_date, end_day, true);

    } else if(entity_type == 'MI' || entity_type == 'FA') {

    	rets = get_index_weekly_rets(s_entity_ids, very_old_date, end_day);
    }

    return rets;

}


/*
 *  取组合周收益
 *  TODO: 增加从本地取数据的功能
 *
 *
 *  get_portfolio_weekly_rets("166002,364640", 1990.01.01, today(), true)
 */
def get_portfolio_weekly_rets(portfolio_ids, start_date, end_date, isFromMySQL) {

    s_portfolio_id = iif(portfolio_ids.isNull(), "", " AND portfolio_id IN (" + portfolio_ids + ")")

    s_query = "SELECT portfolio_id, year_week, price_date, cumulative_nav, ret_1w
               FROM pfdb.pf_portfolio_performance_weekly
               WHERE isvalid = 1 " +
                 s_portfolio_id + "
                 AND ret_1w IS NOT NULL
                 AND price_date BETWEEN '" + start_date$STRING + "' AND '" + end_date$STRING + "'
               ORDER BY portfolio_id, year_week"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t
}

/*
 *  通用取月收益
 *  
 *  @param entity_type <STRING>:
 *  @param entity_ids <VECTOR|STRING>:
 *  @param start_date <DATE>:
 *  @param end_date <DATE>:
 *  @param isFromMySQL <BOOL>:
 * 
 * 
 *  Example: get_monthly_ret('HF', ['HF000004KN','HF000103EU','HF00018WXG'], 2000.01.01, 2024.03.01, true);
 */
def get_monthly_ret(entity_type, entity_ids, start_date, end_date, isFromMySQL) {

    s_entity_ids = ids_to_string(entity_ids);

    if(s_entity_ids == null || s_entity_ids == '') return null;

    tmp = get_performance_table_description(entity_type);

    yyyymm_start = start_date.temporalFormat("yyyy-MM")
    yyyymm_end = end_date.temporalFormat("yyyy-MM")

    if(isFromMySQL == true) {

        s_query = "SELECT " + tmp.sec_id_col[0] + " AS entity_id, end_date, price_date, ret_1m AS ret, " + tmp.cumulative_nav_col[0] + " AS nav
                   FROM " + tmp.table_name[0] + "
                   WHERE " + tmp.sec_id_col[0] + " IN (" + s_entity_ids + ")
                      AND isvalid = 1
                      AND end_date BETWEEN '" + yyyymm_start + "' AND '" + yyyymm_end + "'
                   ORDER BY " + tmp.sec_id_col[0] + ", end_date";
     
        conn = connect_mysql()
     
        t = odbc::query(conn, s_query)
     
        conn.close()

    } else {
        
        tb_local = load_table_from_local("fundit", tmp.table_name[0])

        s_col = (sqlCol(tmp.sec_id_col[0]), sqlCol("end_date"), sqlColAlias(<ret_1m>, "ret"), sqlColAlias(<cumulative_nav>, "nav"), sqlCol("ret_ytd_a"), sqlCol("ret_incep_a"))
        // TODO: how to make the "fund_id" dynamicly decided by tmp.sec_id_col[0], then rename to "entity_id"?
        s_where = expr(<fund_id>, in, s_entity_ids.strReplace("'", "").split(","))
        
        t = sql(s_col, tb_local, s_where).eval()
    
    }

    return t
}

/*
 * 取无风险月度利率
 *
 * get_risk_free_rate(1990.01.01, today())
 */
def get_risk_free_rate(start_date, end_date) {
    
    return get_monthly_ret('MI', "'IN0000000M'", start_date, end_date, true);
}


/*
 * 取基金最新收益及净值
 *
 * get_fund_latest_nav_performance("'HF000004KN','HF00018WXG'")
 */
def get_fund_latest_nav_performance(fund_ids, isFromMySQL) {

    if(isFromMySQL == true) {
    
        s_query = "SELECT *
                   FROM mfdb.fund_latest_nav_performance
                   WHERE fund_id IN (" + fund_ids + ")
                     AND isvalid = 1
                     ORDER BY fund_id"

        conn = connect_mysql()
    
        t = odbc::query(conn, s_query)
    
        conn.close()

    } else {

        tb_local = load_table_from_local("fundit", "mfdb.fund_latest_nav_performance")

        s_col = sqlCol("*")

        s_where = expr(<fund_id>, in, fund_ids.strReplace("'", "").split(","))
        
        t = sql(s_col, tb_local, s_where).eval()

    }

    return t

}

/*
 * 通用取净值
 * 
 * 
 * Create: 202408                                                    Joey
 *                 TODO: add isvalid and nav > 0 for local version
 * 
 *
 * Example: get_nav_by_price_date('HF', "'HF000004KN','HF00018WXG'", 2024.05.01, true);
 *          get_nav_by_price_date('MI', "'IN00000008','IN0000000M'", 2024.05.01, true);
 */
def get_nav_by_price_date(entity_type, entity_ids, price_date, isFromMySQL) {

    s_entity_ids = ids_to_string(entity_ids);

    if(s_entity_ids == null || s_entity_ids == '') return null;

    tmp = get_nav_table_description(entity_type);

    if(isFromMySQL == true) {

        nav_table_name = tmp.table_name[0];
    
        s_query = "SELECT " + tmp.sec_id_col[0] + " AS entity_id, price_date, " + tmp.cumulative_nav_col[0] + " AS cumulative_nav, " + tmp.nav_col[0] + " AS nav
                   FROM " + tmp.table_name[0] + "
                   WHERE " + tmp.sec_id_col[0] + " IN (" + s_entity_ids + ")
                     AND isvalid = 1
                     AND " + tmp.cumulative_nav_col[0] + " > 0
                     AND price_date >= '" + price_date$STRING + "'
                   ORDER BY " + tmp.sec_id_col[0] + ", price_date";
    
        conn = connect_mysql();
    
        t = odbc::query(conn, s_query);
    
        conn.close();

    } else {
    
        tb_local = load_table_from_local("fundit", tmp.table_name[0])

        s_col = sqlCol("*")

        // TODO: how to make the "fund_id" dynamicly decided by tmp.sec_id_col[0]?
        s_where = [expr(<fund_id>, in, s_entity_ids.strReplace("'", "").split(",")), <price_date >= price_date>]
        
        t = sql(s_col, tb_local, s_where).eval()
    
    }

    return t

}


/*
 * 取指数因子点位
 *
 * get_index_nav_by_price_date("'IN00000008','FA00000WKG'", 2024.06.01)
 */
def get_index_nav_by_price_date(index_ids, price_date) {

    s_query = "SELECT index_id, price_date, close AS cumulative_nav
               FROM mfdb.market_indexes
               WHERE index_id IN (" + index_ids + ")
                 AND isvalid = 1
                 AND close > 0
                 AND price_date >= '" + price_date + "'
               UNION
               SELECT index_id AS index_id, price_date, index_value AS cumulative_nav
               FROM mfdb.indexes_ty_index
               WHERE index_id IN (" + index_ids + ")
                 AND isvalid = 1
                 AND index_value > 0
                 AND price_date >= '" + price_date + "'
               UNION
               SELECT factor_id AS index_id, price_date, factor_value AS cumulative_nav
               FROM pfdb.cm_factor_value
               WHERE factor_id IN (" + index_ids + ")
                 AND isvalid = 1
                 AND factor_value > 0
                 AND price_date >= '" + price_date + "'
               ORDER BY price_date"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t

}


/*
 * 取有效基金基本信息
 *
 * Example: get_fund_info("'HF000004KN','HF00018WXG'");
 *          get_fund_info(null);
 * 
 */
def get_fund_info(fund_ids) {

    s_entity_ids = ids_to_string(fund_ids);

    s_entity_sql = iif(s_entity_ids == NULL, '', " AND fi.fund_id IN (" + s_entity_ids + ")");

    s_query = "SELECT fi.fund_id, fi.inception_date, fi.primary_benchmark_id AS benchmark_id, IFNULL(fi.initial_unit_value, 1) AS ini_value, fs.strategy, fs.substrategy, fi.raise_type
               FROM mfdb.fund_information fi
               INNER JOIN mfdb.fund_strategy fs ON fi.fund_id = fs.fund_id
               WHERE fs.isvalid = 1
                 AND fi.isvalid = 1" + 
                 s_entity_sql + "
               ORDER BY fi.fund_id"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t

}

/*
 * 取有效指数基本信息
 *
 * Example: get_index_info("'IN00000008','IN000002GE'");
 *          get_index_info(null);
 * 
 */
def get_index_info(index_ids) {

    s_entity_ids = ids_to_string(index_ids);
    
    s_entity_sql = iif(s_entity_ids == NULL || s_entity_ids == '', '', " AND fi.index_id IN (" + s_entity_ids + ")");

    s_query = "SELECT fi.index_id, fi.inception_date, NULL AS benchmark_id, IFNULL(fi.index_initial_value, 1) AS ini_value, fi.index_code, fi.index_type_id
               FROM mfdb.indexes_profile fi
               WHERE fi.isvalid = 1" +
                 s_entity_sql + "
               ORDER BY fi.index_id";

    conn = connect_mysql();

    t = odbc::query(conn, s_query);

    conn.close();

    return t;

}

/*
 *  取组合有效信息
 * 
 *  NOTE: portfolio 的 strategy 统一为公募混合基金102， sub_strategy 用 sub_type (哪里维护的？）
 * 
 *  Example: get_portfolio_info('166002,166114');
 *           get_portfolio_info(NULL);
 *  
 */
def get_portfolio_info(portfolio_ids) {

    s_entity_ids = ids_to_string(portfolio_ids);
    
    s_entity_sql = iif(s_entity_ids == NULL || s_entity_ids == '', '', " AND cpm.id IN (" + s_entity_ids + ")");

    s_query = "SELECT cpm.id AS portfolio_id, cpm.userid, cpm.customer_id, cpm.inception_date, 1 AS ini_value, cpm.portfolio_source, cpm.portfolio_type, 102 AS strategy, sub_type AS substrategy
               FROM pfdb.`pf_customer_portfolio_map` cpm
               INNER JOIN pfdb.cm_user u ON cpm.userid = u.userid
               WHERE cpm.isvalid = 1
                 AND u.isvalid = 1" + 
                 s_entity_sql + "
               ORDER BY cpm.id"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t
}

/*
 *  取组合有效信息
 * 
 *  Example: get_cus_fund_info(['CF0000005V','CF000000CE']);
 *           get_cus_fund_info(NULL);
 *  
 */
def get_cus_fund_info(fund_ids) {

    s_entity_ids = ids_to_string(fund_ids);
    
    s_entity_sql = iif(s_entity_ids == NULL || s_entity_ids == '', '', " AND fi.fund_id IN (" + s_entity_ids + ")");

    s_query = "SELECT fi.fund_id, fi.userid, fi.inception_date, IFNULL(fi.primary_benchmark_id, 'IN00000008') AS benchmark_id, 
                      IFNULL(initial_unit_value, 1) AS ini_value, raise_type, strategy, substrategy
               FROM pfdb.pf_cus_fund_information fi
               INNER JOIN pfdb.cm_user u ON fi.userid = u.userid
               WHERE fi.isvalid = 1
                 AND u.isvalid = 1" + 
                 s_entity_sql + "
               ORDER BY fi.fund_id"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t
}

/*
 *   取基金组合基础有效信息
 * 
 *   Example: get_entity_info('HF', ['HF000004KN','HF000103EU','HF00018WXG']);
 *            get_entity_info('PF', '166002,166114');
 *            get_entity_info('MI', NULL);
 */
def get_entity_info(entity_type, entity_ids) {

    t = null;

    s_entity_ids = ids_to_string(entity_ids);

    if(entity_type == 'MF' || entity_type == 'HF') {

    	t = get_fund_info(s_entity_ids);
        t.rename!('fund_id', 'entity_id');

    } else if(entity_type == 'PF') {

    	t = get_portfolio_info(s_entity_ids);
        t.rename!('portfolio_id', 'entity_id');    

    } else if(entity_type IN ['MI', 'FI']) {

    	t = get_index_info(s_entity_ids);
        t.rename!('index_id', 'entity_id');    
   	
    } else if(entity_type == 'CF') {

    	t = get_cus_fund_info(s_entity_ids);
    	t.rename!('fund_id', 'entity_id');
    }

	return t;
}


/*
 *  取某时间后更新的各基金组合最早净值日期
 *
 *  @param entity_type <STRING>: MF, HF, EQ, CF, MI, TI, CI, FA, PF
 *  @param entity_ids <VECTOR|STRING>: NULL时取全量
 *  @param update_time <DATETIME>: all updates after this time
 *  @param isFromMySQL <BOOL>:
 *
 *  Example: get_entity_list_by_nav_updatetime('MF', ['MF00003PW1', 'MF00003PW2'], 2024.09.26, true);
 *           get_entity_list_by_nav_updatetime('HF', null, 2024.07.19T10:00:00, true);
 *           get_entity_list_by_nav_updatetime('PF', '166002,166114', 2024.06.20, true);
 * 
 */
def get_entity_list_by_nav_updatetime(entity_type, entity_ids, updatetime, isFromMySQL) {

    tmp = get_nav_table_description(entity_type);

    s_entity_ids = ids_to_string(entity_ids);

    sql_entity_id = '';

    if(s_entity_ids != NULL) {
    	sql_entity_id = " AND " + tmp.sec_id_col[0] + " IN (" + s_entity_ids + ")";
    }

    if(isFromMySQL == true) {

        nav_table_name = tmp.table_name[0];
    
        s_query = "SELECT " + tmp.sec_id_col[0] + " AS entity_id, MIN(price_date) AS price_date 
                   FROM " + tmp.table_name[0] + "
                   WHERE isvalid = 1 " +
                     sql_entity_id + "
                     AND " + tmp.cumulative_nav_col[0] + " > 0
                     AND updatetime >= '" + updatetime$STRING + "'
                   GROUP BY " + tmp.sec_id_col[0] + "
                   ORDER BY " + tmp.sec_id_col[0] + ", price_date";
    
        conn = connect_mysql();
    
        t = odbc::query(conn, s_query);
    
        conn.close();

    } else {
    	//TODO
    }

    return t

}



/*
 * 取私募基金用于月末 fund_performance 表更新的净值
 * 
 * @param fund_ids: 逗号分隔的ID字符串, 每个ID都有''
 * @param month_end: 月末日期字符串  YYYY-MM
 * 
 * 
 */
def get_nav_for_hedge_fund_performance(fund_ids, month_end) {
   
    s_query = "CALL pfdb.sp_get_nav_for_fund_performance(" + fund_ids + ", '" + month_end + "', 1);"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t

}

/*
 *  取某时间段的基金主基准
 *  NOTE: 目前数据库里只存最新的基准，以后很可能会支持时间序列
 * 
 *  Example: get_fund_primary_benchmark("'MF00003PW2', 'MF00003PW1', 'MF00003PXO'", '1990-01', '2024-06');
 */
def get_fund_primary_benchmark(fund_ids, month_start, month_end) {

    s_query = "SELECT fund_id, primary_benchmark_id AS benchmark_id, inception_date
               FROM mfdb.fund_information
               WHERE fund_id IN (" + fund_ids + ")
                 AND isvalid = 1;";

    conn = connect_mysql();

    t = odbc::query(conn, s_query);

    conn.close();

    t.addColumn('end_date', MONTH);
    m_start = temporalParse(month_start, 'yyyy-MM');
    m_end = temporalParse(month_end, 'yyyy-MM');
    tb_end_date = table(m_start..m_end AS end_date);

    return (SELECT t.fund_id, d.end_date, t.benchmark_id FROM t JOIN tb_end_date d WHERE d.end_date >= t.inception_date.month());

}


/*
 *  取某时间段的组合主基准
 *  NOTE: 目前所有Java指标计算组合默认主基准是FA00000VNB，以后很可能会改
 * 
 *  Example: get_portfolio_primary_benchmark("166002,166114", '1990-01', '2024-08');
 */
def get_portfolio_primary_benchmark(portfolio_ids, month_start, month_end) {

    s_query = "SELECT id AS portfolio_id, 'FA00000VNB' AS benchmark_id, inception_date
               FROM pfdb.pf_customer_portfolio_map
               WHERE id IN (" + portfolio_ids + ")
                 AND isvalid = 1;";

    conn = connect_mysql();

    t = odbc::query(conn, s_query);

    conn.close();

    t.addColumn('end_date', MONTH);
    m_start = temporalParse(month_start, 'yyyy-MM');
    m_end = temporalParse(month_end, 'yyyy-MM');
    tb_end_date = table(m_start..m_end AS end_date);

    return (SELECT t.portfolio_id, d.end_date, t.benchmark_id FROM t JOIN tb_end_date d WHERE d.end_date >= t.inception_date.month());

}

/*
 *  取某时间段的基金组合主基准
 *
 *  NOTE: 指数和因子的”主基准”设置为沪深300
 * 
 *  Example: get_entity_primary_benchmark('MF', "'MF00003PW2', 'MF00003PW1', 'MF00003PXO'", '1990-01', '2024-06');
 *           get_entity_primary_benchmark('PF', [166002,166114], '1990-01', '2024-08');
 *           get_entity_primary_benchmark('MI', ['IN00000008', 'IN0000000M'], '2024-07', '2024-08');
 */
def get_entity_primary_benchmark(entity_type, entity_ids, month_start, month_end) {

    t = table(100:0,
             ['entity_id', 'end_date', 'benchmark_id'],
             [iif(entity_type == 'PF', INT, SYMBOL), MONTH, SYMBOL]);

    s_entity_ids = ids_to_string(entity_ids);

    if(s_entity_ids == null || s_entity_ids == '') return null;

    if(entity_type == 'MF' || entity_type == 'HF') {

    	t = get_fund_primary_benchmark(s_entity_ids, month_start, month_end);

        t.rename!('fund_id', 'entity_id');

    } else if(entity_type == 'PF') {

    	t = get_portfolio_primary_benchmark(s_entity_ids, month_start, month_end);

        t.rename!('portfolio_id', 'entity_id');

    } else if(entity_type IN ['MI', 'FI', 'FA', 'CI', 'EQ']) {
        // 对于指数、因子来说，没有什么基准。但为了指标计算不得不在这里设个假的
    	t = SELECT entity_id, end_date, 'IN00000008' AS benchmark_id
    	    FROM cj(get_entity_info(entity_type, s_entity_ids), table(temporalParse(month_start, 'yyyy-MM')..temporalParse(month_end, 'yyyy-MM') AS end_date))
    	    WHERE end_date >= iif(inception_date.isNull(), 1990.01M, inception_date.month())
    }

	return t;
	
}

/*
 *  取某时间段的基金BFI因子
 * 
 *  Example: get_fund_bfi_factors("'MF00003PW2', 'MF00003PW1', 'MF00003PXO'", '1990-01', '2024-06');
 *           get_fund_bfi_factors(['MF00003PW2', 'MF00003PW1', 'MF00003PXO'], '1990-01', '2024-06');
 */
def get_fund_bfi_factors(fund_ids, month_start, month_end) {

    s_entity_ids = ids_to_string(fund_ids);

    if(s_entity_ids == null || s_entity_ids == '') return null;
    
    s_query = "SELECT fund_id, end_date, factor_id
               FROM pfdb.pf_fund_factor_bfi_by_category_group
               WHERE fund_id IN (" + s_entity_ids + ")
                 AND end_date >= '" + month_start + "'
                 AND end_date <= '" + month_end + "'
                 AND isvalid = 1
               ORDER BY fund_id, end_date, factor_id;";

    conn = connect_mysql();

    t = odbc::query(conn, s_query);

    conn.close();

    return t;

}


/*
 *  取某时间段的组合BFI因子
 * 
 *  Example: get_portfolio_bfi_factors("166002,166114", '1900-01', '2024-06');
 */
def get_portfolio_bfi_factors(portfolio_ids, month_start, month_end) {

    s_query = "SELECT portfolio_id, end_date, factor_id
               FROM pfdb.pf_portfolio_factor_bfi_by_category_group
               WHERE portfolio_id IN (" + portfolio_ids + ")
                 AND end_date >= '" + month_start + "'
                 AND end_date <= '" + month_end + "'
                 AND isvalid = 1
               ORDER BY portfolio_id, end_date, factor_id;";

    conn = connect_mysql();

    t = odbc::query(conn, s_query);

    conn.close();

    return t;

}


/*
 *  取某时间段的基金组合BFI基准
 *
 * 
 *  Example: get_entity_bfi_factors('MF', "'MF00003PW2', 'MF00003PW1', 'MF00003PXO'", '1990-01', '2024-06');
 *           get_entity_bfi_factors('PF', [166002,166114], '1990-01', '2024-08');
 */
def get_entity_bfi_factors(entity_type, entity_ids, month_start, month_end) {

    t = null;

    s_entity_ids = ids_to_string(entity_ids);

    if(s_entity_ids == null || s_entity_ids == '') return null;

    if(entity_type == 'MF' || entity_type == 'HF') {

    	t = get_fund_bfi_factors(s_entity_ids, month_start, month_end);

        t.rename!('fund_id', 'entity_id');

    } else if(entity_type == 'PF') {

    	t = get_portfolio_bfi_factors(s_entity_ids, month_start, month_end);

        t.rename!('portfolio_id', 'entity_id');    
    }

	return t;
	
}

/*
 * 取组合交易表
 *
 *
 * Example: get_portfolio_holding_history("166002,364640")
 */
def get_portfolio_holding_history(portfolio_ids) {

    s_query = "SELECT portfolio_id, holding_date, fund_id, amount, fund_share, ROUND(amount/fund_share, 6) AS nav
               FROM pfdb.pf_portfolio_fund_history
               WHERE portfolio_id IN (" + portfolio_ids + ")
                 AND isvalid = 1
               ORDER BY portfolio_id, holding_date";

    conn = connect_mysql();

    t = odbc::query(conn, s_query);

    conn.close();

    return t;

}

/*
 *  取基金证券从某日期后的所有净值及前值
 *  
 *  @param entity_type <STRING>: MF, HF, EQ, CF, MI, TI, CI, FA, PF
 *  @param freq <STRING>: m, w, d
 *  @param json_query <JSON>: [{sec_id:xxx, price_date: yyyy-mm-dd}]
 * 
 */
def get_nav_for_return_calculation(entity_type, freq, json_query) {

    s_query = "CALL pfdb.sp_get_nav_after_date('" + entity_type + "', '" + freq + "', '" + json_query + "')";

    conn = connect_mysql();

    t = odbc::query(conn, s_query);

    conn.close();

    return t;
}

/*
 *   取主基准和BFI的历史月收益率
 *   
 *   @param benchmarks <TABLE>: entity-benchmark 的对应关系表
 *   @param end_day <DATE>: 收益的截止日期
 * 
 *   @return <TABLE>: benchmark_id, end_date, ret
 *   
 */
def get_benchmark_return(benchmarks, end_day) {

    s_index_ids = '';
    s_factor_ids = '';

    if(benchmarks.isVoid() || benchmarks.size() == 0) { return null; }

    // 前缀为 IN 的 benchmark id
    t_index_id = SELECT DISTINCT benchmark_id FROM benchmarks WHERE benchmark_id LIKE 'IN%';
    s_index_ids = iif(isVoid(t_index_id), "", "'" + t_index_id.benchmark_id.concat("','") + "'");

   // 前缀为 FA 的 benchmark id
   t_factor_id = SELECT DISTINCT benchmark_id FROM benchmarks WHERE benchmark_id LIKE 'FA%';
   s_factor_ids = iif(isVoid(t_factor_id), "",  "'" + t_factor_id.benchmark_id.concat("','") + "'");


    // 目前指数的月度业绩存在 fund_performance 表
    t_bmk = SELECT entity_id AS benchmark_id, temporalParse(end_date, 'yyyy-MM') AS end_date, ret FROM get_monthly_ret('MI', s_index_ids, 1990.01.01, end_day, true);

    // 而因子的月度业绩存在 cm_factor_performance 表
    INSERT INTO t_bmk SELECT entity_id AS factor_id, temporalParse(end_date, 'yyyy-MM') AS end_date, ret FROM get_monthly_ret('FA', s_factor_ids, 1990.01.01, end_day, true);

	return t_bmk;
}


/*
 *  取持有基金净值更新的组合列表
 *  
 *  TODO: 需要跑3分钟，待优化
 * 
 *  Example: get_portfolio_list_by_fund_nav_updatetime([166002,166114], 2024.10.28, true);
 */
def get_portfolio_list_by_fund_nav_updatetime(portfolio_ids, updatetime, isFromMySQL) {

    t = null;

    s_entity_ids = ids_to_string(portfolio_ids);

    if(isFromMySQL == true) {

        s_query = "CALL pfdb.sp_get_portfolios_to_cal_nav(" + iif(s_entity_ids.isNull(), 'NULL', "'" + s_entity_ids + "'") + ",'" + updatetime + "')";

    conn = connect_mysql();

    t = odbc::query(conn, s_query);

    conn.close();

    }

    return t
}


/*
 *  取Json中指定的组合当日净值
 * 
 *  @param s_json <JSON>
 *  
 *  Example: get_portfolio_nav_by_date([{"portfolio_id": 166002,"price_date": "2024.10.25"},{"portfolio_id": 166114,"price_date": "2024.03.13"}], true);
 */
def get_portfolio_nav_by_date(s_json, isFromMySQL) {

    t = null;

    if(isFromMySQL == true) {

        s_query = "SELECT t.portfolio_id, t.price_date, nav.cumulative_nav
                   FROM JSON_TABLE ( '" + s_json + "', '$[*]'
                           COLUMNS ( portfolio_id INT PATH '$.portfolio_id',
                                     price_date DATE PATH '$.price_date' ) ) t
                   LEFT JOIN pfdb.pf_portfolio_nav nav ON t.portfolio_id = nav.portfolio_id AND t.price_date = nav.price_date;";

        conn = connect_mysql();
     
        t = odbc::query(conn, s_query);
     
        conn.close();
    }

    return t;
}

/*
 *   取月度指标表
 *   
 *   @param table_name <STRING>: 指标表名
 *   @param end_date <MONTH>
 *   @param isFromMySQL <BOOL>
 * 
 *   Example: get_monthly_indicator_data('mfdb.fund_performance', 2024.09M, true);
 */
def get_monthly_indicator_data(table_name, end_date, isFromMySQL=true) {

    t = null;

    s_end_date_sql = iif(end_date.isNull(), '', " AND end_date = '" + end_date.temporalFormat('yyyy-MM') + "'" );

    if(isFromMySQL == true) {

        s_query = "SELECT * FROM " + table_name + " WHERE isvalid = 1 " + s_end_date_sql;

        conn = connect_mysql();
     
        t = odbc::query(conn, s_query);
     
        conn.close();
    }

    return t;
}

/*
 *  取 pf_fund_indicator_ranking 表
 *  
 *  
 *  Example: get_fund_indicator_ranking("'MF00003PW1'", 2024.09M, 102, true);
 *           get_fund_indicator_ranking(NULL, 2023.09M, [1, 3], true);
 */
def get_fund_indicator_ranking(fund_ids, end_date, strategy, isFromMySQL=true) {

    t = null;

    s_entity_ids = ids_to_string(fund_ids);
    sql_entity_id = '';
    if(s_entity_ids != NULL) sql_entity_id = " AND fund_id IN (" + s_entity_ids + ")";

    s_strategy_ids = ids_to_string(strategy);
    sql_strategy_id = '';
    if(s_strategy_ids != NULL) sql_strategy_id = " AND strategy IN (" + s_strategy_ids + ")";


    if(isFromMySQL == true) {

        s_query = "SELECT * 
                   FROM pfdb.pf_fund_indicator_ranking 
                   WHERE isvalid = 1 
                     AND end_date = '" + end_date.temporalFormat('yyyy-MM') + "'" +
                     sql_strategy_id +
                     sql_entity_id;

        conn = connect_mysql();
     
        t = odbc::query(conn, s_query);
     
        conn.close();
    }

    return t;

}


/*
 *  取 pf_fund_indicator_substrategy_ranking 表
 *  
 *  
 *  Example: get_fund_indicator_substrategy_ranking("'MF00003PW1'", 2024.09M, 23, true);
 *           get_fund_indicator_substrategy_ranking(NULL, 2023.09M, [22, 23], true);
 */
def get_fund_indicator_substrategy_ranking(fund_ids, end_date, substrategy, isFromMySQL=true) {

    t = null;

    s_entity_ids = ids_to_string(fund_ids);
    sql_entity_id = '';
    if(s_entity_ids != NULL) sql_entity_id = " AND fund_id IN (" + s_entity_ids + ")";

    s_strategy_ids = ids_to_string(substrategy);
    sql_strategy_id = '';
    if(s_strategy_ids != NULL) sql_strategy_id = " AND substrategy IN (" + s_strategy_ids + ")";


    if(isFromMySQL == true) {

        s_query = "SELECT * 
                   FROM pfdb.pf_fund_indicator_substrategy_ranking 
                   WHERE isvalid = 1 
                     AND end_date = '" + end_date.temporalFormat('yyyy-MM') + "'" +
                     sql_strategy_id +
                     sql_entity_id;

        conn = connect_mysql();
     
        t = odbc::query(conn, s_query);
     
        conn.close();
    }

    return t;

}


/*
 *  取 pf_fund_bfi_bm_indicator_ranking 表
 *  
 *  
 *  Example: get_fund_bfi_bm_indicator_ranking("'MF00003PW1'", 2024.09M, NULL, true);
 *  
 */
def get_fund_bfi_bm_indicator_ranking(fund_ids, end_date, factor_ids, isFromMySQL=true) {

    t = null;

    s_entity_ids = ids_to_string(fund_ids);
    sql_entity_id = '';
    if(s_entity_ids != NULL) sql_entity_id = " AND fund_id IN (" + s_entity_ids + ")";

    s_factor_ids = ids_to_string(factor_ids);
    sql_factor_id = '';
    if(s_factor_ids != NULL) sql_factor_id = " AND factor_id IN (" + s_factor_ids + ")";


    if(isFromMySQL == true) {

        s_query = "SELECT * 
                   FROM pfdb.pf_fund_bfi_bm_indicator_ranking 
                   WHERE isvalid = 1 
                     AND end_date = '" + end_date.temporalFormat('yyyy-MM') + "'" +
                     sql_factor_id +
                     sql_entity_id;

        conn = connect_mysql();
     
        t = odbc::query(conn, s_query);
     
        conn.close();
    }

    return t;

}


/*
 *  【Morningstar Integration】取某时间后净值更新的公募基金列表
 *
 *  @param entity_ids <STRING|VECTOR>:
 *  @param update_time <DATETIME>: all updates after this time
 *  
 *  TODO: 将 public_nav2 换成 mfdb.public_nav 后，要把 createtime 改成 updatetime
 *
 *  Example： ms_get_fund_list_by_nav_updatetime(['MF00003PW1','MF00003PWC'], 2024.10.26);
 */
def ms_get_fund_list_by_nav_updatetime(entity_ids, updatetime) {

    s_entity_ids = ids_to_string(entity_ids);

    sql_entity_id = '';

    if(s_entity_ids != NULL) {
    	sql_entity_id = " AND fund_id IN (" + s_entity_ids + ")";
    }

    s_query = "SELECT fund_id AS entity_id, MIN(price_date) AS price_date 
               FROM raw_db.public_nav2
               WHERE isvalid = 1 " +
                 sql_entity_id + "
                 AND cumulative_nav > 0
                 AND createtime >= '" + updatetime$STRING + "'
               GROUP BY fund_id
               ORDER BY fund_id, price_date";
    
    conn = connect_mysql();

    t = odbc::query(conn, s_query);
    
    conn.close();

    return t

}


/*
 * 【Morningstar Integration】 
 *
 * Example: ms_get_fund_info("'MF00003PW1','MF00003PWC'");
 * 
 */
def ms_get_fund_info(fund_ids) {

    s_entity_ids = ids_to_string(fund_ids);
    
    if(s_entity_ids == NULL || s_entity_ids == '') return null;

    s_query = "SELECT fi.fund_id, fi.inception_date, fi.primary_benchmark_id AS benchmark_id, IFNULL(fi.initial_unit_value, 1) AS ini_value, fs.strategy, fs.substrategy
               FROM raw_db.fund_information2 fi
               INNER JOIN raw_db.fund_strategy2 fs ON fi.fund_id = fs.fund_id AND fs.isvalid = 1
               WHERE fi.fund_id IN (" + s_entity_ids + ")
               AND fi.isvalid = 1
               ORDER BY fi.fund_id"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t

}


/*
 * 【Morningstar Integration】 
 *
 * Example: ms_get_fund_monthly_nav(['MF00003PW1','MF00003PWC']);
 * 
 */
def ms_get_fund_monthly_nav(fund_ids) {

    s_entity_ids = ids_to_string(fund_ids);
    
    if(s_entity_ids == NULL || s_entity_ids == '') return null;

    s_query = "SELECT n.fund_id AS entity_id, n.price_date, n.cumulative_nav
               FROM raw_db.public_nav2 n INNER JOIN (
                   SELECT fund_id, max(price_date) AS monthend_date
                   FROM raw_db.public_nav2
                   WHERE fund_id IN (" + s_entity_ids + ")
                   AND isvalid = 1
                   AND cumulative_nav > 0
                   GROUP BY fund_id, DATE_FORMAT(price_date, '%Y-%m')
                 ) t ON n.fund_id = t.fund_id AND n.price_date = t.monthend_date
               UNION
               SELECT fi.fund_id, fi.inception_date, IFNULL(fi.initial_unit_value, 1)
               FROM raw_db.fund_information2 fi
               WHERE fi.fund_id IN (" + s_entity_ids + ")
               ORDER BY entity_id, price_date;"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t

}

/*
 *  【Morningstar Integration】取某时间段的基金主基准
 *  NOTE: 目前数据库里只存最新的基准，以后很可能会支持时间序列
 * 
 *  Example: ms_get_fund_primary_benchmark("'MF00003PW2', 'MF00003PW1', 'MF00003PXO'", '1990-01', '2024-06');
 */
def ms_get_fund_primary_benchmark(fund_ids, month_start, month_end) {

    s_query = "SELECT fund_id, primary_benchmark_id AS benchmark_id, inception_date
               FROM raw_db.fund_information2
               WHERE fund_id IN (" + fund_ids + ")
                 AND isvalid = 1;";

    conn = connect_mysql();

    t = odbc::query(conn, s_query);

    conn.close();

    t.addColumn('end_date', MONTH);
    m_start = temporalParse(month_start, 'yyyy-MM');
    m_end = temporalParse(month_end, 'yyyy-MM');
    tb_end_date = table(m_start..m_end AS end_date);

    return (SELECT t.fund_id, d.end_date, t.benchmark_id FROM t JOIN tb_end_date d WHERE d.end_date >= t.inception_date.month());

}


/*
 *  【Morningstar Integration】取某时间段的基金组合主基准
 *
 * 
 *  Example: ms_get_entity_primary_benchmark('MF', "'MF00003PW2', 'MF00003PW1', 'MF00003PXO'", '1990-01', '2024-06');
 *           ms_get_entity_primary_benchmark('PF', [166002,166114], '1990-01', '2024-08');
 */
def ms_get_entity_primary_benchmark(entity_type, entity_ids, month_start, month_end) {

    t = null;

    s_entity_ids = ids_to_string(entity_ids);

    if(s_entity_ids == null || s_entity_ids == '') return null;

    if(entity_type == 'MF' || entity_type == 'HF') {

    	t = ms_get_fund_primary_benchmark(s_entity_ids, month_start, month_end);

        t.rename!('fund_id', 'entity_id');

    } else if(entity_type == 'PF') {

    	t = get_portfolio_primary_benchmark(s_entity_ids, month_start, month_end);

        t.rename!('portfolio_id', 'entity_id');    
    }

	return t;
	
}

/*
 * 【Morningstar Integration】取无风险月度利率
 *
 * ms_get_risk_free_rate(1990.01.01, today())
 */
def ms_get_risk_free_rate(start_date, end_date) {
    
    return get_monthly_ret('MI', "'IN000002EI'", start_date, end_date, true);
}

/*
 *  【Morningstar Integration】取某时间段的基金同类平均指数
 *  NOTE: 目前数据库里只存最新的基准，以后很可能会支持时间序列
 * 
 *  Example: ms_get_fund_category_average("'MF00003PW2', 'MF00003PW1', 'MF00003PXO'", '1990-01', '2024-06');
 */
def ms_get_fund_category_average(fund_ids, month_start, month_end) {

    s_query = "SELECT fi.fund_id, ip.index_id AS benchmark_id, fi.inception_date
               FROM raw_db.fund_information2 fi
               INNER JOIN mfdb.indexes_profile ip ON ip.index_code = concat('MSMWCA.2.', fi.pub_sub_fund_type)
               WHERE fi.fund_id IN (" + fund_ids + ")
                 AND fi.isvalid = 1
                 AND ip.isvalid = 1;";

    conn = connect_mysql();

    t = odbc::query(conn, s_query);

    conn.close();

    t.addColumn('end_date', MONTH);
    m_start = temporalParse(month_start, 'yyyy-MM');
    m_end = temporalParse(month_end, 'yyyy-MM');
    tb_end_date = table(m_start..m_end AS end_date);

    return (SELECT t.fund_id, d.end_date, t.benchmark_id FROM t JOIN tb_end_date d WHERE d.end_date >= t.inception_date.month());

}