tech

用Alpaca API开发交易机器人

在本文中，我们将介绍创建加密货币完整交易算法的基础知识，甚至将其连接到 Alpaca 进行交易。当然，Python 是完成此类任务的自然选择！

一键发币： SOL | BNB | ETH | BASE | Blast | ARB | OP | POLYGON | AVAX | FTM | OK

请注意，本文仅供一般参考。所有截图仅供说明之用。所表达的观点和意见均为作者的观点和意见，并不反映或代表 Alpaca 的观点和意见。Alpaca 不推荐任何特定的证券、加密货币或投资策略。

在本文中，我们将介绍创建加密货币完整交易算法的基础知识，甚至将其连接到 Alpaca 进行交易。当然，Python 是完成此类任务的自然选择！

让我们开始吧。

1、设置 Alpaca 账户

在深入研究代码之前，我们必须在 Alpaca Markets 中创建一个免费经纪账户。这家经纪公司专门为我们提供出色的 API 接口，获取实时市场数据，甚至下订单。

注册表单没有什么特别之处，所以我会给你一些信用，并假设你成功创建了你的账户。

登录新创建的帐户后，你将能够选择“实时”（真实货币）帐户或“模拟”帐户。为了降低您掌握算法交易的成本，我建议你从使用“模拟”帐户开始。你可以在 Alpaca 徽标下方的右上角下拉菜单中切换到此帐户。

进入后，单击 API 密钥部分下的“查看”按钮，然后继续复制两个字符串。

就是这样——现在让我们进入有趣的部分：编写我们的第一个算法交易脚本！

2、使用 Python 调用 Alpaca 的 API

如果你尚未安装 Alpaca 的库，请继续执行：

pip install alpaca-trade-api

启动一个新的 Jupyter 笔记本或空白文件并导入所需的库：

from alpaca_trade_api.rest import REST, TimeFrame
import pandas as pd

我们还来设置一些将在整个教程中使用的参数，别忘了将KEY替换为你自己的值！

BASE_URL = "https://paper-api.alpaca.markets"
KEY_ID = <YOUR_ALPACA_KEY>
SECRET_KEY = <YOUR_ALPACA_SECRET_KEY>

2.1 从 Alpaca 的 API 获取历史数据

现在，我们已准备好连接到 API 并请求第一个历史数据 DataFrame，例如比特币。

# Instantiate REST API Connection
api = REST(key_id=KEY_ID,secret_key=SECRET_KEY,base_url="https://paper-api.alpaca.markets")

# Fetch 1Minute historical bars of Bitcoin
bars = api.get_crypto_bars("BTCUSD", TimeFrame.Minute).df

你可能已经看到，我们正在从多个交易所接收价格。目前，Alpaca 将其客户的订单路由到 FTXU，但我们将坚持使用 CBSE（Coinbase）进行计算。这是因为 FTXU 流动性较差，因此缺少 1 分钟数据，而 Coinbase 是最活跃的加密货币交易所之一。

# Filter data by exchange
bars = bars[bars.exchange == 'CBSE']
print(bars)

2.2 在 Alpaca 的 API 上提交订单

向 Alpaca 提交订单非常简单。在本课中，我们将坚持使用市价单进行交易。

# Create a market order to buy 1 Bitcoin
order_buy = api.submit_order('BTCUSD', qty=1, side='buy')

这不仅会在交易所提交订单，还会返回 Order 类型的对象：

# Create a market order to sell 1 Bitcoin
order_sell = api.submit_order('BTCUSD', qty=1, side='sell')

还可以轻松找出你在特定资产上的当前状况。

# Get current position on Bitcoin. Yields error if we don't hold any
api.get_position('BTC')

# Get all positions and find out if we hold any Bitcoin
positions = api.list_positions()
position_qty = 0
for p in positions:
    if p.symbol == 'BTCUSD':
        position_qty = float(p.qty)
print(position_qty)

了解了我们需要的最重要的功能后，是时候开始做更有用的事情，并创建一个以最纯粹的随机方式进行买卖的算法：通过抛硬币！

别担心，这只是设置我们的逻辑。稍后，我们将继续创建一个由两个相互交叉的移动平均线组成的策略。

3、创建一个简单的交易算法

第一个算法非常简单，包括抛硬币来检查我们是否应该买入或卖出比特币。此外，我们只在还没有头寸的情况下买入。相反，只有当我们的投资组合中目前有比特币时，我们才会卖出。

让我们首先采用之前检查我们当前头寸的脚本，并将其转换为我们可以重用的函数：

def get_position(symbol):
    positions = api.list_positions()
    for p in positions:
        if p.symbol == symbol:
            return float(p.qty)
    return 0

现在，让我们来看看算法本身。为了调试和跟踪正在发生的事情，我冒昧地打印出了几乎所有的东西。

import random

