liuyu/qlib
1
0
Fork 0
mirror of https://github.com/microsoft/qlib.git synced 2026-07-11 14:56:55 +08:00
Code Activity

move backtest to core, fix calendar bugs, add some docstring

Browse Source
This commit is contained in:
bxdd
2021-05-27 21:14:39 +08:00
parent 2ad61f12b3
commit 4085b447aa
27 changed files with 298 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
qlib/backtest/__init__.py Normal file
View File

@@ -0,0 +1,132 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
from .account import Account
from .exchange import Exchange
from .executor import BaseExecutor
from .backtest import backtest as backtest_func
from .backtest import collect_data as data_generator
from ..strategy.base import BaseStrategy
from ..utils import init_instance_by_config
from ..log import get_module_logger
from ..config import C
logger = get_module_logger("backtest caller")
def get_exchange(
exchange=None,
freq="day",
start_time=None,
end_time=None,
codes="all",
subscribe_fields=[],
open_cost=0.0015,
close_cost=0.0025,
min_cost=5.0,
trade_unit=None,
limit_threshold=None,
deal_price=None,
):
"""get_exchange
Parameters
----------
# exchange related arguments
exchange: Exchange().
subscribe_fields: list
subscribe fields.
open_cost : float
open transaction cost.
close_cost : float
close transaction cost.
min_cost : float
min transaction cost.
trade_unit : int
100 for China A.
deal_price: str
dealing price type: 'close', 'open', 'vwap'.
limit_threshold : float
limit move 0.1 (10%) for example, long and short with same limit.
Returns
-------
:class: Exchange
an initialized Exchange object
"""
if trade_unit is None:
trade_unit = C.trade_unit
if limit_threshold is None:
limit_threshold = C.limit_threshold
if deal_price is None:
deal_price = C.deal_price
if exchange is None:
logger.info("Create new exchange")
# handle exception for deal_price
if deal_price[0] != "$":
deal_price = "$" + deal_price
exchange = Exchange(
freq=freq,
start_time=start_time,
end_time=end_time,
codes=codes,
deal_price=deal_price,
subscribe_fields=subscribe_fields,
limit_threshold=limit_threshold,
open_cost=open_cost,
close_cost=close_cost,
trade_unit=trade_unit,
min_cost=min_cost,
)
return exchange
else:
return init_instance_by_config(exchange, accept_types=Exchange)
def get_strategy_executor(
start_time, end_time, strategy, executor, benchmark="SH000300", account=1e9, exchange_kwargs={}
):
trade_account = Account(
init_cash=account,
benchmark_config={
"benchmark": benchmark,
"start_time": start_time,
"end_time": end_time,
},
)
trade_exchange = get_exchange(**exchange_kwargs)
common_infra = {
"trade_account": trade_account,
"trade_exchange": trade_exchange,
}
trade_strategy = init_instance_by_config(strategy, accept_types=BaseStrategy, common_infra=common_infra)
trade_executor = init_instance_by_config(executor, accept_types=BaseExecutor, common_infra=common_infra)
return trade_strategy, trade_executor
def backtest(start_time, end_time, strategy, executor, benchmark="SH000300", account=1e9, exchange_kwargs={}):
trade_strategy, trade_executor = get_strategy_executor(
start_time, end_time, strategy, executor, benchmark, account, exchange_kwargs
)
report_dict = backtest_func(start_time, end_time, trade_strategy, trade_executor)
return report_dict
def collect_data(start_time, end_time, strategy, executor, benchmark="SH000300", account=1e9, exchange_kwargs={}):
trade_strategy, trade_executor = get_strategy_executor(
start_time, end_time, strategy, executor, benchmark, account, exchange_kwargs
)
report_dict = yield from data_generator(start_time, end_time, trade_strategy, trade_executor)
return report_dict

180
qlib/backtest/account.py Normal file
View File

@@ -0,0 +1,180 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
import copy
import warnings
import pandas as pd
from .position import Position
from .report import Report
from .order import Order
"""
rtn & earning in the Account
rtn:
from order's view
1.change if any order is executed, sell order or buy order
2.change at the end of today, (today_clse - stock_price) * amount
earning
from value of current position
earning will be updated at the end of trade date
earning = today_value - pre_value
**is consider cost**
while earning is the difference of two position value, so it considers cost, it is the true return rate
in the specific accomplishment for rtn, it does not consider cost, in other words, rtn - cost = earning
Now rtn has been removed in the hierarchical backtest implemention.
"""
class Account:
def __init__(self, init_cash, freq: str = "day", benchmark_config: dict = {}):
self.init_vars(init_cash, freq, benchmark_config)
def init_vars(self, init_cash, freq: str, benchmark_config: dict):
# init cash
self.init_cash = init_cash
self.current = Position(cash=init_cash)
self.reset(freq=freq, benchmark_config=benchmark_config, init_report=True)
def reset_report(self, freq, benchmark_config):
self.report = Report(freq, benchmark_config)
self.positions = {}
self.rtn = 0
self.ct = 0
self.to = 0
self.val = 0
self.earning = 0
def reset(self, freq=None, benchmark_config=None, init_report=False):
"""reset freq and report of account
Parameters
----------
freq : str, optional
frequency of account & report, by default None
benchmark_config : {}, optional
benchmark config of report, by default None
init_report : bool, optional
whether to initialize the report, by default False
"""
if freq is not None:
self.freq = freq
if benchmark_config is not None:
self.benchmark_config = benchmark_config
if freq is not None or benchmark_config is not None or init_report:
self.reset_report(self.freq, self.benchmark_config)
def get_positions(self):
return self.positions
def get_cash(self):
return self.current.position["cash"]
def _update_state_from_order(self, order, trade_val, cost, trade_price):
# update turnover
self.to += trade_val
# update cost
self.ct += cost
# update return
# update self.rtn from order
trade_amount = trade_val / trade_price
if order.direction == Order.SELL: # 0 for sell
# when sell stock, get profit from price change
profit = trade_val - self.current.get_stock_price(order.stock_id) * trade_amount
self.rtn += profit # note here do not consider cost
elif order.direction == Order.BUY: # 1 for buy
# when buy stock, we get return for the rtn computing method
# profit in buy order is to make self.rtn is consistent with self.earning at the end of date
profit = self.current.get_stock_price(order.stock_id) * trade_amount - trade_val
self.rtn += profit
def update_order(self, order, trade_val, cost, trade_price):
# if stock is sold out, no stock price information in Position, then we should update account first, then update current position
# if stock is bought, there is no stock in current position, update current, then update account
# The cost will be substracted from the cash at last. So the trading logic can ignore the cost calculation
if order.direction == Order.SELL:
# sell stock
self._update_state_from_order(order, trade_val, cost, trade_price)
# update current position
# for may sell all of stock_id
self.current.update_order(order, trade_val, cost, trade_price)
else:
# buy stock
# deal order, then update state
self.current.update_order(order, trade_val, cost, trade_price)
self._update_state_from_order(order, trade_val, cost, trade_price)
def update_bar_count(self):
self.current.add_count_all(bar=self.freq)
def update_bar_report(self, trade_start_time, trade_end_time, trade_exchange):
"""
trade_start_time: pd.TimeStamp
trade_end_time: pd.TimeStamp
quote: pd.DataFrame (code, date), collumns
when the end of trade date
- update rtn
- update price for each asset
- update value for this account
- update earning (2nd view of return )
- update holding day, count of stock
- update position hitory
- update report
:return: None
"""
# update price for stock in the position and the profit from changed_price
stock_list = self.current.get_stock_list()
for code in stock_list:
# if suspend, no new price to be updated, profit is 0
if trade_exchange.check_stock_suspended(code, trade_start_time, trade_end_time):
continue
bar_close = trade_exchange.get_close(code, trade_start_time, trade_end_time)
self.current.update_stock_price(stock_id=code, price=bar_close)
# update holding day count
# update value
self.val = self.current.calculate_value()
# update earning
# account_value - last_account_value
# for the first trade date, account_value - init_cash
# self.report.is_empty() to judge is_first_trade_date
# get last_account_value, now_account_value, now_stock_value
if self.report.is_empty():
last_account_value = self.init_cash
else:
last_account_value = self.report.get_latest_account_value()
now_account_value = self.current.calculate_value()
now_stock_value = self.current.calculate_stock_value()
self.earning = now_account_value - last_account_value
# update report for today
# judge whether the the trading is begin.
# and don't add init account state into report, due to we don't have excess return in those days.
self.report.update_report_record(
trade_start_time=trade_start_time,
trade_end_time=trade_end_time,
account_value=now_account_value,
cash=self.current.position["cash"],
return_rate=(self.earning + self.ct) / last_account_value,
# here use earning to calculate return, position's view, earning consider cost, true return
# in order to make same definition with original backtest in evaluate.py
turnover_rate=self.to / last_account_value,
cost_rate=self.ct / last_account_value,
stock_value=now_stock_value,
)
# set now_account_value to position
self.current.position["now_account_value"] = now_account_value
self.current.update_weight_all()
# update positions
# note use deepcopy
self.positions[trade_start_time] = copy.deepcopy(self.current)
# finish today's updation
# reset the bar variables
self.rtn = 0
self.ct = 0
self.to = 0

