如何对加密货币现货网格交易策略进行Tick级回测？求适配Python库

是否存在可用于对加密货币知名现货网格交易策略进行Tick级回测的Python库？我已从Binance完成Tick数据下载，尝试使用backtesting.py进行回测，但该库似乎不适合Tick级回测。网格策略逻辑十分简洁，我认为Tick级回测应不难实现，相信已有开发者完成相关工作，只是我未找到，因此提出此问题。

该策略为Binance官方介绍的现货网格策略，我还在GitHub上找到几个开源参考：

• crypto-grid-backtest项目中的grid/backtest.py
• aibitgo项目中的strategy/GridStrategyPercent.py
• trading_bot项目中的GridStrategy.py

主运行脚本 #

from backtesting import Backtest
from grid.grid import GridStrategy
from tick_data.data import Kind, load_data

if __name__ == "__main__":
    # 1s interval pickle
    df = load_data(symbol="ETHUSDT", start="2022-12-16", end="2022-12-16", kind=Kind.SPOT, tz="UTC")

    df = df.resample("1s").first().dropna()
    print(f'{df}')

    # Backtest
    bt = Backtest(df, GridStrategy, cash=10_000, commission=.001, exclusive_orders=True)
    stats = bt.run()
    bt.plot()

GridStrategy策略类 #

import numpy as np
import pandas as pd

from enum import Enum
from backtesting import Strategy


class GridType(Enum):
    ARITHMETIC = 1
    GEOMETRIC = 2


class GridStrategy(Strategy):
    lower_limit = 2000
    upper_limit = 10000
    grid_count = 4
    grid_type = GridType.ARITHMETIC
    grids = []

    current_position = None

    def init(self):
        self.grids = self.get_grids(self.lower_limit, self.upper_limit, self.grid_count, self.grid_type)

        print(f'{self.grids}')

    def next(self):
        pass

    @staticmethod
    def get_grids(lower_limit, upper_limit, grid_count, grid_type=GridType.ARITHMETIC):
        if grid_type == GridType.ARITHMETIC:
            grids = np.linspace(lower_limit, upper_limit, grid_count + 1)
        elif grid_type == GridType.GEOMETRIC:
            grids = np.geomspace(lower_limit, upper_limit, grid_count + 1)
        else:
            print("not right range type")
        return grids

数据加载模块 #

import io
import logging
from concurrent.futures import ThreadPoolExecutor
from datetime import date
from enum import Enum
from pathlib import Path
from typing import Optional, Union
from zipfile import ZipFile
from datetime import datetime

import pandas as pd
import httpx
from pandas import DataFrame

DATA_DIR = Path.cwd().joinpath("data")


def create_logger():
    logger_ = logging.getLogger(__name__)
    logger_.setLevel(logging.INFO)
    formatter = logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )
    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    logger_.addHandler(handler)
    return logger_


logger = create_logger()


class Kind(Enum):
    SPOT = "spot"
    FUTURES_UM = "futures/um"
    FUTURES_CM = "futures/cm"


class DataLoader:
    def __init__(
            self,
            kind: Kind,
            symbol: str,
            start: Union[date, str],
            end: Union[date, str],
            tz: str = None,
    ) -> None:
        self.kind = kind
        self.symbol = symbol
        self.start = start
        self.end = end
        self.tz = tz

    def load_data(self) -> pd.DataFrame:
        with ThreadPoolExecutor(max_workers=20) as executor:
            dfs = list(
                executor.map(self.load_daily_data, pd.date_range(self.start, self.end))
            )
        df = pd.concat(dfs)
        if self.tz:
            df.index = df.index.tz_localize("utc").tz_convert(self.tz)
        return df

    def load_daily_data(self, dt: date) -> Optional[pd.DataFrame]:
        try:
            return self.load_local_daily_data(dt)
        except FileNotFoundError:
            return self.download_daily_data(dt)

    def load_local_daily_data(self, dt: date) -> pd.DataFrame:
        pickle_path = self.get_daily_pickle_path(dt)
        return pd.read_pickle(pickle_path)

    def download_daily_data(self, dt: date) -> Optional[pd.DataFrame]:
        logger.info(f'Downloading {self.symbol} {datetime.strftime(dt, "%Y-%m-%d")}')
        url = f'https://data.binance.vision/data/{self.kind.value}/daily/trades/{self.symbol}/' \
              f'{self.symbol}-trades-{datetime.strftime(dt, "%Y-%m-%d")}.zip'

        resp = httpx.get(url)
        resp.raise_for_status()

        with ZipFile(io.BytesIO(resp.content)) as zf:
            with zf.open(zf.namelist()[0]) as f:
                df = pd.read_csv(f, usecols=[1, 4], names=["price", "datetime"])

        df["datetime"] = pd.to_datetime(df.datetime, unit="ms")
        df.set_index("datetime", inplace=True)
        # df = df.resample("1s").first().dropna()

        # df.price.resample("1s").agg({
        #     "Open": "first",
        #     "High": "max",
        #     "Low": "min",
        #     "Close": "last"
        # })

        pkl_path = self.get_daily_pickle_path(dt)
        Path(pkl_path).parent.mkdir(parents=True, exist_ok=True)
        df.to_pickle(pkl_path)

        return df

    def get_daily_pickle_path(self, dt: date) -> Path:
        return DATA_DIR.joinpath(self.kind.value, f"{self.symbol}-{dt.year}-{dt.month}-{dt.day}.pkl")


def load_data(
        symbol: str,
        start: Union[date, str],
        end: Union[date, str] = date.today(),
        kind: Union[Kind, str] = Kind.SPOT,
        tz: str = "UTC",
) -> DataFrame:
    if isinstance(kind, str):
        kind = {"spot": Kind.SPOT, "cm": Kind.FUTURES_CM, "um": Kind.FUTURES_UM}[kind]
    return DataLoader(kind, symbol, start, end, tz).load_data()

内容的提问来源于stack exchange，提问作者nop