加密与Python进行回测:逐步指南
在交易中,加密进行回测是指使用历史市场数据评估交易策略的过程,以查看其过去的表现。
它允许交易者在实时交易中使用策略之前评估盈利能力,风险和整体效率。通过基于过去的数据模拟交易,交易者可以完善自己的方法,确定潜在的弱点,并在不冒险的情况下对其战略获得信心。
在本指南中,我们将利用一个加密货币进行回测工具 Coingecko API,这使得使用技术分析和指标可以轻松测试各种交易策略,从购买蘸酱等简单的方法到更复杂的方法。
像往常一样,您会在文章末尾找到指向GitHub存储库的链接,从而使您可以直接潜入并进行实验。
先决条件
在开始构建加密货币测试工具之前,我们需要以下内容:
- 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代码,您的IDE也可能会询问您是否想使用本地Python编译器 - 选择是。
安装要求
我们现在准备安装项目的要求。最简单的方法是将下面的文件复制到一个名为unignts.txt然后运行的文件中的根目录pip install -r要求.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 |
安装裙子(可选)
我们还有一个需要安装的要求:裙子。这是一个很棒的Python库,用于从原始数据中计算指示器值。与上面的要求不同,ta-lib要求我们使用发行文件并自己构建包装,否则,它在安装过程中将出错。
前往项目的发布页面并选择一个与您的操作系统,CPU体系结构和Python版本匹配的版本。例如,我正在使用Python 3.11和X86 CPU架构运行64位Windows 11。对我来说,正确的版本是TA_LIB-0.6.0-CP311-CP311-WIN_AMD64.WHL。
要在具有Python 3.11和M1或更高版本的MacBook上运行此功能,您可以使用以下版本:TA_LIB-0.6.0.6.0-CP311-CP311-WIN_ARM64.WHL。为计算机下载了正确的版本后,将文件放在项目根中。从您的项目root中,使用您刚下载的文件安装软件包。
例如:PIP安装TA_LIB-0.6.0-CP311-CP311-WIN_AMD64.WHL。这应该照顾所有项目要求。
创建项目脚手架
在您的根目录中,创建服务和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 |
请注意,除了我们存储在其中的API密钥之外cg_api_key,我们还定义了一些基本策略设置,例如take_profit,stop_loss,订单尺寸,一个总金额.
随意调整这些设置以适应您的需求,并在回测期间进行不同的设置,以找到最佳的停止损失组合并为您的策略获利。
如果投入金额远小于资产的价格,则进行回测图书馆可能会不可预测。为了避免这种情况,我们将其设置得足够高以防止问题。由于我们的主要重点是利润百分比,因此在此阶段,绝对价值不是关注的。
步骤3.建立服务
在应用程序的背景下,我们的服务是特定的类,可以帮助我们与所使用的各种工具进行交互。为了使我们的工具工作,我们需要获取史料使用Coingecko API,然后使用Backtesting.py图书馆。
Coingecko服务
让我们从Coingecko服务开始。在服务下,创建一个名为的新文件coingecko_service.py。在此文件中,我们将定义一个Co Ringecko与简单的构造函数类,该构造函数存储API和所需标头的根URL。
我们还将定义一种称为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() |
现在,我们有一种在一个时间范围内获取历史数据的方法,我们可以指定我们的蜡烛是否具有每日或每小时的间隔。这将取决于我们计划测试的策略。
烘焙服务
为了建立我们的后期服务,我们将使用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 |
我们将使用的主要方法是:跑步,优化, 和阴谋。其余的只是属性返回值(如果存在),除了validate_data如果提供的历史数据不包含所需的标题,则我们使用的方法来丢弃错误。
步骤3。制定策略
当我们初始化一个实例时后期上课,我们需要通过战略。我们的目标是能够以最小或根本没有更改代码的不同策略来测试不同的策略,因此我们需要确保我们的策略是插件。
为此,我们将创建一个名为的新目录策略在我们项目的根源。
购买浸入策略
让我们从一个简单的价格更改策略开始。每天,每天的价格下跌超过10%时,我们都想购买比特币。
在策略目录中,创建一个名为的新文件buy_the_dip_strategy.py
为了使我们的策略与Backtester.py,我们必须定义一种称为的方法下一个()。您可能已经注意到我们的策略没有迭代逻辑。那是因为后卫正在幕后进行繁重的工作,在我们的历史数据中应用了每种蜡烛中定义的逻辑。
我们的逻辑将如下评估:对于我们的历史数据阵列中的每个蜡烛,计算价格变化为百分比。如果价格变化小于-10%,我们将下订单。
没有出售逻辑,因为我们正在传递take_profit和stop_loss,这两者都是由backtester.py自动处理的。但是,如果您想添加出售逻辑,则可以通过致电self.sell()在适当的条件下。
黄金交叉策略
让我们定义另一种策略 - 这次使用SMA技术指标。在交易中,金色交叉是指50个周期SMA穿过200段SMA的那一刻。
在同一策略目录中,创建一个名为的新文件golden_cross_strategy.py.
和以前一样,我们将继承战略上课并定义我们的构造函数,下一个()方法。这次的主要区别是我们需要使用指标数据,而不仅仅是原始价格数据。我们可以手动计算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() |
请注意,我们的历史数据的时间范围必须作为UNIX时间戳传递。现在剩下的就是运行脚本并分析您的结果。这是运行我们的示例输出的示例buythedip战略。
如果您包括backtester.plot()在main.py的末尾,您会注意到目录的根部创建了一个新的HTML文件。这代表了该策略的买卖活动的绩效图。只需使用浏览器打开此文件即可可视化策略的性能:
恭喜,您已经成功地测试了交易策略!要测试其他策略,只需将其传递给后期。我们构建的框架应该使您可以轻松测试和开发任何数量的策略。
考虑因素
二次测试是评估的强大工具交易策略,这并非没有限制。过去的绩效并不总是表明未来的回报,因为市场周期的变化,流动性状况发生了变化,而不可预见的事件也会破坏最经过良好测试的策略。
过度拟合是另一种风险,在这种风险中,一项策略经过微调,可以在历史数据上表现出色,但由于其对过去可能不再相关的过去模式的依赖,现场市场失败了。进行回测的还假设了完美的贸易执行,忽略了滑板,交易成本和订单深度等因素,这可能会影响现实世界中的盈利能力。
重要的是要谨慎行事,并通过使用纸交易或少量预算。
如果您发现这篇文章有帮助,请务必查看我们的完整的加密进行回测指南!