add highfreq_backtest

examples/workflow_with_highfreq_backtest.py

@@ -0,0 +1,174 @@
+#  Copyright (c) Microsoft Corporation.
+#  Licensed under the MIT License.
+
+import sys
+from pathlib import Path
+
+import qlib
+import pandas as pd
+from qlib.config import REG_CN
+from qlib.contrib.model.gbdt import LGBModel
+from qlib.contrib.data.handler import Alpha158
+from qlib.contrib.strategy.strategy import TopkDropoutStrategy
+from qlib.contrib.evaluate import (
+    backtest as normal_backtest,
+    risk_analysis,
+)
+from qlib.utils import exists_qlib_data, init_instance_by_config, flatten_dict
+from qlib.workflow import R
+from qlib.workflow.record_temp import SignalRecord, PortAnaRecord
+
+
+if __name__ == "__main__":
+
+    # use default data
+    provider_uri = "~/.qlib/qlib_data/cn_data"  # target_dir
+    if not exists_qlib_data(provider_uri):
+        print(f"Qlib data is not found in {provider_uri}")
+        sys.path.append(str(Path(__file__).resolve().parent.parent.joinpath("scripts")))
+        from get_data import GetData
+
+        GetData().qlib_data(target_dir=provider_uri, region=REG_CN)
+
+    qlib.init(provider_uri=provider_uri, region=REG_CN)
+
+    market = "csi300"
+    benchmark = "SH000300"
+
+    ###################################
+    # train model
+    ###################################
+    data_handler_config = {
+        "start_time": "2008-01-01",
+        "end_time": "2020-08-01",
+        "fit_start_time": "2008-01-01",
+        "fit_end_time": "2014-12-31",
+        "instruments": market,
+    }
+
+    task = {
+        "model": {
+            "class": "LGBModel",
+            "module_path": "qlib.contrib.model.gbdt",
+            "kwargs": {
+                "loss": "mse",
+                "colsample_bytree": 0.8879,
+                "learning_rate": 0.0421,
+                "subsample": 0.8789,
+                "lambda_l1": 205.6999,
+                "lambda_l2": 580.9768,
+                "max_depth": 8,
+                "num_leaves": 210,
+                "num_threads": 20,
+            },
+        },
+        "dataset": {
+            "class": "DatasetH",
+            "module_path": "qlib.data.dataset",
+            "kwargs": {
+                "handler": {
+                    "class": "Alpha158",
+                    "module_path": "qlib.contrib.data.handler",
+                    "kwargs": data_handler_config,
+                },
+                "segments": {
+                    "train": ("2008-01-01", "2014-12-31"),
+                    "valid": ("2015-01-01", "2016-12-31"),
+                    "test": ("2017-01-01", "2020-08-01"),
+                },
+            },
+        },
+    }
+
+    highfreq_executor_config = {
+        "log_dir": '/shared_data/data/v-xiabi/highfreq-exe/log/',
+        "is_multi": True,
+        "resources": {
+            "num_cpus": 48,
+            "num_gpus": 2,
+            'device': 'cpu',
+        },
+        "paths": {
+            "raw_dir": "/shared_data/data/v-xiabi/highfreq-exe/data/backtest_test_multi/",
+            "feature_conf": "/shared_data/data/v-xiabi/highfreq-exe/code/rl4execution/config/test_feature_all1620.json",
+        },
+        "env_conf": {
+            "name": "MARL_Accelerated",
+            "max_step_num": 237,
+            "limit": 10,
+            "time_interval": 30,
+            "interval_num": 8,
+            "features": "raw_30",
+            "max_agent_num": 49,
+            "log": True,
+            "obs": {
+                "name": "MultiTeacherObs",
+                "config": {}
+            },
+            "action": {
+                "name": "Multi_Static",
+                "config": {
+                    'action_num':5,
+                    'action_map': [0, 0.25, 0.5, 0.75, 1],
+                }
+            },
+            "reward": {
+                "name": "Multi_VP_Penalty_small",
+                "config": {
+                    "action_penalty": 100,
+                    "hit_penalty": 1.,
+                }
+            },
+        },
+        "policy_conf": {
+            "name": "Multi_RL_backtest",
+            "config": {
+                "buy_policy": '/shared_data/data/v-xiabi/highfreq-exe/model/OPDS_buy/policy_best',
+                'sell_policy': '/shared_data/data/v-xiabi/highfreq-exe/model/OPDS_sell/policy_best',
+            },
+        },
+    }
+    
+    port_analysis_config = {
+        "strategy": {
+            "class": "TopkDropoutStrategy",
+            "module_path": "qlib.contrib.strategy.strategy",
+            "kwargs": {
+                "topk": 50,
+                "n_drop": 5,
+            },
+        },
+        "backtest": {
+            "verbose": False,
+            "limit_threshold": 0.095,
+            "account": 100000000,
+            "benchmark": benchmark,
+            "deal_price": "close",
+            "open_cost": 0.0005,
+            "close_cost": 0.0015,
+            "min_cost": 5,
+            "highfreq_executor": {
+                "class": "Online_Executor",
+                "module_path": "/shared_data/data/v-xiabi/highfreq-exe/code/rl4execution/executor.py",
+                "kwargs": highfreq_executor_config,
+            }
+        },
+    }
+
+    # model initiaiton
+    model = init_instance_by_config(task["model"])
+    dataset = init_instance_by_config(task["dataset"])
+
+    # start exp
+    with R.start(experiment_name="workflow"):
+        R.log_params(**flatten_dict(task))
+        model.fit(dataset)
+
+        # prediction
+        recorder = R.get_recorder()
+        sr = SignalRecord(model, dataset, recorder)
+        sr.generate()
+
+        # backtest
+        par = PortAnaRecord(recorder, port_analysis_config)
+        par.generate()

