Repository: microsoft/qlib
File: qlib/rl/contrib/backtest.py (Python)

Code before:

def _convert_indicator_to_dataframe(indicator: dict) -> Optional[pd.DataFrame]:
    record_list = []
    for time, value_dict in indicator.items():
        if isinstance(value_dict, BaseOrderIndicator):
            # HACK: for qlib v0.8
            value_dict = value_dict.to_series()
        try:
            value_dict = {k: v for k, v in value_dict.items()}
            if value_dict["ffr"].empty:
                continue
        except Exception:
            value_dict = {k: v for k, v in value_dict.items() if k != "pa"}
        value_dict = pd.DataFrame(value_dict)
        value_dict["datetime"] = time
        record_list.append(value_dict)

    if not record_list:
        return None

    records: pd.DataFrame = pd.concat(record_list, 0).reset_index().rename(columns={"index": "instrument"})
    records = records.set_index(["instrument", "datetime"])
    return records


def _generate_report(decisions: list, report_dicts: List[dict]) -> dict:
    indicator_dict = defaultdict(list)
    indicator_his = defaultdict(list)
    for report_dict in report_dicts:
        for key, value in report_dict["indicator"].items():
            if key.endswith("_obj"):
                indicator_his[key].append(value.order_indicator_his)
            else:
                indicator_dict[key].append(value)

    report = {}
    decision_details = pd.concat([d.details for d in decisions if hasattr(d, "details")])
    for key in ["1min", "5min", "30min", "1day"]:
        if key not in indicator_dict:
            continue

        report[key] = pd.concat(indicator_dict[key])
        report[key + "_obj"] = pd.concat([_convert_indicator_to_dataframe(his) for his in indicator_his[key + "_obj"]])

        cur_details = decision_details[decision_details.freq == key].set_index(["instrument", "datetime"])
        if len(cur_details) > 0:
            cur_details.pop("freq")
            report[key + "_obj"] = report[key + "_obj"].join(cur_details, how="outer")

    return report


Review comment (suggestion):
I think there should be richer annotation for the input (e.g. report)   and the returned  report 

For example,  @ dataclass  with typed fields and detailed docstrings

Code after:

def _convert_indicator_to_dataframe(indicator: dict) -> Optional[pd.DataFrame]:
    record_list = []
    for time, value_dict in indicator.items():
        if isinstance(value_dict, BaseOrderIndicator):
            # HACK: for qlib v0.8
            value_dict = value_dict.to_series()
        try:
            value_dict = copy.deepcopy(value_dict)
            if value_dict["ffr"].empty:
                continue
        except Exception:
            value_dict = {k: v for k, v in value_dict.items() if k != "pa"}
        value_dict = pd.DataFrame(value_dict)
        value_dict["datetime"] = time
        record_list.append(value_dict)

    if not record_list:
        return None

    records: pd.DataFrame = pd.concat(record_list, 0).reset_index().rename(columns={"index": "instrument"})
    records = records.set_index(["instrument", "datetime"])
    return records


# TODO: there should be richer annotation for the input (e.g. report) and the returned report
# TODO: For example, @ dataclass with typed fields and detailed docstrings.
def _generate_report(decisions: List[BaseTradeDecision], report_indicators: List[dict]) -> dict:
    """Generate backtest reports

    Parameters
    ----------
    decisions:
        List of trade decisions.
    report_indicators
        List of indicator reports.
    Returns
    -------

    """
    indicator_dict = defaultdict(list)
    indicator_his = defaultdict(list)
    for report_indicator in report_indicators:
        for key, value in report_indicator.items():
            if key.endswith("_obj"):
                indicator_his[key].append(value.order_indicator_his)
            else:
                indicator_dict[key].append(value)

    report = {}
    decision_details = pd.concat([getattr(d, "details") for d in decisions if hasattr(d, "details")])