当前位置：首页 >> 加密百科 >> 使用 Python 进行加密货币回测：分步指南

使用 Python 进行加密货币回测：分步指南

admin 2025-04-26 10:34:22 加密百科 9

在本指南中，我们将利用CoinGecko API构建一个加密回测工具，该工具可以轻松测试各种交易策略，从简单的基于价格的方法（如逢低买入）到使用技术分析和指标的更复杂的方法。

与往常一样，您会在文章末尾找到 GitHub 存储库的链接，让您可以直接进入并进行实验。

什么是加密货币回测？

在交易中，加密货币回测是指使用历史市场数据评估交易策略以了解其过去的表现的过程。

它允许交易者在实际交易中使用该策略之前评估盈利能力、风险和整体有效性。通过基于过往数据的模拟交易，交易者可以改进策略，识别潜在的弱点，并在不冒实际资金风险的情况下增强对策略的信心。

加密货币回测 - 如何使用 Python 回测加密货币交易策略 - coin market cap api

先决条件

在开始构建加密回测工具之前，我们需要以下内容：

Python 3.10+
IDE
CoinGecko API 密钥

要获取 CoinGecko API 密钥，请前往开发者面板，然后点击右上角的“+添加新密钥” 。有关生成和设置密钥的详细说明，请参阅本指南。

我们将使用时间范围内的 OHLC 图表端点来获取历史数据，该端点适用于分析师套餐及以上版本。您也可以使用免费替代方案，使用此端点。唯一的区别是，在演示端点上，您无法指定时间范围。

步骤 1. 设置您的环境

首先，创建一个空目录，作为项目的根目录。在根目录中创建一个新的虚拟环境，这样我们就可以在本地安装所需的代码，而无需对全局 Python 环境进行任何更改。

现在让我们配置我们的 Python 应用程序。运行以下命令来创建并激活您的环境：

	# Create a virtual environment
	python -m venv env

	# Activate env on macOS/Linux
	source env/bin/activate # For macOS/Linux

	# activate env on Windows
	env\scripts\activate

如果使用 VS Code，您的 IDE 可能还会询问您是否要使用本地 Python 编译器 - 选择“是”。

安装要求

现在我们可以安装项目所需的依赖了。最简单的方法是将下面的文件复制到根目录下的 requirements.txt 文件中，然后运行pip install -r requirements.txt。

	backtesting==0.6.1
	bokeh==3.6.2
	certifi==2025.1.31
	charset-normalizer==3.4.1
	contourpy==1.3.1
	idna==3.10
	Jinja2==3.1.5
	joblib==1.4.2
	MarkupSafe==3.0.2
	numpy==2.2.2
	packaging==24.2
	pandas==2.2.3
	pillow==11.1.0
	python-dateutil==2.9.0.post0
	python-dotenv==1.0.1
	pytz==2025.1
	PyYAML==6.0.2
	requests==2.32.3
	six==1.17.0
	tornado==6.4.2
	tzdata==2025.1
	urllib3==2.3.0
	xyzservices==2025.1.0

安装 Ta-Lib（可选）

💡 专业提示：如果您的策略不使用技术指标并且仅对原始蜡烛数据进行操作，则可以跳过此步骤。

我们还需要安装一个依赖包：ta-lib。这是一个非常棒的 Python 库，可以根据原始数据计算指标值。与上述依赖包不同，ta-lib要求我们使用发布文件并自行构建包，否则安装过程中会出错。

前往项目的发布页面，选择与您的操作系统、CPU 架构和 Python 版本匹配的版本。例如，我运行的是 64 位 Windows 11 系统，Python 3.11 和 x86 CPU 架构。对我来说，正确的版本是 ta_lib-0.6.0-cp311-cp311-win_amd64.whl。

要在搭载 Python 3.11 和 M1 或更高版本芯片的 Macbook 上运行此程序，您可以使用以下版本：ta_lib-0.6.0-cp311-cp311-win_arm64.whl。下载适合您机器的正确版本后，将文件拖放到项目根目录下。从项目根目录，使用刚刚下载的文件安装软件包。

例如：pip install ta_lib-0.6.0-cp311-cp311-win_amd64.whl。这应该可以满足所有项目要求。

创建项目脚手架

在根目录中，创建services和utils目录，以及一个空的.env文件和一个空的main.py文件。它应该如下所示：

在.env文件中，定义一个名为CG_API_KEY的变量，并将你的 CoinGecko API 密钥指定为其值。我们将使用它来安全地将密钥加载到我们的应用程序中，而无需在项目文件中对其进行硬编码。

步骤 2. 定义实用程序