SYMBOL = 'BTCUSD'
while True:
    # GET OUR CURRENT POSITION
    position = get_position(symbol=SYMBOL)
    
    # SCIENTIFICALLY CHECK IF WE SHOULD BUY OR SELL
    gods_say_buy = random.choice([True, False])
    print(f"Holding: {position} / Gods: {gods_say_buy}")

    #CHECK IF WE SHOULD BUY
    if position == 0 and gods_say_buy == True:
        # WE BUY ONE BITCOIN
        print('The gods have spoken:')
        print(f'Symbol: {SYMBOL} / Side: BUY / Quantity: 1')
        api.submit_order(SYMBOL, qty=1, side='buy')
    #HECK IF WE SHOULD SELL
    elif position > 0 and gods_say_buy == False:
        # WE SELL ONE BITCOIN
        print('The gods have spoken:')
        print(f'Symbol: {SYMBOL} / Side: SELL / Quantity: 1')
        api.submit_order(SYMBOL, qty=1, side='sell')
    print('Lets wait for the gods to manifest again...')
    print("*"*20)
    time.sleep(10)

为了提高我们的胜算，我求助于某种更高级的存在，我简单地称之为“神”。代码几乎是不言自明的，但简而言之，我们有一个无限循环，每十秒检查一次神是否希望我们购买比特币。根据我们是否持有头寸，我们服从他们的意愿。

4、基于规则（移动平均线交叉）交易算法

该算法将遵循由两个移动平均线组成的交易策略。第一个，我们称之为“快速”，它将包括计算前 12 分钟收盘价的平均值。另一方面，我们有“慢速”移动平均线，它的作用相同，但考虑了前 24 分钟的收盘价。

它们被称为“快速”和“慢速”的原因很简单：每个新价格对“快速”移动平均线的影响更大，因为我们在计算中使用的周期更少。整个脚本如下所示（假设我们已经启动了 API 并导入了以前的库）。

from datetime import datetime, timedelta
import math
import time

SYMBOL = 'BTCUSD'
SMA_FAST = 12
SMA_SLOW = 24
QTY_PER_TRADE = 1


# Description is given in the article
def get_pause():
    now = datetime.now()
    next_min = now.replace(second=0, microsecond=0) + timedelta(minutes=1)
    pause = math.ceil((next_min - now).seconds)
    print(f"Sleep for {pause}")
    return pause

# Same as the function in the random version
def get_position(symbol):
    positions = api.list_positions()
    for p in positions:
        if p.symbol == symbol:
            return float(p.qty)
    return 0


# Returns a series with the moving average
def get_sma(series, periods):
    return series.rolling(periods).mean()

# Checks whether we should buy (fast ma > slow ma)
def get_signal(fast, slow):
    print(f"Fast {fast[-1]}  /  Slow: {slow[-1]}")
    return fast[-1] > slow[-1]

# Get up-to-date 1 minute data from Alpaca and add the moving averages
def get_bars(symbol):
    bars = api.get_crypto_bars(symbol, TimeFrame.Minute).df
    bars = bars[bars.exchange == 'CBSE']
    bars[f'sma_fast'] = get_sma(bars.close, SMA_FAST)
    bars[f'sma_slow'] = get_sma(bars.close, SMA_SLOW)
    return bars

while True:
    # GET DATA
    bars = get_bars(symbol=SYMBOL)
    # CHECK POSITIONS
    position = get_position(symbol=SYMBOL)
    should_buy = get_signal(bars.sma_fast,bars.sma_slow)
    print(f"Position: {position} / Should Buy: {should_buy}")
    if position == 0 and should_buy == True:
        # WE BUY ONE BITCOIN
        api.submit_order(SYMBOL, qty=QTY_PER_TRADE, side='buy')
        print(f'Symbol: {SYMBOL} / Side: BUY / Quantity: {QTY_PER_TRADE}')
    elif position > 0 and should_buy == False:
        # WE SELL ONE BITCOIN
        api.submit_order(SYMBOL, qty=QTY_PER_TRADE, side='sell')
        print(f'Symbol: {SYMBOL} / Side: SELL / Quantity: {QTY_PER_TRADE}')

    time.sleep(get_pause())
    print("*"*20)

你可能想知道 get_pause 函数的全部含义。基本上，由于我们使用的是 1 分钟数据进行交易，因此我们希望每 60 秒运行一次循环。不仅如此，我们还希望在每分钟开始时开始每次迭代，以便在我们可用时立即获取最新的 1 分钟条形图。

get_position 函数与随机算法中的完全相同，而 get_signal 检查最近的“快速”移动平均线是否大于“慢速”移动平均线。 get_bars 获取目标资产的 1 分钟条形图，仅保留与 Coinbase（交易所等于“CBSE”）相对应的行，并在返回 DataFrame 之前添加两个移动平均线。