30
qlib/backtest/backtest.py Normal file
View File

@@ -0,0 +1,30 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
def backtest(start_time, end_time, trade_strategy, trade_executor):
trade_executor.reset(start_time=start_time, end_time=end_time)
level_infra = trade_executor.get_level_infra()
trade_strategy.reset(level_infra=level_infra)
_execute_result = None
while not trade_executor.finished():
_trade_decision = trade_strategy.generate_trade_decision(_execute_result)
_execute_result = trade_executor.execute(_trade_decision)
return trade_executor.get_report()
def collect_data(start_time, end_time, trade_strategy, trade_executor):
trade_executor.reset(start_time=start_time, end_time=end_time)
level_infra = trade_executor.get_level_infra()
trade_strategy.reset(level_infra=level_infra)
_execute_result = None
while not trade_executor.finished():
_trade_decision = trade_strategy.generate_trade_decision(_execute_result)
_execute_result = yield from trade_executor.collect_data(_trade_decision)
return trade_executor.get_report()

459
qlib/backtest/exchange.py Normal file
View File

@@ -0,0 +1,459 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
import random
import logging
import numpy as np
import pandas as pd
from ..data.data import D
from ..data.dataset.utils import get_level_index
from ..config import C, REG_CN
from ..utils.resam import resam_ts_data
from ..log import get_module_logger
from .order import Order
class Exchange:
def __init__(
self,
freq="day",
start_time=None,
end_time=None,
codes="all",
deal_price=None,
subscribe_fields=[],
limit_threshold=None,
open_cost=0.0015,
close_cost=0.0025,
trade_unit=None,
min_cost=5,
extra_quote=None,
):
"""__init__
:param freq: frequency of data
:param start_time: closed start time for backtest
:param end_time: closed end time for backtest
:param codes: list stock_id list or a string of instruments(i.e. all, csi500, sse50)
:param deal_price: str, 'close', 'open', 'vwap'
:param subscribe_fields: list, subscribe fields
:param limit_threshold: float, 0.1 for example, default None
:param open_cost: cost rate for open, default 0.0015
:param close_cost: cost rate for close, default 0.0025
:param trade_unit: trade unit, 100 for China A market
:param min_cost: min cost, default 5
:param extra_quote: pandas, dataframe consists of
columns: like ['$vwap', '$close', '$factor', 'limit'].
The limit indicates that the etf is tradable on a specific day.
Necessary fields:
$close is for calculating the total value at end of each day.
Optional fields:
$vwap is only necessary when we use the $vwap price as the deal price
$factor is for rounding to the trading unit
limit will be set to False by default(False indicates we can buy this
target on this day).
index: MultipleIndex(instrument, pd.Datetime)
"""
self.freq = freq
self.start_time = start_time
self.end_time = end_time
if trade_unit is None:
trade_unit = C.trade_unit
if limit_threshold is None:
limit_threshold = C.limit_threshold
if deal_price is None:
deal_price = C.deal_price
self.logger = get_module_logger("online operator", level=logging.INFO)
self.trade_unit = trade_unit
# TODO: the quote, trade_dates, codes are not necessray.
# It is just for performance consideration.
if limit_threshold is None:
if C.region == REG_CN:
self.logger.warning(f"limit_threshold not set. The stocks hit the limit may be bought/sold")
elif abs(limit_threshold) > 0.1:
if C.region == REG_CN:
self.logger.warning(f"limit_threshold may not be set to a reasonable value")
if deal_price[0] != "$":
self.deal_price = "$" + deal_price
else:
self.deal_price = deal_price
if isinstance(codes, str):
codes = D.instruments(codes)
self.codes = codes
# Necessary fields
# $close is for calculating the total value at end of each day.
# $factor is for rounding to the trading unit
# $change is for calculating the limit of the stock
necessary_fields = {self.deal_price, "$close", "$change", "$factor", "$volume"}
subscribe_fields = list(necessary_fields | set(subscribe_fields))
all_fields = list(necessary_fields | set(subscribe_fields))
self.all_fields = all_fields
self.open_cost = open_cost
self.close_cost = close_cost
self.min_cost = min_cost
self.limit_threshold = limit_threshold
self.extra_quote = extra_quote
self.set_quote(codes, start_time, end_time)
def set_quote(self, codes, start_time, end_time):
if len(codes) == 0:
codes = D.instruments()
self.quote = D.features(codes, self.all_fields, start_time, end_time, freq=self.freq, disk_cache=True).dropna(
subset=["$close"]
)
self.quote.columns = self.all_fields
if self.quote[self.deal_price].isna().any():
self.logger.warning("{} field data contains nan.".format(self.deal_price))
if self.quote["$factor"].isna().any():
# The 'factor.day.bin' file not exists, and `factor` field contains `nan`
# Use adjusted price
self.trade_w_adj_price = True
self.logger.warning("factor.day.bin file not exists or factor contains `nan`. Order using adjusted_price.")
else:
# The `factor.day.bin` file exists and all data `close` and `factor` are not `nan`
# Use normal price
self.trade_w_adj_price = False
# update limit
# check limit_threshold
if self.limit_threshold is None:
self.quote["limit"] = False
else:
# set limit
self._update_limit(buy_limit=self.limit_threshold, sell_limit=self.limit_threshold)
quote_df = self.quote
if self.extra_quote is not None:
# process extra_quote
if "$close" not in self.extra_quote:
raise ValueError("$close is necessray in extra_quote")
if self.deal_price not in self.extra_quote.columns:
self.extra_quote[self.deal_price] = self.extra_quote["$close"]
self.logger.warning("No deal_price set for extra_quote. Use $close as deal_price.")
if "$factor" not in self.extra_quote.columns:
self.extra_quote["$factor"] = 1.0
self.logger.warning("No $factor set for extra_quote. Use 1.0 as $factor.")
if "limit" not in self.extra_quote.columns:
self.extra_quote["limit"] = False
self.logger.warning("No limit set for extra_quote. All stock will be tradable.")
assert set(self.extra_quote.columns) == set(quote_df.columns) - {"$change"}
quote_df = pd.concat([quote_df, self.extra_quote], sort=False, axis=0)
# update quote: pd.DataFrame to dict, for search use
if get_level_index(quote_df, level="datetime") == 1:
quote_df = quote_df.swaplevel().sort_index()
quote_dict = {}
for stock_id, stock_val in quote_df.groupby(level="instrument"):
quote_dict[stock_id] = stock_val
self.quote = quote_dict
def _update_limit(self, buy_limit, sell_limit):
self.quote["limit"] = ~self.quote["$change"].between(-sell_limit, buy_limit, inclusive=False)
def check_stock_limit(self, stock_id, start_time, end_time):
"""Parameter
stock_id
trade_date
is limtited
"""
return resam_ts_data(self.quote[stock_id]["limit"], start_time, end_time, method="all").iloc[0]
def check_stock_suspended(self, stock_id, start_time, end_time):
# is suspended
if stock_id in self.quote:
return resam_ts_data(self.quote[stock_id], start_time, end_time, method=None) is None
else:
return True
def is_stock_tradable(self, stock_id, start_time, end_time):
# check if stock can be traded
# same as check in check_order
if self.check_stock_suspended(stock_id, start_time, end_time) or self.check_stock_limit(
stock_id, start_time, end_time
):
return False
else:
return True
def check_order(self, order):
# check limit and suspended
if self.check_stock_suspended(order.stock_id, order.start_time, order.end_time) or self.check_stock_limit(
order.stock_id, order.start_time, order.end_time
):
return False
else:
return True
def deal_order(self, order, trade_account=None, position=None):
"""
Deal order when the actual transaction
:param order: Deal the order.
:param trade_account: Trade account to be updated after dealing the order.
:param position: position to be updated after dealing the order.
:return: trade_val, trade_cost, trade_price
"""
# need to check order first
# TODO: check the order unit limit in the exchange!!!!
# The order limit is related to the adj factor and the cur_amount.
# factor = self.quote[(order.stock_id, order.trade_date)]['$factor']
# cur_amount = trade_account.current.get_stock_amount(order.stock_id)
if self.check_order(order) is False:
raise AttributeError("need to check order first")
if trade_account is not None and position is not None:
raise ValueError("trade_account and position can only choose one")
trade_price = self.get_deal_price(order.stock_id, order.start_time, order.end_time)
trade_val, trade_cost = self._calc_trade_info_by_order(
order, trade_account.current if trade_account else position
)
# update account
if trade_val > 0:
# If the order can only be deal 0 trade_val. Nothing to be updated
# Otherwise, it will result some stock with 0 amount in the position
if trade_account:
trade_account.update_order(order=order, trade_val=trade_val, cost=trade_cost, trade_price=trade_price)
elif position:
position.update_order(order=order, trade_val=trade_val, cost=trade_cost, trade_price=trade_price)
return trade_val, trade_cost, trade_price
def get_quote_info(self, stock_id, start_time, end_time):
return resam_ts_data(self.quote[stock_id], start_time, end_time, method="last").iloc[0]
def get_close(self, stock_id, start_time, end_time):
return resam_ts_data(self.quote[stock_id]["$close"], start_time, end_time, method="last").iloc[0]
def get_volume(self, stock_id, start_time, end_time):
return resam_ts_data(self.quote[stock_id]["$volume"], start_time, end_time, method="sum").iloc[0]
def get_deal_price(self, stock_id, start_time, end_time):
deal_price = resam_ts_data(self.quote[stock_id][self.deal_price], start_time, end_time, method="last").iloc[0]
if np.isclose(deal_price, 0.0) or np.isnan(deal_price):
self.logger.warning(
f"(stock_id:{stock_id}, trade_time:{(start_time, end_time)}, {self.deal_price}): {deal_price}!!!"
)
self.logger.warning(f"setting deal_price to close price")
deal_price = self.get_close(stock_id, start_time, end_time)
return deal_price
def get_factor(self, stock_id, start_time, end_time):
return resam_ts_data(self.quote[stock_id]["$factor"], start_time, end_time, method="last").iloc[0]
def generate_amount_position_from_weight_position(self, weight_position, cash, start_time, end_time):
"""
The generate the target position according to the weight and the cash.
NOTE: All the cash will assigned to the tadable stock.
Parameter:
weight_position : dict {stock_id : weight}; allocate cash by weight_position
among then, weight must be in this range: 0 < weight < 1
cash : cash
trade_date : trade date
"""
# calculate the total weight of tradable value
tradable_weight = 0.0
for stock_id in weight_position:
if self.is_stock_tradable(stock_id=stock_id, start_time=start_time, end_time=end_time):
# weight_position must be greater than 0 and less than 1
if weight_position[stock_id] < 0 or weight_position[stock_id] > 1:
raise ValueError(
"weight_position is {}, "
"weight_position is not in the range of (0, 1).".format(weight_position[stock_id])
)
tradable_weight += weight_position[stock_id]
if tradable_weight - 1.0 >= 1e-5:
raise ValueError("tradable_weight is {}, can not greater than 1.".format(tradable_weight))
amount_dict = {}
for stock_id in weight_position:
if weight_position[stock_id] > 0.0 and self.is_stock_tradable(
stock_id=stock_id, start_time=start_time, end_time=end_time
):
amount_dict[stock_id] = (
cash
* weight_position[stock_id]
/ tradable_weight
// self.get_deal_price(stock_id=stock_id, start_time=start_time, end_time=end_time)
)
return amount_dict
def get_real_deal_amount(self, current_amount, target_amount, factor):
"""
Calculate the real adjust deal amount when considering the trading unit
:param current_amount:
:param target_amount:
:param factor:
:return real_deal_amount; Positive deal_amount indicates buying more stock.
"""
if current_amount == target_amount:
return 0
elif current_amount < target_amount:
deal_amount = target_amount - current_amount
deal_amount = self.round_amount_by_trade_unit(deal_amount, factor)
return deal_amount
else:
if target_amount == 0:
return -current_amount
else:
deal_amount = current_amount - target_amount
deal_amount = self.round_amount_by_trade_unit(deal_amount, factor)
return -deal_amount
def generate_order_for_target_amount_position(self, target_position, current_position, start_time, end_time):
"""Parameter:
target_position : dict { stock_id : amount }
current_postion : dict { stock_id : amount}
trade_unit : trade_unit
down sample : for amount 321 and trade_unit 100, deal_amount is 300
deal order on trade_date
"""
# split buy and sell for further use
buy_order_list = []
sell_order_list = []
# three parts: kept stock_id, dropped stock_id, new stock_id
# handle kept stock_id
# because the order of the set is not fixed, the trading order of the stock is different, so that the backtest results of the same parameter are different;
# so here we sort stock_id, and then randomly shuffle the order of stock_id
# because the same random seed is used, the final stock_id order is fixed
sorted_ids = sorted(set(list(current_position.keys()) + list(target_position.keys())))
random.seed(0)
random.shuffle(sorted_ids)
for stock_id in sorted_ids:
# Do not generate order for the nontradable stocks
if not self.is_stock_tradable(stock_id=stock_id, start_time=start_time, end_time=end_time):
continue
target_amount = target_position.get(stock_id, 0)
current_amount = current_position.get(stock_id, 0)
factor = self.get_factor(stock_id, start_time=start_time, end_time=end_time)
deal_amount = self.get_real_deal_amount(current_amount, target_amount, factor)
if deal_amount == 0:
continue
elif deal_amount > 0:
# buy stock
buy_order_list.append(
Order(
stock_id=stock_id,
amount=deal_amount,
direction=Order.BUY,
start_time=start_time,
end_time=end_time,
factor=factor,
)
)
else:
# sell stock
sell_order_list.append(
Order(
stock_id=stock_id,
amount=abs(deal_amount),
direction=Order.SELL,
start_time=start_time,
end_time=end_time,
factor=factor,
)
)
# return order_list : buy + sell
return sell_order_list + buy_order_list
def calculate_amount_position_value(self, amount_dict, start_time, end_time, only_tradable=False):
"""Parameter
position : Position()
amount_dict : {stock_id : amount}
"""
value = 0
for stock_id in amount_dict:
if (
self.check_stock_suspended(stock_id=stock_id, start_time=start_time, end_time=end_time) is False
and self.check_stock_limit(stock_id=stock_id, start_time=start_time, end_time=end_time) is False
):
value += (
self.get_deal_price(stock_id=stock_id, start_time=start_time, end_time=end_time)
* amount_dict[stock_id]
)
return value
def get_amount_of_trade_unit(self, factor):
if not self.trade_w_adj_price:
return self.trade_unit / factor
else:
return None
def round_amount_by_trade_unit(self, deal_amount, factor):
"""Parameter
deal_amount : float, adjusted amount
factor : float, adjusted factor
return : float, real amount
"""
if not self.trade_w_adj_price:
# the minimal amount is 1. Add 0.1 for solving precision problem.
return (deal_amount * factor + 0.1) // self.trade_unit * self.trade_unit / factor
return deal_amount
def _calc_trade_info_by_order(self, order, position):
"""
Calculation of trade info
:param order:
:param position: Position
:return: trade_val, trade_cost
"""
trade_price = self.get_deal_price(order.stock_id, order.start_time, order.end_time)
if order.direction == Order.SELL:
# sell
if position is not None:
if np.isclose(order.amount, position.get_stock_amount(order.stock_id)):
# when selling last stock. The amount don't need rounding
order.deal_amount = order.amount
else:
order.deal_amount = self.round_amount_by_trade_unit(order.amount, order.factor)
else:
# TODO: We don't know current position.
# We choose to sell all
order.deal_amount = order.amount
trade_val = order.deal_amount * trade_price
trade_cost = max(trade_val * self.close_cost, self.min_cost)
elif order.direction == Order.BUY:
# buy
if position is not None:
cash = position.get_cash()
trade_val = order.amount * trade_price
if cash < trade_val * (1 + self.open_cost):
# The money is not enough
order.deal_amount = self.round_amount_by_trade_unit(
cash / (1 + self.open_cost) / trade_price, order.factor
)
else:
# THe money is enough
order.deal_amount = self.round_amount_by_trade_unit(order.amount, order.factor)
else:
# Unknown amount of money. Just round the amount
order.deal_amount = self.round_amount_by_trade_unit(order.amount, order.factor)
trade_val = order.deal_amount * trade_price
trade_cost = trade_val * self.open_cost
else:
raise NotImplementedError("order type {} error".format(order.type))
return trade_val, trade_cost

