DolphinDB/modules/operationDataPuller.dos

module fundit::operationDataPuller

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_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, fi.p_fund_id
               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, u.org_id
               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_factor_info(['FA00000VNB','FA00000VMJ']);
 *           get_factor_info(NULL);
 *  
 */
def get_factor_info(factor_ids) {

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

    s_query = "SELECT fi.factor_id, fi.factor_type, fi.inception_date, NULL AS benchmark_id, IFNULL(initial_value, 1) AS ini_value
               FROM pfdb.cm_factor_information fi
               WHERE fi.isvalid = 1 " +
                 s_entity_sql + "
               ORDER BY fi.factor_id"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t
}

/*
 *  取基金经理有效信息
 * 
 *  NOTE: Will return multiple records per company!
 *  
 *  Example: get_personnel_info(['PL000000AN', 'PL000001GH']);
 */
def get_personnel_info(personnel_ids) {

    s_entity_ids = ids_to_string(personnel_ids);
    
    s_entity_sql = iif(s_entity_ids == NULL || s_entity_ids == '', '', " AND pi.personnel_id IN (" + s_entity_ids + ")");

    s_query = "SELECT pi.personnel_id AS manager_id, fi.raise_type, fs.strategy, 
                      MIN(map.management_start_date) AS inception_date, 
                      CASE fs.strategy WHEN 1 THEN 'IN00000008'
                                       WHEN 2 THEN 'IN00000008'
                                       WHEN 3 THEN 'IN0000008S'
                                       WHEN 4 THEN 'IN00000008'
                                       WHEN 5 THEN 'IN00000008'
                                       WHEN 6 THEN 'IN00000077'
                                       WHEN 7 THEN 'FA00000VNB'
                                       WHEN 8 THEN 'FA00000VNB'
                                       WHEN 101 THEN 'IN00000008'
                                       WHEN 102 THEN 'FA00000VNB'
                                       WHEN 103 THEN 'IN00000077'
                                       WHEN 104 THEN 'IN0000007G'
                                       WHEN 105 THEN 'IN0000009M'
                                       ELSE 'IN00000008' END AS benchmark_id, 
                      1 AS ini_value
               FROM mfdb.personnel_information pi
               INNER JOIN mfdb.fund_manager_mapping map ON pi.personnel_id = map.fund_manager_id
               INNER JOIN mfdb.fund_information fi ON map.fund_id = fi.fund_id
               INNER JOIN mfdb.fund_strategy fs ON fi.fund_id = fs.fund_id
               WHERE pi.isvalid = 1 " +
                 s_entity_sql + "
                 AND map.isvalid = 1
                 AND fi.isvalid = 1 
                 AND fs.isvalid = 1
                 AND fs.strategy > 0
               GROUP BY pi.personnel_id, fi.raise_type, fs.strategy
               ORDER BY pi.personnel_id"

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t
}


/*
 *  取基金经理简版信息 (curve_type + 全strategy)
 * 
 *  NOTE: 仿照 MySQL 的设定，curve_type: 1-私募, 4-公募, 7-公私募综合; strategy: 0-不分策略
 *  
 *  Example: get_personnel_info_for_perf(['PL000000AN', 'PL000001GH']);
 */
def get_personnel_info_for_perf(personnel_ids) {

    s_entity_ids = ids_to_string(personnel_ids);
    
    s_entity_sql = iif(s_entity_ids == NULL || s_entity_ids == '', '', " AND pi.personnel_id IN (" + s_entity_ids + ")");

    s_query = "SELECT * FROM (
	    		   SELECT pi.personnel_id AS manager_id,
	    				  IF(fi.raise_type = 1, 1, 4) AS curve_type,
	    				  0 AS strategy, 
	                      MIN(map.management_start_date) AS inception_date, 
	                      'FA00000VNB' AS benchmark_id, 
	                      1 AS ini_value
	               FROM mfdb.personnel_information pi
	               INNER JOIN mfdb.fund_manager_mapping map ON pi.personnel_id = map.fund_manager_id
	               INNER JOIN mfdb.fund_information fi ON map.fund_id = fi.fund_id
	               WHERE pi.isvalid = 1 " +
	                 s_entity_sql + "
	                 AND map.isvalid = 1
	                 AND fi.isvalid = 1 
	               GROUP BY pi.personnel_id, fi.raise_type
	               UNION
	               SELECT pi.personnel_id AS manager_id,
	    				  7 AS curve_type,
	    				  0 AS strategy, 
	                      MIN(map.management_start_date) AS inception_date, 
	                      'FA00000VNB' AS benchmark_id, 
	                      1 AS ini_value
	               FROM mfdb.personnel_information pi
	               INNER JOIN mfdb.fund_manager_mapping map ON pi.personnel_id = map.fund_manager_id
	               INNER JOIN mfdb.fund_information fi ON map.fund_id = fi.fund_id
	               WHERE pi.isvalid = 1 " +
	                 s_entity_sql + "
	                 AND map.isvalid = 1
	                 AND fi.isvalid = 1 
	               GROUP BY pi.personnel_id) t
	           ORDER BY manager_id, curve_type, strategy;";

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t
}