qlib/contrib/backtest/backtest.py

@@ -15,7 +15,7 @@ from ...data.dataset.utils import get_level_index
 LOG = get_module_logger("backtest")
 
 
-def backtest(pred, strategy, trade_exchange, shift, verbose, account, benchmark):
+def backtest(pred, strategy, trade_exchange, shift, verbose, account, benchmark, return_order):
     """Parameters
     ----------
     pred : pandas.DataFrame
@@ -71,7 +71,7 @@ def backtest(pred, strategy, trade_exchange, shift, verbose, account, benchmark)
 
     trade_dates = np.append(predict_dates[shift:], get_date_range(predict_dates[-1], left_shift=1, right_shift=shift))
     executor = SimulatorExecutor(trade_exchange, verbose=verbose)
-
+    order_set = []
     # trading apart
     for pred_date, trade_date in zip(predict_dates, trade_dates):
         # for loop predict date and trading date
@@ -103,6 +103,8 @@ def backtest(pred, strategy, trade_exchange, shift, verbose, account, benchmark)
             )
         else:
             order_list = []
+
+        order_set.append((trade_account, order_list, trade_date))
         # 4. Get result after executing order list
         # NOTE: The following operation will modify order.amount.
         # NOTE: If it is buy and the cash is insufficient, the tradable amount will be recalculated
@@ -111,12 +113,49 @@ def backtest(pred, strategy, trade_exchange, shift, verbose, account, benchmark)
         # 5. Update account information according to transaction
         update_account(trade_account, trade_info, trade_exchange, trade_date)
 
-    # generate backtest report
+    if return_order:
-    report_df = trade_account.report.generate_report_dataframe()
+        return order_set
-    report_df["bench"] = bench
+    else:
-    positions = trade_account.get_positions()
+        # generate backtest report
-    return report_df, positions
+        report_df = trade_account.report.generate_report_dataframe()
+        report_df["bench"] = bench
+        positions = trade_account.get_positions()
+        return report_df, positions
 
+def backtest_highfreq(pred, executor, trade_exchange, shift, order_set, verbose, account, benchmark):
+    if get_level_index(pred, level="datetime") == 1:
+        pred = pred.swaplevel().sort_index()
+
+    trade_account_highfreq = Account(init_cash=account)
+    _pred_dates = pred.index.get_level_values(level="datetime")
+    predict_dates = D.calendar(start_time=_pred_dates.min(), end_time=_pred_dates.max())
+    
+    if isinstance(benchmark, pd.Series):
+        bench = benchmark
+    else:
+        _codes = benchmark if isinstance(benchmark, list) else [benchmark]
+        _temp_result = D.features(
+            _codes,
+            ["$close/Ref($close,1)-1"],
+            predict_dates[0],
+            get_date_by_shift(predict_dates[-1], shift=shift),
+            disk_cache=1,
+        )
+        if len(_temp_result) == 0:
+            raise ValueError(f"The benchmark {_codes} does not exist. Please provide the right benchmark")
+        bench = _temp_result.groupby(level="datetime")[_temp_result.columns.tolist()[0]].mean()
+
+    for trade_account, order_list, trade_date in order_set:
+        if verbose:
+            LOG.info("[I {:%Y-%m-%d}]: highfreq trade begin.".format(trade_date))
+        ## TODO: kanren group need to merge code here
+        trade_info = executor.execute(trade_account, order_list, trade_date)
+        update_account(trade_account_highfreq, trade_info, trade_exchange, trade_date)
+
+    report_df = trade_account_highfreq.report.generate_report_dataframe()
+    report_df["bench"] = bench
+    positions = trade_account_highfreq.get_positions()
+    return report_df, positions
 
 def update_account(trade_account, trade_info, trade_exchange, trade_date):
     """Update the account and strategy