354
qlib/backtest/executor.py Normal file
View File

@@ -0,0 +1,354 @@
import copy
import warnings
import pandas as pd
from typing import Union
from ..utils import init_instance_by_config
from ..utils.resam import parse_freq
from .order import Order
from .exchange import Exchange
from .utils import TradeCalendarManager
class BaseExecutor:
"""Base executor for trading"""
def __init__(
self,
time_per_step: str,
start_time: Union[str, pd.Timestamp] = None,
end_time: Union[str, pd.Timestamp] = None,
generate_report: bool = False,
verbose: bool = False,
track_data: bool = False,
common_infra: dict = {},
**kwargs,
):
"""
Parameters
----------
time_per_step : str
trade time per trading step, used for genreate the trade calendar
generate_report : bool, optional
whether to generate report, by default False
verbose : bool, optional
whether to print trading info, by default False
track_data : bool, optional
whether to generate trade_decision, will be used when making data for multi-level training
- If `self.track_data` is true, when making data for training, the input `trade_decision` of `execute` will be generated by `collect_data`
- Else, `trade_decision` will not be generated
common_infra : dict, optional:
common infrastructure for backtesting, may including:
- trade_account : Account, optional
trade account for trading
- trade_exchange : Exchange, optional
exchange that provides market info
"""
self.time_per_step = time_per_step
self.generate_report = generate_report
self.verbose = verbose
self.track_data = track_data
self.reset(start_time=start_time, end_time=end_time, track_data=track_data, common_infra=common_infra)
def reset_common_infra(self, common_infra):
"""
reset infrastructure for trading
- reset trade_account
"""
if not hasattr(self, "common_infra"):
self.common_infra = common_infra
else:
self.common_infra.update(common_infra)
if "trade_account" in common_infra:
self.trade_account = copy.copy(common_infra.get("trade_account"))
self.trade_account.reset(freq=self.time_per_step, init_report=True)
def reset(self, track_data: bool = None, common_infra: dict = None, **kwargs):
"""
- reset `start_time` and `end_time`, used in trade calendar
- reset `track_data`, used when making data for multi-level training
- reset `common_infra`, used to reset `trade_account`, `trade_exchange`, .etc
"""
if track_data is not None:
self.track_data = track_data
if "start_time" in kwargs or "end_time" in kwargs:
start_time = kwargs.get("start_time")
end_time = kwargs.get("end_time")
self.trade_calendar = TradeCalendarManager(
freq=self.time_per_step, start_time=start_time, end_time=end_time
)
if common_infra is not None:
self.reset_common_infra(common_infra)
def get_level_infra(self):
return {"trade_calendar": self.trade_calendar}
def finished(self):
return self.trade_calendar.finished()
def execute(self, trade_decision):
"""execute the trade decision and return the executed result
Parameters
----------
trade_decision : object
Returns
----------
execute_result : List[object]
the executed result for trade decison
"""
raise NotImplementedError("execute is not implemented!")
def collect_data(self, trade_decision):
if self.track_data:
yield trade_decision
return self.execute(trade_decision)
def get_trade_account(self):
raise NotImplementedError("get_trade_account is not implemented!")
def get_report(self):
raise NotImplementedError("get_report is not implemented!")
class NestedExecutor(BaseExecutor):
"""
Nested Executor with inner strategy and executor
- At each time `execute` is called, it will call the inner strategy and executor to execute the `trade_decision` in a higher frequency env.
"""
from ..strategy.base import BaseStrategy
def __init__(
self,
time_per_step: str,
inner_executor: Union[BaseExecutor, dict],
inner_strategy: Union[BaseStrategy, dict],
start_time: Union[str, pd.Timestamp] = None,
end_time: Union[str, pd.Timestamp] = None,
generate_report: bool = False,
verbose: bool = False,
track_data: bool = False,
trade_exchange: Exchange = None,
common_infra: dict = {},
**kwargs,
):
"""
Parameters
----------
inner_executor : BaseExecutor
trading env in each trading bar.
inner_strategy : BaseStrategy
trading strategy in each trading bar
trade_exchange : Exchange
exchange that provides market info, used to generate report
- If generate_report is None, trade_exchange will be ignored
- Else If `trade_exchange` is None, self.trade_exchange will be set with common_infra
"""
self.inner_executor = init_instance_by_config(
inner_executor, common_infra=common_infra, accept_types=BaseExecutor
)
self.inner_strategy = init_instance_by_config(
inner_strategy, common_infra=common_infra, accept_types=self.BaseStrategy
)
super(NestedExecutor, self).__init__(
time_per_step=time_per_step,
start_time=start_time,
end_time=end_time,
generate_report=generate_report,
verbose=verbose,
track_data=track_data,
common_infra=common_infra,
**kwargs,
)
if generate_report and trade_exchange is not None:
self.trade_exchange = trade_exchange
def reset_common_infra(self, common_infra):
"""
reset infrastructure for trading
- reset trade_exchange
- reset inner_strategyand inner_executor common infra
"""
super(NestedExecutor, self).reset_common_infra(common_infra)
if self.generate_report and "trade_exchange" in common_infra:
self.trade_exchange = common_infra.get("trade_exchange")
self.inner_executor.reset_common_infra(common_infra)
self.inner_strategy.reset_common_infra(common_infra)
def _init_sub_trading(self, trade_decision):
trade_step = self.trade_calendar.get_trade_step()
trade_start_time, trade_end_time = self.trade_calendar.get_step_time(trade_step)
self.inner_executor.reset(start_time=trade_start_time, end_time=trade_end_time)
sub_level_infra = self.inner_executor.get_level_infra()
self.inner_strategy.reset(level_infra=sub_level_infra, outer_trade_decision=trade_decision)
def _update_trade_account(self):
trade_step = self.trade_calendar.get_trade_step()
trade_start_time, trade_end_time = self.trade_calendar.get_step_time(trade_step)
self.trade_account.update_bar_count()
if self.generate_report:
self.trade_account.update_bar_report(
trade_start_time=trade_start_time,
trade_end_time=trade_end_time,
trade_exchange=self.trade_exchange,
)
def execute(self, trade_decision):
self._init_sub_trading(trade_decision)
execute_result = []
_inner_execute_result = None
while not self.inner_executor.finished():
_inner_trade_decision = self.inner_strategy.generate_trade_decision(_inner_execute_result)
_inner_execute_result = self.inner_executor.execute(trade_decision=_inner_trade_decision)
execute_result.extend(_inner_execute_result)
if hasattr(self, "trade_account"):
self._update_trade_account()
self.trade_calendar.step()
return execute_result
def collect_data(self, trade_decision):
if self.track_data:
yield trade_decision
self.trade_calendar.step()
self._init_sub_trading(trade_decision)
execute_result = []
_inner_execute_result = None
while not self.inner_executor.finished():
_inner_trade_decision = self.inner_strategy.generate_trade_decision(_inner_execute_result)
_inner_execute_result = yield from self.inner_executor.collect_data(trade_decision=_inner_trade_decision)
execute_result.extend(_inner_execute_result)
if hasattr(self, "trade_account"):
self._update_trade_account()
return execute_result
def get_report(self):
sub_env_report_dict = self.inner_executor.get_report()
if self.generate_report:
_report = self.trade_account.report.generate_report_dataframe()
_positions = self.trade_account.get_positions()
_count, _freq = parse_freq(self.time_per_step)
sub_env_report_dict.update({f"{_count}{_freq}": (_report, _positions)})
return sub_env_report_dict
class SimulatorExecutor(BaseExecutor):
"""Executor that simulate the true market"""
def __init__(
self,
time_per_step: str,
start_time: Union[str, pd.Timestamp] = None,
end_time: Union[str, pd.Timestamp] = None,
generate_report: bool = False,
verbose: bool = False,
track_data: bool = False,
trade_exchange: Exchange = None,
common_infra: dict = {},
**kwargs,
):
"""
Parameters
----------
trade_exchange : Exchange
exchange that provides market info, used to deal order and generate report
- If `trade_exchange` is None, self.trade_exchange will be set with common_infra
"""
super(SimulatorExecutor, self).__init__(
time_per_step=time_per_step,
start_time=start_time,
end_time=end_time,
generate_report=generate_report,
verbose=verbose,
track_data=track_data,
common_infra=common_infra,
**kwargs,
)
if trade_exchange is not None:
self.trade_exchange = trade_exchange
def reset_common_infra(self, common_infra):
"""
reset infrastructure for trading
- reset trade_exchange
"""
super(SimulatorExecutor, self).reset_common_infra(common_infra)
if "trade_exchange" in common_infra:
self.trade_exchange = common_infra.get("trade_exchange")
def execute(self, trade_decision):
trade_step = self.trade_calendar.get_trade_step()
trade_start_time, trade_end_time = self.trade_calendar.get_step_time(trade_step)
execute_result = []
for order in trade_decision:
if self.trade_exchange.check_order(order) is True:
# execute the order
trade_val, trade_cost, trade_price = self.trade_exchange.deal_order(
order, trade_account=self.trade_account
)
execute_result.append((order, trade_val, trade_cost, trade_price))
if self.verbose:
if order.direction == Order.SELL: # sell
print(
"[I {:%Y-%m-%d}]: sell {}, price {:.2f}, amount {}, deal_amount {}, factor {}, value {:.2f}.".format(
trade_start_time,
order.stock_id,
trade_price,
order.amount,
order.deal_amount,
order.factor,
trade_val,
)
)
else:
print(
"[I {:%Y-%m-%d}]: buy {}, price {:.2f}, amount {}, deal_amount {}, factor {}, value {:.2f}.".format(
trade_start_time,
order.stock_id,
trade_price,
order.amount,
order.deal_amount,
order.factor,
trade_val,
)
)
else:
if self.verbose:
print("[W {:%Y-%m-%d}]: {} wrong.".format(trade_start_time, order.stock_id))
# do nothing
pass
self.trade_account.update_bar_count()
if self.generate_report:
self.trade_account.update_bar_report(
trade_start_time=trade_start_time,
trade_end_time=trade_end_time,
trade_exchange=self.trade_exchange,
)
self.trade_calendar.step()
return execute_result
def get_report(self):
if self.generate_report:
_report = self.trade_account.report.generate_report_dataframe()
_positions = self.trade_account.get_positions()
_count, _freq = parse_freq(self.time_per_step)
return {f"{_count}{_freq}": (_report, _positions)}
else:
return {}