在我们定义的utils目录中，创建一个名为load_env.py的文件。这将帮助我们加载 API 密钥并定义我们可能拥有的任何其他配置选项。

	import os
	from dotenv import load_dotenv

	load_dotenv()
	cg_api_key = os.getenv("CG_API_KEY")
	take_profit = 1.1 # 10%
	stop_loss = 0.9 # -10%
	size = 0.1 # 10% from total amount
	total_amount = 10000000

请注意，除了我们存储在cg_api_key 中的 API 密钥之外，我们还定义了一些基本的策略设置，例如take_profit、stop_loss、order size和total_amount。

您可以随意调整这些设置以满足您的需求，并在回测期间尝试不同的设置，以找到适合您策略的止损和获利的最佳组合。

如果输入金额远小于资产价格，回测库的行为可能会变得难以预测。为了避免这种情况，我们将输入金额设置得足够高，以防止出现问题。由于我们主要关注的是利润百分比，因此在此阶段无需考虑绝对值。

步骤 3. 构建服务

在我们的应用程序上下文中，我们的服务是一些特定的类，它们将帮助我们与正在使用的各种工具进行交互。为了使我们的工具正常工作，我们需要使用 CoinGecko API 获取历史数据，然后使用Backtesting.py库对我们的策略进行回测。

CoinGecko 服务

让我们从 CoinGecko 服务开始。在服务下，创建一个名为coingecko_service.py的新文件。在这个文件中，我们将定义一个CoinGecko类，其中包含一个简单的构造函数，用于存储 API 的根 URL 和所需的标头。

💡 专业提示：如果您使用的是 Demo API 计划，请务必将 root.headers 中存储 API 密钥的属性从“x_cg_pro_api_key”更改为“x-cg-demo-api-key”。

我们还将定义一个名为get_historical_prices()的方法。

	from typing import Literal
	import requests
	from utils.load_env import *


	class CoinGecko:
	def __init__(self):
	self.root = "https://pro-api.coingecko.com/api/v3"
	self.headers = {
	"accept": "application/json",
	"x_cg_pro_api_key": f"{cg_api_key}",
	}

	def get_historical_prices(
	self,
	coin_id: str,
	vs_currency: str,
	from_unix: int,
	to_unix: int,
	interval: Literal["daily", "hourly"],
	):
	request_url = (
	self.root
	+ f"/coins/{coin_id}/ohlc/range?vs_currency={vs_currency}&from={from_unix}&to={to_unix}&interval={interval}"
	)

	return requests.get(request_url, self.headers).json()

我们现在有了一个获取特定时间范围内历史数据的方法，并且可以指定 K 线图的间隔是按天还是按小时。这取决于我们计划测试的策略。

💡 专业提示：如果您使用的是 Demo API 计划，请记住将请求 URL 更改为/ohlc（而不是 /ohlc/range）并删除“from”和“to”参数。

回测服务

为了构建我们的回测服务，我们将使用Backtester.py库。这将省去我们自己构建回测引擎的麻烦。我们只需为要使用的方法编写自己的包装器即可。

	import pandas as pd
	from backtesting import Backtest, Strategy
	from typing import Type, Optional, Dict, Any, Union


	class BackTester:
	def __init__(
	self,
	data: pd.DataFrame,
	strategy: Type[Strategy],
	cash: float = 10000,
	commission: float = 0.0,
	spread: float = 0.0,
	trade_on_close: bool = False,
	**kwargs
	):

	if not self.validate_data(data):
	raise ValueError("Invalid data format. Missing required OHLC columns.")

	self._backtest = Backtest(
	data=data,
	strategy=strategy,
	cash=cash,
	commission=commission,
	spread=spread,
	trade_on_close=trade_on_close,
	**kwargs
	)
	self._results: Optional[pd.Series] = None

	@staticmethod
	def validate_data(data: pd.DataFrame) -> bool:
	"""Validate DataFrame contains required OHLC columns"""
	required = {"Open", "High", "Low", "Close"}
	return required.issubset(data.columns)

	def run(self, **strategy_params: Any) -> pd.Series:
	self._results = self._backtest.run(**strategy_params)
	return self._results

	def optimize(
	self,
	maximize: Union[str, callable] = "SQN",
	method: str = "grid",
	max_tries: Optional[Union[int, float]] = None,
	constraint: Optional[callable] = None,
	**params: Any
	) -> pd.Series:
	return self._backtest.optimize(
	maximize=maximize,
	method=method,
	max_tries=max_tries,
	constraint=constraint,
	**params
	)

	def plot(
	self,
	results: Optional[pd.Series] = None,
	filename: Optional[str] = None,
	plot_width: Optional[int] = None,
	**plot_kwargs: Any
	) -> None:
	if results is None and self._results is None:
	raise ValueError("No results to plot. Run backtest first.")

	self._backtest.plot(
	results=results or self._results,
	filename=filename,
	plot_width=plot_width,
	**plot_kwargs
	)

	@property
	def results(self) -> Optional[pd.Series]:
	"""Get latest backtest results"""
	return self._results

	@property
	def trades(self) -> Optional[pd.DataFrame]:
	"""Get detailed trades DataFrame"""
	return self._results._trades if self._results else None

	@property
	def equity_curve(self) -> Optional[pd.DataFrame]:
	"""Get equity curve DataFrame"""
	return self._results._equity_curve if self._results else None