/*
 *  取基金公司简版信息 (curve_type + 全strategy)
 * 
 *  NOTE: 仿照 MySQL 的设定，curve_type: 1-私募, 4-公募, 7-公私募综合; strategy: 0-不分策略
 *  
 *  Example: get_company_info_for_perf(['CO0001003W', 'CO00000017']);
 */
def get_company_info_for_perf(company_ids) {

    s_entity_ids = ids_to_string(company_ids);
    
    s_entity_sql = iif(s_entity_ids == NULL || s_entity_ids == '', '', " AND ci.company_id IN (" + s_entity_ids + ")");

    s_query = "SELECT * FROM (
	    		   SELECT ci.company_id,
	    				  IF(fi.raise_type = 1, 1, 4) AS curve_type,
	    				  0 AS strategy, 
	                      MIN(IFNULL(fi.inception_date, ci.establish_date)) AS inception_date, 
	                      'FA00000VNB' AS benchmark_id, 
	                      1 AS ini_value
	               FROM mfdb.company_information ci
	               INNER JOIN mfdb.fund_information fi ON ci.company_id = fi.advisor_id
	               WHERE ci.isvalid = 1 " +
	                 s_entity_sql + "
	                 AND fi.isvalid = 1 
	               GROUP BY ci.company_id, fi.raise_type
	               UNION
	               SELECT ci.company_id,
	    				  7 AS curve_type,
	    				  0 AS strategy, 
	                      MIN(IFNULL(fi.inception_date, ci.establish_date)) AS inception_date,
	                      'FA00000VNB' AS benchmark_id, 
	                      1 AS ini_value
	               FROM mfdb.company_information ci
	               INNER JOIN mfdb.fund_information fi ON ci.company_id = fi.advisor_id
	               WHERE ci.isvalid = 1 " +
	                 s_entity_sql + "
	                 AND fi.isvalid = 1 
	               GROUP BY ci.company_id) t
	           ORDER BY company_id, curve_type, strategy;";

    conn = connect_mysql()

    t = odbc::query(conn, s_query)

    conn.close()

    return t
}

/*
 *  取基金公司有效信息
 * 
 *  NOTE: Will return multiple records per company!
 *  
 *  Example: get_company_info(['CO0001003W', 'CO00000017']);
 * 
 */