30
qlib/backtest/order.py Normal file
View File

@@ -0,0 +1,30 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
class Order:
SELL = 0
BUY = 1
def __init__(self, stock_id, amount, start_time, end_time, direction, factor):
"""Parameter
direction : Order.SELL for sell; Order.BUY for buy
stock_id : str
amount : float
trade_date : pd.Timestamp
factor : float
presents the weight factor assigned in Exchange()
"""
# check direction
if direction not in {Order.SELL, Order.BUY}:
raise NotImplementedError("direction not supported, `Order.SELL` for sell, `Order.BUY` for buy")
self.stock_id = stock_id
# amount of generated orders
self.amount = amount
# amount of successfully completed orders
self.deal_amount = 0
self.start_time = start_time
self.end_time = end_time
self.direction = direction
self.factor = factor

214
qlib/backtest/position.py Normal file
View File

@@ -0,0 +1,214 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
import copy
import pathlib
import pandas as pd
import numpy as np
from .order import Order
"""
Position module
"""
"""
current state of position
a typical example is :{
<instrument_id>: {
'count': <how many days the security has been hold>,
'amount': <the amount of the security>,
'price': <the close price of security in the last trading day>,
'weight': <the security weight of total position value>,
},
}
"""
class Position:
"""Position"""
def __init__(self, cash=0, position_dict={}, now_account_value=0):
# NOTE: The position dict must be copied!!!
# Otherwise the initial value
self.init_cash = cash
self.position = position_dict.copy()
self.position["cash"] = cash
self.position["now_account_value"] = now_account_value
def init_stock(self, stock_id, amount, price=None):
self.position[stock_id] = {}
self.position[stock_id]["amount"] = amount
self.position[stock_id]["price"] = price
self.position[stock_id]["weight"] = 0 # update the weight in the end of the trade date
def buy_stock(self, stock_id, trade_val, cost, trade_price):
trade_amount = trade_val / trade_price
if stock_id not in self.position:
self.init_stock(stock_id=stock_id, amount=trade_amount, price=trade_price)
else:
# exist, add amount
self.position[stock_id]["amount"] += trade_amount
self.position["cash"] -= trade_val + cost
def sell_stock(self, stock_id, trade_val, cost, trade_price):
trade_amount = trade_val / trade_price
if stock_id not in self.position:
raise KeyError("{} not in current position".format(stock_id))
else:
# decrease the amount of stock
self.position[stock_id]["amount"] -= trade_amount
# check if to delete
if self.position[stock_id]["amount"] < -1e-5:
raise ValueError(
"only have {} {}, require {}".format(self.position[stock_id]["amount"], stock_id, trade_amount)
)
elif abs(self.position[stock_id]["amount"]) <= 1e-5:
self.del_stock(stock_id)
self.position["cash"] += trade_val - cost
def del_stock(self, stock_id):
del self.position[stock_id]
def update_order(self, order, trade_val, cost, trade_price):
# handle order, order is a order class, defined in exchange.py
if order.direction == Order.BUY:
# BUY
self.buy_stock(order.stock_id, trade_val, cost, trade_price)
elif order.direction == Order.SELL:
# SELL
self.sell_stock(order.stock_id, trade_val, cost, trade_price)
else:
raise NotImplementedError("do not support order direction {}".format(order.direction))
def update_stock_price(self, stock_id, price):
self.position[stock_id]["price"] = price
def update_stock_count(self, stock_id, bar, count):
self.position[stock_id][f"count_{bar}"] = count
def update_stock_weight(self, stock_id, weight):
self.position[stock_id]["weight"] = weight
def update_cash(self, cash):
self.position["cash"] = cash
def calculate_stock_value(self):
stock_list = self.get_stock_list()
value = 0
for stock_id in stock_list:
value += self.position[stock_id]["amount"] * self.position[stock_id]["price"]
return value
def calculate_value(self):
value = self.calculate_stock_value()
value += self.position["cash"]
return value
def get_stock_list(self):
stock_list = list(set(self.position.keys()) - {"cash", "now_account_value"})
return stock_list
def get_stock_price(self, code):
return self.position[code]["price"]
def get_stock_amount(self, code):
return self.position[code]["amount"]
def get_stock_count(self, code, bar):
if f"count_{bar}" in self.position[code]:
return self.position[code][f"count_{bar}"]
else:
return 0
def get_stock_weight(self, code):
return self.position[code]["weight"]
def get_cash(self):
return self.position["cash"]
def get_stock_amount_dict(self):
"""generate stock amount dict {stock_id : amount of stock}"""
d = {}
stock_list = self.get_stock_list()
for stock_code in stock_list:
d[stock_code] = self.get_stock_amount(code=stock_code)
return d
def get_stock_weight_dict(self, only_stock=False):
"""get_stock_weight_dict
generate stock weight fict {stock_id : value weight of stock in the position}
it is meaningful in the beginning or the end of each trade date
:param only_stock: If only_stock=True, the weight of each stock in total stock will be returned
If only_stock=False, the weight of each stock in total assets(stock + cash) will be returned
"""
if only_stock:
position_value = self.calculate_stock_value()
else:
position_value = self.calculate_value()
d = {}
stock_list = self.get_stock_list()
for stock_code in stock_list:
d[stock_code] = self.position[stock_code]["amount"] * self.position[stock_code]["price"] / position_value
return d
def add_count_all(self, bar):
stock_list = self.get_stock_list()
for code in stock_list:
if f"count_{bar}" in self.position[code]:
self.position[code][f"count_{bar}"] += 1
else:
self.position[code][f"count_{bar}"] = 1
def update_weight_all(self):
weight_dict = self.get_stock_weight_dict()
for stock_code, weight in weight_dict.items():
self.update_stock_weight(stock_code, weight)
def save_position(self, path):
path = pathlib.Path(path)
p = copy.deepcopy(self.position)
cash = pd.Series(dtype=np.float)
cash["init_cash"] = self.init_cash
cash["cash"] = p["cash"]
cash["now_account_value"] = p["now_account_value"]
del p["cash"]
del p["now_account_value"]
positions = pd.DataFrame.from_dict(p, orient="index")
with pd.ExcelWriter(path) as writer:
positions.to_excel(writer, sheet_name="position")
cash.to_excel(writer, sheet_name="info")
def load_position(self, path):
"""load position information from a file
should have format below
sheet "position"
columns: ['stock', f'count_{bar}', 'amount', 'price', 'weight']
f'count_{bar}': <how many bars the security has been hold>,
'amount': <the amount of the security>,
'price': <the close price of security in the last trading day>,
'weight': <the security weight of total position value>,
sheet "cash"
index: ['init_cash', 'cash', 'now_account_value']
'init_cash': <inital cash when account was created>,
'cash': <current cash in account>,
'now_account_value': <current total account value, should equal to sum(price[stock]*amount[stock])>
"""
path = pathlib.Path(path)
positions = pd.read_excel(open(path, "rb"), sheet_name="position", index_col=0)
cash_record = pd.read_excel(open(path, "rb"), sheet_name="info", index_col=0)
positions = positions.to_dict(orient="index")
init_cash = cash_record.loc["init_cash"].values[0]
cash = cash_record.loc["cash"].values[0]
now_account_value = cash_record.loc["now_account_value"].values[0]
# assign values
self.position = {}
self.init_cash = init_cash
self.position = positions
self.position["cash"] = cash
self.position["now_account_value"] = now_account_value

