DolphinDB/modules/performanceDataPuller.dos

module fundit::performanceDataPuller

use fundit::sqlUtilities


/*
 *   通用取有更新的数据
 *   
 *   NOTE: isvalid = 0 的记录也会被返回
 * 
 *   Example: get_data_by_updatetime('mfdb.fund_performance_weekly', 2024.11.14T11:10:00, true);
 * 
 */
def get_data_by_updatetime(table_name, updatetime, isFromMySQL=true) {


    if(isFromMySQL == true) {

        s_query = "SELECT * FROM " + table_name + "
                   WHERE updatetime >= '" + updatetime + "'";
     
        conn = connect_mysql()
     
        t = odbc::query(conn, s_query)
     
        conn.close()

    } else {
        
    
    }

    return t
}

/*
 *  取有更新的最早周收益数据日期
 * 
 *  Example: 
 */
def get_entity_list_by_weekly_return_updatetime(entity_type, entity_ids, updatetime, isFromMySQL=true) {

    tmp = get_performance_weekly_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 {
        
    
    }

    return t
}


/*
 *   通用取周净值
 * 
 *   Example: get_weekly_ret('MF', ['MF00003PW1'], 2024.06.01, today(), true);
 *            get_weekly_ret('PF', [166002], 2024.06.01, today(), true);
 */
def get_weekly_ret(entity_type, entity_ids, start_date=1990.01.01, end_date=2099.12.31, isFromMySQL=true){

    s_entity_ids = ids_to_string(entity_ids);

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

    tmp = get_performance_weekly_table_description(entity_type);

    if(isFromMySQL == true) {

        s_query = "SELECT " + tmp.sec_id_col[0] + " AS entity_id, year_week, price_date, ret_1w 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 price_date BETWEEN '" + start_date + "' AND '" + end_date + "'
                   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])
    
    }

    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();

    return get_weekly_ret(entity_type, entity_info.entity_id, very_old_date, end_day, true);

}

/*
 *  通用取月收益
 *  
 *  @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);
 *           get_monthly_ret('PF', [166002], 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

}


/*
 *  取某时间后更新的各基金组合最早净值日期
 *
 *  @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

}


/*
 *  取基金证券从某日期后的所有净值及前值
 *  
 *  @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;
}

/*
 *  将十位标准字符型ID分门别类
 * 
 *  @param entity_ids <VECTOR>: vector of ids
 *  @return <DICT>: entity_type - vector of ids
 *  
 *  NOTE: "IN"打头的ID无法判断是市场指数还是图译指数，所以 MI, FI 都会被返回
 *  
 *  Example: check_universe(['MF00003PW1','HF12345678','MF00003PW2','IN00000008']);
 */
def check_universe (entity_ids) {

   	v_id = array(STRING);
   	dic = dict(STRING, ANY);

    if(entity_ids.isVoid() || entity_ids.size() == 0) return dic;

   	for(r in get_entity_id_info()) {

   		v_id = entity_ids[like(entity_ids, r.prefix + '%') && strlen(entity_ids)==10];

		if(v_id.size() > 0) {
			dic[r.entity_type] = v_id;
		}
   	}

	return dic;
}

/*
 *   通用取收益数据
 *   
 *   @param entity_type <STRING>:
 *   @param entity_ids <VECTOR>
 *   @param freq <STRING>: m, w, d
 *   @param start_day <DATE>
 *   @param end_day <DATE>
 *   @param isFromMySQL <BOOL>
 * 
 *   Example: get_entity_return('MF', ['MF00003PW1', 'MF00003PW2'], 'w', 2024.02.01, 2024.09.30, true);
 *            get_entity_return(NULL, ['FA00000VML', 'FA00000VMM', 'FA00000VMN', 'FA00000VMO', 'IN0000007G'], 'w');
 *            get_entity_return('PF', [166002], 'm');
 */
def get_entity_return(entity_type, entity_ids, freq='m', start_day=1990.01.01, end_day=2099.12.31, isFromMySQL=true) {

	t_ret = table(1000:0, ['entity_id', 'effective_date', 'price_date', 'ret', 'nav'], [iif(entity_type=='PF', INT, SYMBOL), STRING, DATE, DOUBLE, DOUBLE]);

	d = dict(STRING, ANY);

	if(entity_type != null) d[entity_type] = entity_ids;
	else
		d = check_universe(entity_ids);

	for(univ in d.keys()) {

		if(freq == 'm') {
			
			t = SELECT entity_id, end_date$STRING AS effective_date, price_date, ret, nav
			    FROM get_monthly_ret(univ, d[univ], start_day, end_day, isFromMySQL);
	
			t_ret.tableInsert(t);
			
		} else if(freq == 'w') {
	
			t = SELECT entity_id, year_week$STRING AS effective_date, price_date, ret, nav
			    FROM get_weekly_ret(univ, d[univ], start_day, end_day, isFromMySQL);
	
			t_ret.tableInsert(t);
			
		} else { // 缺省是日收益
	
			s_json = table(d[univ] AS sec_id, take(start_day, d[univ].size()) AS price_date).toStdJson();
	            
			// 取基金组合在包括各自的某净值日期的前值及之后的所有净值
			tb_nav = get_nav_for_return_calculation(univ, 'd', s_json);
			if(!tb_nav.isVoid() && tb_nav.size() > 0) {
	
				tb_nav.rename!('sec_id', 'entity_id').sortBy!(['entity_id', 'price_date'], [1, 1]);
	
				t = SELECT entity_id, price_date$STRING AS effective_date, price_date, cumulative_nav.ratios()-1 AS ret, cumulative_nav AS nav
				    FROM tb_nav
				    WHERE price_date <= end_day CONTEXT BY entity_id;

				t_ret.tableInsert(t);
			}
		}
	}

	return t_ret;
	
}

/*
 *  取持有基金净值更新的组合列表
 *  
 *  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;

}