Common

`errors` ¶

定义了在backtest中常用的异常类型及异常类型基类

`AccountError (BacktestError)` ¶

账户冲突，或者已冻结

Source code in backtest/common/errors.py

class AccountError(BacktestError):
    """账户冲突，或者已冻结"""

    def __init__(self, msg: str = None):
        super().__init__(msg or "账户冲突，或者已冻结")

    def __str__(self):
        return self.message

`BacktestError (BaseException)` ¶

错误基类

Source code in backtest/common/errors.py

class BacktestError(BaseException):
    """错误基类"""

    def __init__(self, message: str, *args):
        self.message = message
        self.args = args

    def __str__(self):
        return f"{self.message}: {self.args}"

`BadParameterError (BacktestError)` ¶

参数错误

Source code in backtest/common/errors.py

class BadParameterError(BacktestError):
    """参数错误"""

    pass

`EntrustError (BacktestError)` ¶

交易过程中发生的异常

Source code in backtest/common/errors.py

class EntrustError(BacktestError):
    """交易过程中发生的异常"""

    GENERIC_ERROR = -1
    NO_CASH = -2
    REACH_BUY_LIMIT = -3
    REACH_SELL_LIMIT = -4
    NO_POSITION = -5
    PRICE_NOT_MEET = -6
    NODATA_FOR_MATCH = -7
    NODATA = -8
    TIME_REWIND = -9
    VOLUME_NOT_ENOUGH = -10

    def __init__(self, status_code: int, **kwargs):
        self.status_code = status_code
        self.message = self.__template__().format(**kwargs)

    def __template__(self):
        return {
            EntrustError.GENERIC_ERROR: "委托失败{security}, {time}",
            EntrustError.NO_CASH: "账户{account}资金不足, 需要{required}, 当前{available}",
            EntrustError.REACH_BUY_LIMIT: "不能在涨停板上买入{security}, {time}",
            EntrustError.REACH_SELL_LIMIT: "不能在跌停板上卖出{security}, {time}",
            EntrustError.NO_POSITION: "{security}在{time}期间没有持仓",
            EntrustError.PRICE_NOT_MEET: "{security}现价未达到委托价:{entrust}",
            EntrustError.NODATA_FOR_MATCH: "没有匹配到{security}在{time}的成交数据",
            EntrustError.NODATA: "获取{security}在{time}的行情数据失败，请检查日期是否为交易日，或者当天是否停牌",
            EntrustError.TIME_REWIND: "委托时间必须递增出现。当前{time}, 前一个委托时间{last_trade_time}",
            EntrustError.VOLUME_NOT_ENOUGH: "{security}委托价{price}达到，但成交量为零。",
        }.get(self.status_code)

`helper` ¶

`get_call_stack(e)` ¶

get exception callstack as a string

Source code in backtest/common/helper.py

def get_call_stack(e: Exception) -> str:
    """get exception callstack as a string"""
    return "".join(traceback.format_exception(None, e, e.__traceback__))

`jsonify(obj)` ¶

将对象obj转换成为可以通过json.dumps序列化的字典

本方法可以将str, int, float, bool, datetime.date, datetime.datetime, 或者提供了isoformat方法的其它时间类型，提供了to_dict方法的对象类型（比如自定义对象），提供了tolist或者__iter__方法的序列对象（比如numpy数组），或者提供了__dict__方法的对象，以及上述对象的复合对象，都可以被正确地转换。

转换中依照以下顺序进行：

简单类型，如str, int, float, bool
提供了to_dict的自定义类型
如果是numpy数组，优先按tolist方法进行转换
如果是提供了isoformat的时间类型，优先转换
如果对象是dict, 按dict进行转换
如果对象提供了__iter__方法，按序列进行转换
如果对象提供了__dict__方法，按dict进行转换
抛出异常

Parameters:

Name	Type	Description	Default
`obj`		object to convert	required

Returns:

Type	Description
`dict`	A dict able to be json dumps

Source code in backtest/common/helper.py