def get_company_info(company_ids) {

    s_entity_ids = ids_to_string(company_ids);
    
    s_entity_sql = iif(s_entity_ids == NULL || s_entity_ids == '', '', " AND ci.company_id IN (" + s_entity_ids + ")");

    s_query = "SELECT ci.company_id, fi.raise_type, 
                      fs.strategy, 
                      MIN(IFNULL(fi.inception_date, ci.establish_date)) AS inception_date, 
                      CASE fs.strategy WHEN 1 THEN 'IN00000008'
                                       WHEN 2 THEN 'IN00000008'
                                       WHEN 3 THEN 'IN0000008S'
                                       WHEN 4 THEN 'IN00000008'
                                       WHEN 5 THEN 'IN00000008'
                                       WHEN 6 THEN 'IN00000077'
                                       WHEN 7 THEN 'FA00000VNB'
                                       WHEN 8 THEN 'FA00000VNB'
                                       WHEN 101 THEN 'IN00000008'
                                       WHEN 102 THEN 'FA00000VNB'
                                       WHEN 103 THEN 'IN00000077'
                                       WHEN 104 THEN 'IN0000007G'
                                       WHEN 105 THEN 'IN0000009M'
                                       ELSE 'IN00000008' END AS benchmark_id, 
                      1 AS ini_value
               FROM mfdb.company_information ci
               INNER JOIN mfdb.fund_information fi ON ci.company_id = fi.advisor_id
               INNER JOIN mfdb.fund_strategy fs ON fi.fund_id = fs.fund_id 
               WHERE ci.isvalid = 1 " +
                 s_entity_sql + "
                 AND fi.isvalid = 1
                 AND fs.isvalid = 1
                 AND fs.strategy > 0
               GROUP BY ci.company_id, fi.raise_type, fs.strategy
               ORDER BY ci.company_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');

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

    	t = get_factor_info(s_entity_ids);
    	t.rename!('factor_id', 'entity_id');
    } else if(entity_type == 'PL') {

    	t = get_personnel_info_for_perf(s_entity_ids);
    	t.rename!('manager_id', 'entity_id');
    	
    } else if(entity_type == 'CO') {

    	t = get_company_info_for_perf(s_entity_ids);
    	t.rename!('company_id', 'entity_id');    	
    }

	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');
 *           get_entity_primary_benchmark('FA', ['FA00000VNB', 'FA00000VMJ'], '2024-01', '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, NULL 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_entity_bfi_factors('MF', "'MF00003PW2', 'MF00003PW1', 'MF00003PXO'", 1990.01M, 2024.06M);
 *           get_entity_bfi_factors('PF', [166002,166114], 1990.01M, 2029.12M, 2024.10.01);
 */
def get_entity_bfi_factors(entity_type, entity_ids, month_start, month_end, updatetime=1990.01.01) {

    tmp = get_bfi_by_category_group_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 + ")";
    }

    s_query = "SELECT " + tmp.sec_id_col[0] + " AS entity_id, end_date, factor_id 
               FROM " + tmp.table_name[0] + "
               WHERE isvalid = 1 " +
                 sql_entity_id + "
                 AND end_date >= '" + month_start.temporalFormat('yyyy-MM') + "'
                 AND end_date <= '" + month_end.temporalFormat('yyyy-MM') + "'
                 AND updatetime >= '" + updatetime$STRING + "'
               ORDER BY " + tmp.sec_id_col[0] + ", end_date, factor_id";

    conn = connect_mysql();

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

    conn.close();

    return t;
}


/*
 *  取某时间段的基金经理 BFI基准
 *
 * 
 *  Example: get_mc_bfi_factors('PL', ['PL000000NS', 'PL00000ICF'], 1990.01M, 2024.06M);
 * 
 */
def get_mc_bfi_factors(entity_type, entity_ids, month_start, month_end, updatetime=1990.01.01) {

    tmp = get_bfi_by_category_group_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 + ")";
    }

    s_query = "SELECT " + tmp.sec_id_col[0] + " AS entity_id, curve_type, strategy, end_date, factor_id 
               FROM " + tmp.table_name[0] + "
               WHERE isvalid = 1 " +
                 sql_entity_id + "
                 AND end_date >= '" + month_start.temporalFormat('yyyy-MM') + "'
                 AND end_date <= '" + month_end.temporalFormat('yyyy-MM') + "'
                 AND updatetime >= '" + updatetime$STRING + "'
               ORDER BY " + tmp.sec_id_col[0] + ", end_date, factor_id";

    conn = connect_mysql();

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

    conn.close();

    return t;
}


/*
 *   取基金-经理关系表
 *   
 *   Example：get_fund_manager_mapping(['MF00003PW1', 'MF00003PW2']);
 */
def get_fund_manager_mapping(fund_ids) {

	t = null;

	s_entity_ids = ids_to_string(fund_ids);
	
    s_query = "SELECT fund_id, fund_manager_id, management_start_date, management_end_date
               FROM mfdb.fund_manager_mapping
               WHERE fund_id IN (" + s_entity_ids + ")
                 AND isvalid = 1
               ORDER BY fund_id, management_start_date, management_end_date;";

    conn = connect_mysql();

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

    conn.close();

    return t;
}