324
qlib/backtest/profit_attribution.py Normal file
View File

@@ -0,0 +1,324 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
import numpy as np
import pandas as pd
from .position import Position
from ..data import D
from ..config import C
import datetime
from pathlib import Path
def get_benchmark_weight(
bench,
start_date=None,
end_date=None,
path=None,
):
"""get_benchmark_weight
get the stock weight distribution of the benchmark
:param bench:
:param start_date:
:param end_date:
:param path:
:return: The weight distribution of the the benchmark described by a pandas dataframe
Every row corresponds to a trading day.
Every column corresponds to a stock.
Every cell represents the strategy.
"""
if not path:
path = Path(C.get_data_path()).expanduser() / "raw" / "AIndexMembers" / "weights.csv"
# TODO: the storage of weights should be implemented in a more elegent way
# TODO: The benchmark is not consistant with the filename in instruments.
bench_weight_df = pd.read_csv(path, usecols=["code", "date", "index", "weight"])
bench_weight_df = bench_weight_df[bench_weight_df["index"] == bench]
bench_weight_df["date"] = pd.to_datetime(bench_weight_df["date"])
if start_date is not None:
bench_weight_df = bench_weight_df[bench_weight_df.date >= start_date]
if end_date is not None:
bench_weight_df = bench_weight_df[bench_weight_df.date <= end_date]
bench_stock_weight = bench_weight_df.pivot_table(index="date", columns="code", values="weight") / 100.0
return bench_stock_weight
def get_stock_weight_df(positions):
"""get_stock_weight_df
:param positions: Given a positions from backtest result.
:return: A weight distribution for the position
"""
stock_weight = []
index = []
for date in sorted(positions.keys()):
pos = positions[date]
if isinstance(pos, dict):
pos = Position(position_dict=pos)
index.append(date)
stock_weight.append(pos.get_stock_weight_dict(only_stock=True))
return pd.DataFrame(stock_weight, index=index)
def decompose_portofolio_weight(stock_weight_df, stock_group_df):
"""decompose_portofolio_weight
'''
:param stock_weight_df: a pandas dataframe to describe the portofolio by weight.
every row corresponds to a day
every column corresponds to a stock.
Here is an example below.
code SH600004 SH600006 SH600017 SH600022 SH600026 SH600037 \
date
2016-01-05 0.001543 0.001570 0.002732 0.001320 0.003000 NaN
2016-01-06 0.001538 0.001569 0.002770 0.001417 0.002945 NaN
....
:param stock_group_df: a pandas dataframe to describe the stock group.
every row corresponds to a day
every column corresponds to a stock.
the value in the cell repreponds the group id.
Here is a example by for stock_group_df for industry. The value is the industry code
instrument SH600000 SH600004 SH600005 SH600006 SH600007 SH600008 \
datetime
2016-01-05 801780.0 801170.0 801040.0 801880.0 801180.0 801160.0
2016-01-06 801780.0 801170.0 801040.0 801880.0 801180.0 801160.0
...
:return: Two dict will be returned. The group_weight and the stock_weight_in_group.
The key is the group. The value is a Series or Dataframe to describe the weight of group or weight of stock
"""
all_group = np.unique(stock_group_df.values.flatten())
all_group = all_group[~np.isnan(all_group)]
group_weight = {}
stock_weight_in_group = {}
for group_key in all_group:
group_mask = stock_group_df == group_key
group_weight[group_key] = stock_weight_df[group_mask].sum(axis=1)
stock_weight_in_group[group_key] = stock_weight_df[group_mask].divide(group_weight[group_key], axis=0)
return group_weight, stock_weight_in_group
def decompose_portofolio(stock_weight_df, stock_group_df, stock_ret_df):
"""
:param stock_weight_df: a pandas dataframe to describe the portofolio by weight.
every row corresponds to a day
every column corresponds to a stock.
Here is an example below.
code SH600004 SH600006 SH600017 SH600022 SH600026 SH600037 \
date
2016-01-05 0.001543 0.001570 0.002732 0.001320 0.003000 NaN
2016-01-06 0.001538 0.001569 0.002770 0.001417 0.002945 NaN
2016-01-07 0.001555 0.001546 0.002772 0.001393 0.002904 NaN
2016-01-08 0.001564 0.001527 0.002791 0.001506 0.002948 NaN
2016-01-11 0.001597 0.001476 0.002738 0.001493 0.003043 NaN
....
:param stock_group_df: a pandas dataframe to describe the stock group.
every row corresponds to a day
every column corresponds to a stock.
the value in the cell repreponds the group id.
Here is a example by for stock_group_df for industry. The value is the industry code
instrument SH600000 SH600004 SH600005 SH600006 SH600007 SH600008 \
datetime
2016-01-05 801780.0 801170.0 801040.0 801880.0 801180.0 801160.0
2016-01-06 801780.0 801170.0 801040.0 801880.0 801180.0 801160.0
2016-01-07 801780.0 801170.0 801040.0 801880.0 801180.0 801160.0
2016-01-08 801780.0 801170.0 801040.0 801880.0 801180.0 801160.0
2016-01-11 801780.0 801170.0 801040.0 801880.0 801180.0 801160.0
...
:param stock_ret_df: a pandas dataframe to describe the stock return.
every row corresponds to a day
every column corresponds to a stock.
the value in the cell repreponds the return of the group.
Here is a example by for stock_ret_df.
instrument SH600000 SH600004 SH600005 SH600006 SH600007 SH600008 \
datetime
2016-01-05 0.007795 0.022070 0.099099 0.024707 0.009473 0.016216
2016-01-06 -0.032597 -0.075205 -0.098361 -0.098985 -0.099707 -0.098936
2016-01-07 -0.001142 0.022544 0.100000 0.004225 0.000651 0.047226
2016-01-08 -0.025157 -0.047244 -0.038567 -0.098177 -0.099609 -0.074408
2016-01-11 0.023460 0.004959 -0.034384 0.018663 0.014461 0.010962
...
:return: It will decompose the portofolio to the group weight and group return.
"""
all_group = np.unique(stock_group_df.values.flatten())
all_group = all_group[~np.isnan(all_group)]
group_weight, stock_weight_in_group = decompose_portofolio_weight(stock_weight_df, stock_group_df)
group_ret = {}
for group_key in stock_weight_in_group:
stock_weight_in_group_start_date = min(stock_weight_in_group[group_key].index)
stock_weight_in_group_end_date = max(stock_weight_in_group[group_key].index)
temp_stock_ret_df = stock_ret_df[
(stock_ret_df.index >= stock_weight_in_group_start_date)
& (stock_ret_df.index <= stock_weight_in_group_end_date)
]
group_ret[group_key] = (temp_stock_ret_df * stock_weight_in_group[group_key]).sum(axis=1)
# If no weight is assigned, then the return of group will be np.nan
group_ret[group_key][group_weight[group_key] == 0.0] = np.nan
group_weight_df = pd.DataFrame(group_weight)
group_ret_df = pd.DataFrame(group_ret)
return group_weight_df, group_ret_df
def get_daily_bin_group(bench_values, stock_values, group_n):
"""get_daily_bin_group
Group the values of the stocks of benchmark into several bins in a day.
Put the stocks into these bins.
:param bench_values: A series contains the value of stocks in benchmark.
The index is the stock code.
:param stock_values: A series contains the value of stocks of your portofolio
The index is the stock code.
:param group_n: Bins will be produced
:return: A series with the same size and index as the stock_value.
The value in the series is the group id of the bins.
The No.1 bin contains the biggest values.
"""
stock_group = stock_values.copy()
# get the bin split points based on the daily proportion of benchmark
split_points = np.percentile(bench_values[~bench_values.isna()], np.linspace(0, 100, group_n + 1))
# Modify the biggest uppper bound and smallest lowerbound
split_points[0], split_points[-1] = -np.inf, np.inf
for i, (lb, up) in enumerate(zip(split_points, split_points[1:])):
stock_group.loc[stock_values[(stock_values >= lb) & (stock_values < up)].index] = group_n - i
return stock_group
def get_stock_group(stock_group_field_df, bench_stock_weight_df, group_method, group_n=None):
if group_method == "category":
# use the value of the benchmark as the category
return stock_group_field_df
elif group_method == "bins":
assert group_n is not None
# place the values into `group_n` fields.
# Each bin corresponds to a category.
new_stock_group_df = stock_group_field_df.copy().loc[
bench_stock_weight_df.index.min() : bench_stock_weight_df.index.max()
]
for idx, row in (~bench_stock_weight_df.isna()).iterrows():
bench_values = stock_group_field_df.loc[idx, row[row].index]
new_stock_group_df.loc[idx] = get_daily_bin_group(
bench_values, stock_group_field_df.loc[idx], group_n=group_n
)
return new_stock_group_df
def brinson_pa(
positions,
bench="SH000905",
group_field="industry",
group_method="category",
group_n=None,
deal_price="vwap",
):
"""brinson profit attribution
:param positions: The position produced by the backtest class
:param bench: The benchmark for comparing. TODO: if no benchmark is set, the equal-weighted is used.
:param group_field: The field used to set the group for assets allocation.
`industry` and `market_value` is often used.
:param group_method: 'category' or 'bins'. The method used to set the group for asstes allocation
`bin` will split the value into `group_n` bins and each bins represents a group
:param group_n: . Only used when group_method == 'bins'.
:return:
A dataframe with three columns: RAA(excess Return of Assets Allocation), RSS(excess Return of Stock Selectino), RTotal(Total excess Return)
Every row corresponds to a trading day, the value corresponds to the next return for this trading day
The middle info of brinson profit attribution
"""
# group_method will decide how to group the group_field.
dates = sorted(positions.keys())
start_date, end_date = min(dates), max(dates)
bench_stock_weight = get_benchmark_weight(bench, start_date, end_date)
# The attributes for allocation will not
if not group_field.startswith("$"):
group_field = "$" + group_field
if not deal_price.startswith("$"):
deal_price = "$" + deal_price
# FIXME: In current version. Some attributes(such as market_value) of some
# suspend stock is NAN. So we have to get more date to forward fill the NAN
shift_start_date = start_date - datetime.timedelta(days=250)
instruments = D.list_instruments(
D.instruments(market="all"),
start_time=shift_start_date,
end_time=end_date,
as_list=True,
)
stock_df = D.features(
instruments,
[group_field, deal_price],
start_time=shift_start_date,
end_time=end_date,
freq="day",
)
stock_df.columns = [group_field, "deal_price"]
stock_group_field = stock_df[group_field].unstack().T
# FIXME: some attributes of some suspend stock is NAN.
stock_group_field = stock_group_field.fillna(method="ffill")
stock_group_field = stock_group_field.loc[start_date:end_date]
stock_group = get_stock_group(stock_group_field, bench_stock_weight, group_method, group_n)
deal_price_df = stock_df["deal_price"].unstack().T
deal_price_df = deal_price_df.fillna(method="ffill")
# NOTE:
# The return will be slightly different from the of the return in the report.
# Here the position are adjusted at the end of the trading day with close
stock_ret = (deal_price_df - deal_price_df.shift(1)) / deal_price_df.shift(1)
stock_ret = stock_ret.shift(-1).loc[start_date:end_date]
port_stock_weight_df = get_stock_weight_df(positions)
# decomposing the portofolio
port_group_weight_df, port_group_ret_df = decompose_portofolio(port_stock_weight_df, stock_group, stock_ret)
bench_group_weight_df, bench_group_ret_df = decompose_portofolio(bench_stock_weight, stock_group, stock_ret)
# if the group return of the portofolio is NaN, replace it with the market
# value
mod_port_group_ret_df = port_group_ret_df.copy()
mod_port_group_ret_df[mod_port_group_ret_df.isna()] = bench_group_ret_df
Q1 = (bench_group_weight_df * bench_group_ret_df).sum(axis=1)
Q2 = (port_group_weight_df * bench_group_ret_df).sum(axis=1)
Q3 = (bench_group_weight_df * mod_port_group_ret_df).sum(axis=1)
Q4 = (port_group_weight_df * mod_port_group_ret_df).sum(axis=1)
return (
pd.DataFrame(
{
"RAA": Q2 - Q1, # The excess profit from the assets allocation
"RSS": Q3 - Q1, # The excess profit from the stocks selection
# The excess profit from the interaction of assets allocation and stocks selection
"RIN": Q4 - Q3 - Q2 + Q1,
"RTotal": Q4 - Q1, # The totoal excess profit
}
),
{
"port_group_ret": port_group_ret_df,
"port_group_weight": port_group_weight_df,
"bench_group_ret": bench_group_ret_df,
"bench_group_weight": bench_group_weight_df,
"stock_group": stock_group,
"bench_stock_weight": bench_stock_weight,
"port_stock_weight": port_stock_weight_df,
"stock_ret": stock_ret,
},
)