qlib/contrib/evaluate.py

@@ -11,7 +11,7 @@ from ..log import get_module_logger
 from . import strategy as strategy_pool
 from .strategy.strategy import BaseStrategy
 from .backtest.exchange import Exchange
-from .backtest.backtest import backtest as backtest_func, get_date_range
+from .backtest.backtest import backtest as backtest_func, get_date_range, backtest_highfreq as backtest_highfreq_func
 
 from ..data import D
 from ..config import C
@@ -272,19 +272,46 @@ def backtest(pred, account=1e9, shift=1, benchmark="SH000905", verbose=True, **k
     ex_args = {k: v for k, v in kwargs.items() if k in spec.args}
     trade_exchange = get_exchange(pred, **ex_args)
 
-    # run backtest
+    
-    report_df, positions = backtest_func(
+    if kwargs.get('highfreq_executor', False):
-        pred=pred,
+        order_set = backtest_func(
-        strategy=strategy,
+            pred=pred,
-        trade_exchange=trade_exchange,
+            strategy=strategy,
-        shift=shift,
+            trade_exchange=trade_exchange,
-        verbose=verbose,
+            shift=shift,
-        account=account,
+            verbose=verbose,
-        benchmark=benchmark,
+            account=account,
-    )
+            benchmark=benchmark,
-    # for  compatibility of the old API. return the dict positions
+            return_order=True,
-    positions = {k: p.position for k, p in positions.items()}
+        )
-    return report_df, positions
+        executor = init_instance_by_config(kwargs.get('highfreq_executor'))
+        report_df, positions = backtest_highfreq_func(
+            pred=pred,
+            executor=executor,
+            trade_exchange=trade_exchange,
+            shift=shift,
+            order_set=order_set,
+            verbose=verbose,
+            account=account,
+            benchmark=benchmark
+        )
+        positions = {k: p.position for k, p in positions.items()}
+        return report_df, positions
+    else:
+        # run backtest
+        report_df, positions = backtest_func(
+            pred=pred,
+            strategy=strategy,
+            trade_exchange=trade_exchange,
+            shift=shift,
+            verbose=verbose,
+            account=account,
+            benchmark=benchmark,
+            return_order=False,
+        )
+        # for  compatibility of the old API. return the dict positions
+        positions = {k: p.position for k, p in positions.items()}
+        return report_df, positions
 
 
 def long_short_backtest(