我们将使用的主要方法是：run、optimize和plot。其余的只是属性，如果存在则返回一个值，除了validate_data方法，如果提供的历史数据不包含我们需要的标题，我们将使用该方法抛出错误。

步骤3.制定策略

当我们初始化BackTester类的实例时，我们需要传递一个Strategy参数。我们的目标是能够在代码改动极少甚至完全不改动的情况下测试不同的策略，因此我们需要确保我们的策略是即插即用的。

为此，我们将在项目根目录下创建一个名为“strategies”的新目录。

逢低买入策略

让我们从一个简单的价格变化策略开始。我们希望每当比特币价格在一天内下跌超过 10% 时就买入。

在策略目录中，创建一个名为buy_the_dip_strategy.py的新文件

为了使我们的策略与Backtester.py兼容，我们必须定义一个名为next()的方法。您可能已经注意到，我们的策略中没有迭代逻辑。这是因为 Backtester 在幕后完成了繁重的工作，将本策略中定义的逻辑应用于历史数据中的每根 K 线。

我们的逻辑评估如下：针对历史数据数组中的每根 K 线，计算其价格变化百分比。如果价格变化小于 -10%，我们就会下单。

这里没有卖出逻辑，因为我们传递了止盈和止损，这两个函数都由 Backtester.py 自动处理。但是，如果您希望添加卖出逻辑，可以在适当的条件下调用self.sell()来实现。

黄金交叉策略

让我们定义另一个策略——这次使用SMA技术指标。在交易中，黄金交叉是指50周期SMA与200周期SMA交叉的时刻，预示着进一步上涨。

在同一个策略目录中，创建一个名为golden_cross_strategy.py的新文件。

和以前一样，我们将继承Strategy类的属性，并定义构造函数和next()方法。这次的主要区别在于，我们需要处理指标数据，而不仅仅是原始价格数据。我们可以手动计算 SMA，也可以让 ta-lib 和 Backtester 帮我们完成繁重的工作：

	from backtesting import Strategy
	import talib as ta
	from utils.load_env import *


	class GoldenCrossStrategy(Strategy):
	short_sma_period = 50 # Short-term SMA
	long_sma_period = 200 # Long-term SMA

	def init(self):
	self.short_sma = self.I(ta.SMA, self.data.Close, self.short_sma_period)
	self.long_sma = self.I(ta.SMA, self.data.Close, self.long_sma_period)

	def next(self):
	if (
	self.short_sma[-1] > self.long_sma[-1]
	and self.short_sma[-2] <= self.long_sma[-2]
	):
	self.buy(
	size=size,
	sl=stop_loss * self.data.Close[-1],
	tp=take_profit * self.data.Close[-1],
	)
	print(f"Golden cross detected! Bought at {self.data.Close[-1]}")
	elif (
	self.short_sma[-1] < self.long_sma[-1]
	and self.short_sma[-2] >= self.long_sma[-2]
	):
	self.sell()
	print(f"Death cross detected! Sold at {self.data.Close[-1]}")

现在，该策略将在每次出现黄金交叉时下达买入订单，并在每次遇到死亡交叉时（即当 SMA 50 穿过 SMA 200 时）下达卖出订单。

步骤 4. 使用 Python 进行加密货币回测

策略定义好了，我们就可以开始测试了。剩下的就是获取数据，并将其与策略一起传递给回测器。

	from services.backtester_service import BackTester
	from services.coingecko_service import CoinGecko
	from utils.load_env import *
	import pandas as pd
	from strategies.buy_the_dip_strategy import BuyTheDip
	from strategies.golden_cross_strategy import GoldenCross

	cg = CoinGecko()
	data = cg.get_historical_prices("bitcoin", "usd", 1736424000, 1738152000, "hourly")

	# Define column names
	columns = ["Timestamp", "Open", "High", "Low", "Close"]

	# Convert to DataFrame
	df = pd.DataFrame(data, columns=columns)

	# Initialize backtester
	backtester = BackTester(
	data=df,
	strategy=BuyTheDip,
	cash=total_amount,
	commission=0.001,
	)
	output = backtester.run()
	print(output)
	backtester.plot()