190
qlib/backtest/report.py Normal file
View File

@@ -0,0 +1,190 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
from collections import OrderedDict
from logging import warning
import pandas as pd
import pathlib
import warnings
from pandas.core.frame import DataFrame
from ..utils.resam import parse_freq, resam_ts_data
from ..data import D
class Report:
# daily report of the account
# contain those followings: returns, costs turnovers, accounts, cash, bench, value
# update report
def __init__(self, freq: str = "day", benchmark_config: dict = {}):
"""
Parameters
----------
freq : str
frequency of trading bar, used for updating hold count of trading bar
benchmark_config : dict
config of benchmark, may including the following arguments:
- benchmark : Union[str, list, pd.Series]
- If `benchmark` is pd.Series, `index` is trading date; the value T is the change from T-1 to T.
example:
print(D.features(D.instruments('csi500'), ['$close/Ref($close, 1)-1'])['$close/Ref($close, 1)-1'].head())
2017-01-04 0.011693
2017-01-05 0.000721
2017-01-06 -0.004322
2017-01-09 0.006874
2017-01-10 -0.003350
- If `benchmark` is list, will use the daily average change of the stock pool in the list as the 'bench'.
- If `benchmark` is str, will use the daily change as the 'bench'.
benchmark code, default is SH000300 CSI300
- start_time : Union[str, pd.Timestamp], optional
- If `benchmark` is pd.Series, it will be ignored
- Else, it represent start time of benchmark, by default None
- end_time : Union[str, pd.Timestamp], optional
- If `benchmark` is pd.Series, it will be ignored
- Else, it represent end time of benchmark, by default None
"""
self.init_vars()
self.init_bench(freq=freq, benchmark_config=benchmark_config)
def init_vars(self):
self.accounts = OrderedDict() # account postion value for each trade date
self.returns = OrderedDict() # daily return rate for each trade date
self.turnovers = OrderedDict() # turnover for each trade date
self.costs = OrderedDict() # trade cost for each trade date
self.values = OrderedDict() # value for each trade date
self.cashes = OrderedDict()
self.benches = OrderedDict()
self.latest_report_time = None # pd.TimeStamp
def init_bench(self, freq=None, benchmark_config=None):
if freq is not None:
self.freq = freq
if benchmark_config is not None:
self.benchmark_config = benchmark_config
self.bench = self._cal_benchmark(self.benchmark_config, self.freq)
def _cal_benchmark(self, benchmark_config, freq):
benchmark = benchmark_config.get("benchmark", "SH000300")
if isinstance(benchmark, pd.Series):
return benchmark
else:
start_time = benchmark_config.get("start_time", None)
end_time = benchmark_config.get("end_time", None)
if freq is None:
raise ValueError("benchmark freq can't be None!")
_codes = benchmark if isinstance(benchmark, list) else [benchmark]
fields = ["$close/Ref($close,1)-1"]
try:
_temp_result = D.features(_codes, fields, start_time, end_time, freq=freq, disk_cache=1)
except ValueError:
_, norm_freq = parse_freq(freq)
if norm_freq in ["month", "week", "day"]:
try:
_temp_result = D.features(_codes, fields, start_time, end_time, freq="day", disk_cache=1)
except ValueError:
_temp_result = D.features(_codes, fields, start_time, end_time, freq="1min", disk_cache=1)
elif norm_freq == "minute":
_temp_result = D.features(_codes, fields, start_time, end_time, freq="1min", disk_cache=1)
else:
raise ValueError(f"benchmark freq {freq} is not supported")
if len(_temp_result) == 0:
raise ValueError(f"The benchmark {_codes} does not exist. Please provide the right benchmark")
return _temp_result.groupby(level="datetime")[_temp_result.columns.tolist()[0]].mean().fillna(0)
def _sample_benchmark(self, bench, trade_start_time, trade_end_time):
def cal_change(x):
return (x + 1).prod() - 1
_ret = resam_ts_data(bench, trade_start_time, trade_end_time, method=cal_change)
return 0.0 if _ret is None else _ret
def is_empty(self):
return len(self.accounts) == 0
def get_latest_date(self):
return self.latest_report_time
def get_latest_account_value(self):
return self.accounts[self.latest_report_time]
def update_report_record(
self,
trade_start_time=None,
trade_end_time=None,
account_value=None,
cash=None,
return_rate=None,
turnover_rate=None,
cost_rate=None,
stock_value=None,
):
# check data
if None in [
trade_start_time,
trade_end_time,
account_value,
cash,
return_rate,
turnover_rate,
cost_rate,
stock_value,
]:
raise ValueError(
"None in [trade_start_time, trade_end_time, account_value, cash, return_rate, turnover_rate, cost_rate, stock_value]"
)
# update report data
self.accounts[trade_start_time] = account_value
self.returns[trade_start_time] = return_rate
self.turnovers[trade_start_time] = turnover_rate
self.costs[trade_start_time] = cost_rate
self.values[trade_start_time] = stock_value
self.cashes[trade_start_time] = cash
self.benches[trade_start_time] = self._sample_benchmark(self.bench, trade_start_time, trade_end_time)
# update latest_report_date
self.latest_report_time = trade_start_time
# finish daily report update
def generate_report_dataframe(self):
report = pd.DataFrame()
report["account"] = pd.Series(self.accounts)
report["return"] = pd.Series(self.returns)
report["turnover"] = pd.Series(self.turnovers)
report["cost"] = pd.Series(self.costs)
report["value"] = pd.Series(self.values)
report["cash"] = pd.Series(self.cashes)
report["bench"] = pd.Series(self.benches)
report.index.name = "datetime"
return report
def save_report(self, path):
r = self.generate_report_dataframe()
r.to_csv(path)
def load_report(self, path):
"""load report from a file
should have format like
columns = ['account', 'return', 'turnover', 'cost', 'value', 'cash', 'bench']
:param
path: str/ pathlib.Path()
"""
path = pathlib.Path(path)
r = pd.read_csv(open(path, "rb"), index_col=0)
r.index = pd.DatetimeIndex(r.index)
index = r.index
self.init_vars()
for trade_time in index:
self.update_report_record(
trade_time=trade_time,
account_value=r.loc[trade_time]["account"],
cash=r.loc[trade_time]["cash"],
return_rate=r.loc[trade_time]["return"],
turnover_rate=r.loc[trade_time]["turnover"],
cost_rate=r.loc[trade_time]["cost"],
stock_value=r.loc[trade_time]["value"],
bench_value=r.loc[trade_time]["bench"],
)