def jsonify(obj) -> dict:
    """将对象`obj`转换成为可以通过json.dumps序列化的字典

    本方法可以将str, int, float, bool, datetime.date, datetime.datetime, 或者提供了isoformat方法的其它时间类型， 提供了to_dict方法的对象类型（比如自定义对象），提供了tolist或者__iter__方法的序列对象（比如numpy数组），或者提供了__dict__方法的对象，以及上述对象的复合对象，都可以被正确地转换。

    转换中依照以下顺序进行：

    1. 简单类型，如str, int, float, bool
    2. 提供了to_dict的自定义类型
    3. 如果是numpy数组，优先按tolist方法进行转换
    4. 如果是提供了isoformat的时间类型，优先转换
    5. 如果对象是dict, 按dict进行转换
    6. 如果对象提供了__iter__方法，按序列进行转换
    7. 如果对象提供了__dict__方法，按dict进行转换
    8. 抛出异常
    Args:
        obj : object to convert

    Returns:
        A dict able to be json dumps
    """
    if obj is None or isinstance(obj, (str, int, float, bool)):
        return obj
    elif getattr(obj, "to_dict", False):
        return jsonify(obj.to_dict())
    elif getattr(obj, "tolist", False):  # for numpy array
        return jsonify(obj.tolist())
    elif getattr(obj, "isoformat", False):
        return obj.isoformat()
    elif isinstance(obj, dict):
        return {k: jsonify(v) for k, v in obj.items()}
    elif getattr(obj, "__iter__", False):  # 注意dict类型也有__iter__
        return [jsonify(x) for x in obj]
    elif getattr(obj, "__dict__", False):
        return {k: jsonify(v) for k, v in obj.__dict__.items()}
    else:
        raise ValueError(f"{obj} is not jsonable")

`protected(wrapped)` ¶

check token and duplicated request

Source code in backtest/common/helper.py

def protected(wrapped):
    """check token and duplicated request"""

    def decorator(f):
        @wraps(f)
        async def decorated_function(request, *args, **kwargs):
            is_authenticated = check_token(request)
            is_duplicated = check_duplicated_request(request)
            params = request.json or request.args
            command = request.server_path.split("/")[-1]

            if is_authenticated and not is_duplicated:
                try:
                    logger.info("received request: %s, params %s", command, params)
                    result = await f(request, *args, **kwargs)
                    logger.info("finished request: %s, params %s", command, params)
                    return result
                except EntrustError as e:
                    logger.exception(e)
                    logger.warning("request: %s failed: %s", command, params)
                    return response.text(
                        f"{e.status_code} {e.message}\n{get_call_stack(e)}", status=499
                    )
                except Exception as e:
                    logger.exception(e)
                    logger.warning("%s error: %s", f.__name__, params)
                    return response.text(get_call_stack(e), status=499)
            elif not is_authenticated:
                logger.warning("token is invalid: [%s]", request.token)
                return response.json({"msg": "token is invalid"}, 401)
            elif is_duplicated:
                return response.json({"msg": "duplicated request"}, 200)

        return decorated_function

    return decorator(wrapped)

`protected_admin(wrapped)` ¶

check token and duplicated request

Source code in backtest/common/helper.py

def protected_admin(wrapped):
    """check token and duplicated request"""

    def decorator(f):
        @wraps(f)
        async def decorated_function(request, *args, **kwargs):
            is_authenticated = check_admin_token(request)
            is_duplicated = check_duplicated_request(request)

            if is_authenticated and not is_duplicated:
                try:
                    result = await f(request, *args, **kwargs)
                    return result
                except Exception as e:
                    logger.exception(e)
                    return response.text(str(e), status=500)
            elif not is_authenticated:
                logger.warning("admin token is invalid: [%s]", request.token)
                return response.text(f"token({request.token}) is invalid", 401)
            elif is_duplicated:
                return response.text(
                    f"duplicated request: {request.ctx.request_id}", 200
                )

        return decorated_function

    return decorator(wrapped)

`tabulate_numpy_array(arr)` ¶

将numpy structured array 格式化为表格对齐的字符串

Parameters:

Name	Type	Description	Default
`arr`		description	required

Returns:

Type	Description
`str`	description

Source code in backtest/common/helper.py

def tabulate_numpy_array(arr: np.ndarray) -> str:
    """将numpy structured array 格式化为表格对齐的字符串

    Args:
        arr : _description_

    Returns:
        _description_
    """
    table = tabulate(arr, headers=arr.dtype.names, tablefmt="fancy_grid")
    return table

Common

errors ¶

AccountError (BacktestError) ¶

BacktestError (BaseException) ¶

BadParameterError (BacktestError) ¶

EntrustError (BacktestError) ¶

helper ¶

get_call_stack(e) ¶

jsonify(obj) ¶

protected(wrapped) ¶

protected_admin(wrapped) ¶

tabulate_numpy_array(arr) ¶

`errors` ¶

`AccountError (BacktestError)` ¶

`BacktestError (BaseException)` ¶

`BadParameterError (BacktestError)` ¶

`EntrustError (BacktestError)` ¶

`helper` ¶

`get_call_stack(e)` ¶

`jsonify(obj)` ¶

`protected(wrapped)` ¶

`protected_admin(wrapped)` ¶

`tabulate_numpy_array(arr)` ¶