98
qlib/backtest/utils.py Normal file
View File

@@ -0,0 +1,98 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
import pandas as pd
from typing import Union
from ..utils.resam import get_resam_calendar
from ..data.data import Cal
class TradeCalendarManager:
"""
Manager for trading calendar
- BaseStrategy and BaseExecutor will use it
"""
def __init__(
self, freq: str, start_time: Union[str, pd.Timestamp] = None, end_time: Union[str, pd.Timestamp] = None
):
"""
Parameters
----------
freq : str
frequency of trading calendar, also trade time per trading step
start_time : Union[str, pd.Timestamp], optional
closed start of the trading calendar, by default None
If `start_time` is None, it must be reset before trading.
end_time : Union[str, pd.Timestamp], optional
closed end of the trade time range, by default None
If `end_time` is None, it must be reset before trading.
"""
self.freq = freq
self.start_time = pd.Timestamp(start_time) if start_time else None
self.end_time = pd.Timestamp(end_time) if end_time else None
self._init_trade_calendar(freq=freq, start_time=start_time, end_time=end_time)
def _init_trade_calendar(self, freq, start_time, end_time):
"""
Reset the trade calendar
- self.trade_len : The total count for trading step
- self.trade_step : The number of trading step finished, self.trade_step can be [0, 1, 2, ..., self.trade_len - 1]
"""
_calendar, freq, freq_sam = get_resam_calendar(freq=freq)
self.trade_calendar = _calendar
_, _, _start_index, _end_index = Cal.locate_index(start_time, end_time, freq=freq, freq_sam=freq_sam)
self.start_index = _start_index
self.end_index = _end_index
self.trade_len = _end_index - _start_index + 1
self.trade_step = 0
def finished(self):
"""
Check if the trading finished
- Should check before calling strategy.generate_decisions and executor.execute
- If self.trade_step >= self.self.trade_len, it means the trading is finished
- If self.trade_step < self.self.trade_len, it means the number of trading step finished is self.trade_step
"""
return self.trade_step >= self.trade_len
def step(self):
if self.finished():
raise RuntimeError(f"The calendar is finished, please reset it if you want to call it!")
self.trade_step = self.trade_step + 1
def get_freq(self):
return self.freq
def get_trade_len(self):
return self.trade_len
def get_trade_step(self):
return self.trade_step
def get_step_time(self, trade_step=0, shift=0):
"""
Get the time range of trading step
Parameters
----------
trade_step : int, optional
the number of trading step finished, by default 0
shift : int, optional
shift bars , by default 0
Returns
-------
Tuple[pd.Timestamp, pd.Timestap]
- If shift == 0, return the trading time range
- If shift > 0, return the trading time range of the earlier shift bars
- If shift < 0, return the trading time range of the later shift bar
"""
trade_step = trade_step - shift
calendar_index = self.start_index + trade_step
return self.trade_calendar[calendar_index], self.trade_calendar[calendar_index + 1] - pd.Timedelta(seconds=1)
def get_all_time(self):
"""Get the start_time and end_time for trading"""
return self.start_time, self.end_time