
在本系列的第一部分中,我们讨论了文件夹结构,添加了我们的第一个Price业务对象,并为策略抽象层进行编码。
下面将通过编码我们需要的货币和订单的业务对象开始。
此处,货币业务对象存储所有不同的加密货币或法定货币。
./models/order.py
from models.model import AbstractModel
class Currency(AbstractModel):
name: str = ''
symbol: str = ''
fiat: bool
def __init__(self, **kwargs):
super().__init__(**kwargs)
然后是订单业务对象处理我们在交易所创建的订单。
./models/order.py
from models.model import AbstractModel
class Order(AbstractModel):
BUY = 'BUY'
SELL = 'SELL'
TYPE_LIMIT = 'LIMIT'
TYPE_MARKET = 'MARKET'
TYPE_STOP_LOSS = 'STOP_LOSS'
TYPE_STOP_LOSS_LIMIT = 'STOP_LOSS_LIMIT'
TYPE_TAKE_PROFIT = 'TAKE_PROFIT'
TYPE_TAKE_PROFIT_LIMIT = 'TAKE_PROFIT_LIMIT'
TYPE_LIMIT_MAKER = 'LIMIT_MAKER'
uuid = ''
side: str = ''
type: str = TYPE_LIMIT
symbol: str = ''
currency: str = ''
asset: str = ''
price: float = 0
quantity: int = 0
test: bool = False
def __init__(self, **kwargs):
super().__init__(**kwargs)
然后构建交换抽象层,并开发我们的第一个连接器。由于这里不是编写API包装器的重点,因此为了方便起见,我们将使用非官方的Binance API包装器库python-binance(https://python-binance.readthedocs.io/en/latest/binance.html)。
./exchanges/exchange.py
import datetime
from api import utils
from abc import ABC, abstractmethod
from twisted.internet import reactor
from strategies.strategy import Strategy
from models.order import Order
class Exchange(ABC):
currency: str
asset: str
strategy: Strategy
def __init__(self, key: str, secret: str):
self.apiKey = key
self.apiSecret = secret
self.name = None
self.client = None
self.socketManager = None
self.socket = None
self.currency = ''
self.asset = ''
self.strategy = None
def set_currency(self, symbol: str):
self.currency = symbol
def set_asset(self, symbol: str):
self.asset = symbol
def set_strategy(self, strategy: Strategy):
self.strategy = strategy
def compute_symbol_pair(self):
return utils.format_pair(self.currency, self.asset)
# abstract methods
# Override to set current exchange symbol pair notation (default with _ separator currency_asset ex: eur_btc)
@abstractmethod
def get_symbol(self):
return self.compute_symbol_pair(self)
# Get current symbol ticker
@abstractmethod
def symbol_ticker(self):
pass
# Get current symbol ticker candle for given interval
@abstractmethod
def symbol_ticker_candle(self, interval):
pass
# Get current symbol historic value
@abstractmethod
def historical_symbol_ticker_candle(self, start: datetime, end=None, interval=60):
pass
# Get balance for a given currency
@abstractmethod
def get_asset_balance(self, currency):
pass
# Create an exchange order
@abstractmethod
def order(self, order: Order):
pass
# Create an exchange test order
@abstractmethod
def test_order(self, order: Order):
pass
# Check an exchange order status
@abstractmethod
def check_order(self, orderId):
pass
# Cancel an exchange order
@abstractmethod
def cancel_order(self, orderId):
pass
# WebSocket related methods
@abstractmethod
def get_socket_manager(self, purchase):
pass
@abstractmethod
def websocket_event_handler(self, msg):
pass
def start_socket(self):
print('Starting WebSocket connection...')
self.socketManager.start()
def close_socket(self):
self.socketManager.stop_socket(self.socket)
self.socketManager.close()
# properly terminate WebSocket
reactor.stop()
@abstractmethod
def start_symbol_ticker_socket(self, symbol: str):
pass
我们的第一个Binance API连接器。
./exchanges/binance.py
现在,只需调用我们的策略启动方法,我们便拥有了一个极简而强大的交易机器人系统。使用自己的指标和策略,您可以通过传递订单来开始买卖。但是,在开始编写自己的策略代码之前,请确保对其进行测试并安全运行。所以你需要一个系统来做到这一点。
在下一部分中,我们将实现回测模式,以针对交易所的历史数据来测试您的策略,并提供导入这些数据的服务。
另外,我们将所有这些部分与全局配置和一些依赖项一起打包进命令行工具中。
最后我们将通过查看如何对该代码进行容器化和工业化实现来进一步介绍。

在本系列的第一部分中,我们讨论了文件夹结构,添加了我们的第一个Price业务对象,并为策略抽象层进行编码。
下面将通过编码我们需要的货币和订单的业务对象开始。
此处,货币业务对象存储所有不同的加密货币或法定货币。
./models/order.py
from models.model import AbstractModel
class Currency(AbstractModel):
name: str = ''
symbol: str = ''
fiat: bool
def __init__(self, **kwargs):
super().__init__(**kwargs)
然后是订单业务对象处理我们在交易所创建的订单。
./models/order.py
from models.model import AbstractModel
class Order(AbstractModel):
BUY = 'BUY'
SELL = 'SELL'
TYPE_LIMIT = 'LIMIT'
TYPE_MARKET = 'MARKET'
TYPE_STOP_LOSS = 'STOP_LOSS'
TYPE_STOP_LOSS_LIMIT = 'STOP_LOSS_LIMIT'
TYPE_TAKE_PROFIT = 'TAKE_PROFIT'
TYPE_TAKE_PROFIT_LIMIT = 'TAKE_PROFIT_LIMIT'
TYPE_LIMIT_MAKER = 'LIMIT_MAKER'
uuid = ''
side: str = ''
type: str = TYPE_LIMIT
symbol: str = ''
currency: str = ''
asset: str = ''
price: float = 0
quantity: int = 0
test: bool = False
def __init__(self, **kwargs):
super().__init__(**kwargs)
然后构建交换抽象层,并开发我们的第一个连接器。由于这里不是编写API包装器的重点,因此为了方便起见,我们将使用非官方的Binance API包装器库python-binance(https://python-binance.readthedocs.io/en/latest/binance.html)。
./exchanges/exchange.py
import datetime
from api import utils
from abc import ABC, abstractmethod
from twisted.internet import reactor
from strategies.strategy import Strategy
from models.order import Order
class Exchange(ABC):
currency: str
asset: str
strategy: Strategy
def __init__(self, key: str, secret: str):
self.apiKey = key
self.apiSecret = secret
self.name = None
self.client = None
self.socketManager = None
self.socket = None
self.currency = ''
self.asset = ''
self.strategy = None
def set_currency(self, symbol: str):
self.currency = symbol
def set_asset(self, symbol: str):
self.asset = symbol
def set_strategy(self, strategy: Strategy):
self.strategy = strategy
def compute_symbol_pair(self):
return utils.format_pair(self.currency, self.asset)
# abstract methods
# Override to set current exchange symbol pair notation (default with _ separator currency_asset ex: eur_btc)
@abstractmethod
def get_symbol(self):
return self.compute_symbol_pair(self)
# Get current symbol ticker
@abstractmethod
def symbol_ticker(self):
pass
# Get current symbol ticker candle for given interval
@abstractmethod
def symbol_ticker_candle(self, interval):
pass
# Get current symbol historic value
@abstractmethod
def historical_symbol_ticker_candle(self, start: datetime, end=None, interval=60):
pass
# Get balance for a given currency
@abstractmethod
def get_asset_balance(self, currency):
pass
# Create an exchange order
@abstractmethod
def order(self, order: Order):
pass
# Create an exchange test order
@abstractmethod
def test_order(self, order: Order):
pass
# Check an exchange order status
@abstractmethod
def check_order(self, orderId):
pass
# Cancel an exchange order
@abstractmethod
def cancel_order(self, orderId):
pass
# WebSocket related methods
@abstractmethod
def get_socket_manager(self, purchase):
pass
@abstractmethod
def websocket_event_handler(self, msg):
pass
def start_socket(self):
print('Starting WebSocket connection...')
self.socketManager.start()
def close_socket(self):
self.socketManager.stop_socket(self.socket)
self.socketManager.close()
# properly terminate WebSocket
reactor.stop()
@abstractmethod
def start_symbol_ticker_socket(self, symbol: str):
pass
我们的第一个Binance API连接器。
./exchanges/binance.py
现在,只需调用我们的策略启动方法,我们便拥有了一个极简而强大的交易机器人系统。使用自己的指标和策略,您可以通过传递订单来开始买卖。但是,在开始编写自己的策略代码之前,请确保对其进行测试并安全运行。所以你需要一个系统来做到这一点。
在下一部分中,我们将实现回测模式,以针对交易所的历史数据来测试您的策略,并提供导入这些数据的服务。
另外,我们将所有这些部分与全局配置和一些依赖项一起打包进命令行工具中。
最后我们将通过查看如何对该代码进行容器化和工业化实现来进一步介绍。
from datetime import datetime
from math import floor
from binance.client import Client
from binance.enums import *
from binance.websockets import BinanceSocketManager
from api import utils
from exchanges import exchange
from models.order import Order
from models.price import Price
class Binance(exchange.Exchange):
def __init__(self, key: str, secret: str):
super().__init__(key, secret)
self.client = Client(self.apiKey, self.apiSecret)
self.name = self.__class__.__name__
def get_client(self):
return self.client
def get_symbol(self):
return self.currency + self.asset
def symbol_ticker(self):
response = self.client.get_symbol_ticker(symbol=self.get_symbol())
return Price(pair=self.get_symbol(), currency=self.currency.lower(), asset=self.asset.lower(), exchange=self.name.lower(),
current=response['price'])
def symbol_ticker_candle(self, interval=Client.KLINE_INTERVAL_1MINUTE):
return self.client.get_klines(symbol=self.get_symbol(), interval=interval)
def historical_symbol_ticker_candle(self, start: datetime, end=None, interval=Client.KLINE_INTERVAL_1MINUTE):
# Convert default seconds interval to string like "1m"
if isinstance(interval, int):
interval = str(floor(interval/60)) + 'm'
output = []
for candle in self.client.get_historical_klines_generator(self.get_symbol(), interval, start, end):
output.append(
Price(pair=self.compute_symbol_pair(), currency=self.currency.lower(), asset=self.asset.lower(), exchange=self.name.lower(),
current=candle[1], lowest=candle[3], highest=candle[2], volume=candle[5], openAt=utils.format_date(datetime.fromtimestamp(int(candle[0])/1000)))
)
return output
def get_asset_balance(self, currency):
response = self.client.get_asset_balance(currency)
return response['free']
def order(self, order: Order):
return self.client.create_order(
symbol=order.symbol,
side=order.side,
type=order.type,
timeInForce=TIME_IN_FORCE_GTC,
quantity=order.quantity,
price=order.price
)
def test_order(self, order: Order):
return self.client.create_test_order(
symbol=order.symbol,
side=order.side,
type=order.type,
timeInForce=TIME_IN_FORCE_GTC,
quantity=order.quantity,
price=order.price
)
def check_order(self, orderId):
return self.client.get_order(
symbol=self.get_symbol(),
orderId=orderId
)
def cancel_order(self, orderId):
return self.client.cancel_order(
symbol=self.get_symbol(),
orderId=orderId
)
def get_socket_manager(self):
return BinanceSocketManager(self.client)
def start_symbol_ticker_socket(self, symbol: str):
self.socketManager = self.get_socket_manager()
self.socket = self.socketManager.start_symbol_ticker_socket(
symbol=self.get_symbol(),
callback=self.websocket_event_handler
)
self.start_socket()
def websocket_event_handler(self, msg):
if msg['e'] == 'error':
print(msg)
self.close_socket()
else:
self.strategy.set_price(
Price(pair=self.compute_symbol_pair(), currency=self.currency, asset=self.asset, exchange=self.name,
current=msg['b'], lowest=msg['l'], highest=msg['h'])
)
self.strategy.run()
from datetime import datetime
from math import floor
from binance.client import Client
from binance.enums import *
from binance.websockets import BinanceSocketManager
from api import utils
from exchanges import exchange
from models.order import Order
from models.price import Price
class Binance(exchange.Exchange):
def __init__(self, key: str, secret: str):
super().__init__(key, secret)
self.client = Client(self.apiKey, self.apiSecret)
self.name = self.__class__.__name__
def get_client(self):
return self.client
def get_symbol(self):
return self.currency + self.asset
def symbol_ticker(self):
response = self.client.get_symbol_ticker(symbol=self.get_symbol())
return Price(pair=self.get_symbol(), currency=self.currency.lower(), asset=self.asset.lower(), exchange=self.name.lower(),
current=response['price'])
def symbol_ticker_candle(self, interval=Client.KLINE_INTERVAL_1MINUTE):
return self.client.get_klines(symbol=self.get_symbol(), interval=interval)
def historical_symbol_ticker_candle(self, start: datetime, end=None, interval=Client.KLINE_INTERVAL_1MINUTE):
# Convert default seconds interval to string like "1m"
if isinstance(interval, int):
interval = str(floor(interval/60)) + 'm'
output = []
for candle in self.client.get_historical_klines_generator(self.get_symbol(), interval, start, end):
output.append(
Price(pair=self.compute_symbol_pair(), currency=self.currency.lower(), asset=self.asset.lower(), exchange=self.name.lower(),
current=candle[1], lowest=candle[3], highest=candle[2], volume=candle[5], openAt=utils.format_date(datetime.fromtimestamp(int(candle[0])/1000)))
)
return output
def get_asset_balance(self, currency):
response = self.client.get_asset_balance(currency)
return response['free']
def order(self, order: Order):
return self.client.create_order(
symbol=order.symbol,
side=order.side,
type=order.type,
timeInForce=TIME_IN_FORCE_GTC,
quantity=order.quantity,
price=order.price
)
def test_order(self, order: Order):
return self.client.create_test_order(
symbol=order.symbol,
side=order.side,
type=order.type,
timeInForce=TIME_IN_FORCE_GTC,
quantity=order.quantity,
price=order.price
)
def check_order(self, orderId):
return self.client.get_order(
symbol=self.get_symbol(),
orderId=orderId
)
def cancel_order(self, orderId):
return self.client.cancel_order(
symbol=self.get_symbol(),
orderId=orderId
)
def get_socket_manager(self):
return BinanceSocketManager(self.client)
def start_symbol_ticker_socket(self, symbol: str):
self.socketManager = self.get_socket_manager()
self.socket = self.socketManager.start_symbol_ticker_socket(
symbol=self.get_symbol(),
callback=self.websocket_event_handler
)
self.start_socket()
def websocket_event_handler(self, msg):
if msg['e'] == 'error':
print(msg)
self.close_socket()
else:
self.strategy.set_price(
Price(pair=self.compute_symbol_pair(), currency=self.currency, asset=self.asset, exchange=self.name,
current=msg['b'], lowest=msg['l'], highest=msg['h'])
)
self.strategy.run()
CCXT:加密货币量化交易神器
CCXT框架 是一个Python/Javascript/PHP的一个交易API框架,对接超过130多个交易所。可用于世界各地的加密货币交易所的连接和交易,以及转账支付处理,可用于存储数据,分析,可视化,指标开发,算法交易,是一个非常容易集成的开箱即用的统一API。 CCXT框架Github地址: https://github.com/ccxt/ccxt 当前功能列表:支持许多交易市场,甚至即将推出的为所有交易提供完整的公共和私人API所有货币,山寨币和标记,价格,订单,交易,代码等…提供用于交叉交换或跨货币分析和套利的可选标准化数据开箱即用的统一的一体化API,非常易于集成适用于Node7.6+,Python2和3,PHP5.4+,Web浏览器认证交易所支持的加密货币交易所 ccxt库目前支持以下131个加密货币交易所和交易API,可以在github查看。上面的列表经常更新,新的加密市场,山寨币交换,错误修复,API端点定期引入和添加。有关详细信息,请参阅手册,如果你没有在上面的列表中找到加密货币交易市场和/或想要添加其他交易所,请通过GitHub或通过电子邮件在此处发布问题来发...
用基于 Python 的开发框架 Brownie 部署以太坊智能合约
在本文中,我们将使用Python部署智能合约。这篇文章可能是您走向智能合约和区块链开发的桥梁! 介绍 我希望可以在任何开发场景都尽量用Python。在区块链开发中,常用的是以太坊虚拟机智能合约语言Solidity,它具有许多不错的功能,并且仍然可以使用 Python 进行部署。刚开始使用Solidity时,我使用了Remix(https://remix.ethereum.org/),这是一个强大的Web IDE,可让您进行智能合约可视化。Remix很棒,我现在仍然使用它,但是在单个IDE之外可以实现很多其他功能。后来我开始学习Truffle(https://www.trufflesuite.com/)和HardHat(https://hardhat.org/guides/mainnet-forking.html),它们是用于部署智能合约的Node.js框架。 这些是到目前为止我所见过的主要框架,这些框架都不错,但是我更喜欢Python。所以当我发现Brownie 和web3.py:一个用于部署智能合约的Python框架和一个用于区块链开发的开源协议之后非常兴奋。我们将在本文中同时...
用 Python 和币安 API 构建数字货币交易机器人(一)
交易数字货币或任何其他资产的首要任务是要有目标和策略来实现。在这里,我们不是在谈论交易策略,而只是构建一个简单而功能强大的数字货币交易机器人来应用您的策略。本系列更像是一个数字货币自动交易机器人框架。 我们将使用python 3.9(3.9.2)首先创建项目文件结构。 /exchanges /strategies /models 在这里,“ exchanges”文件夹存储了Exchange API包装器,为您的策略制定策略并为我们将要使用的业务对象建模。 模型 我们将为此项目定义几个业务对象,例如价格,货币类型或订单。首先,让我们从价格开始,将其用于我们随后将要创建的策略。 在我们为业务对象编写通用抽象层之前: ./models/model.pyfrom datetime import datetime class AbstractModel: created: datetime def __init__(self, **kwargs): for key, value in kwargs.items(): setattr(self, key, value) 然后是我们的第一个Pr...
CCXT:加密货币量化交易神器
CCXT框架 是一个Python/Javascript/PHP的一个交易API框架,对接超过130多个交易所。可用于世界各地的加密货币交易所的连接和交易,以及转账支付处理,可用于存储数据,分析,可视化,指标开发,算法交易,是一个非常容易集成的开箱即用的统一API。 CCXT框架Github地址: https://github.com/ccxt/ccxt 当前功能列表:支持许多交易市场,甚至即将推出的为所有交易提供完整的公共和私人API所有货币,山寨币和标记,价格,订单,交易,代码等…提供用于交叉交换或跨货币分析和套利的可选标准化数据开箱即用的统一的一体化API,非常易于集成适用于Node7.6+,Python2和3,PHP5.4+,Web浏览器认证交易所支持的加密货币交易所 ccxt库目前支持以下131个加密货币交易所和交易API,可以在github查看。上面的列表经常更新,新的加密市场,山寨币交换,错误修复,API端点定期引入和添加。有关详细信息,请参阅手册,如果你没有在上面的列表中找到加密货币交易市场和/或想要添加其他交易所,请通过GitHub或通过电子邮件在此处发布问题来发...
用基于 Python 的开发框架 Brownie 部署以太坊智能合约
在本文中,我们将使用Python部署智能合约。这篇文章可能是您走向智能合约和区块链开发的桥梁! 介绍 我希望可以在任何开发场景都尽量用Python。在区块链开发中,常用的是以太坊虚拟机智能合约语言Solidity,它具有许多不错的功能,并且仍然可以使用 Python 进行部署。刚开始使用Solidity时,我使用了Remix(https://remix.ethereum.org/),这是一个强大的Web IDE,可让您进行智能合约可视化。Remix很棒,我现在仍然使用它,但是在单个IDE之外可以实现很多其他功能。后来我开始学习Truffle(https://www.trufflesuite.com/)和HardHat(https://hardhat.org/guides/mainnet-forking.html),它们是用于部署智能合约的Node.js框架。 这些是到目前为止我所见过的主要框架,这些框架都不错,但是我更喜欢Python。所以当我发现Brownie 和web3.py:一个用于部署智能合约的Python框架和一个用于区块链开发的开源协议之后非常兴奋。我们将在本文中同时...
用 Python 和币安 API 构建数字货币交易机器人(一)
交易数字货币或任何其他资产的首要任务是要有目标和策略来实现。在这里,我们不是在谈论交易策略,而只是构建一个简单而功能强大的数字货币交易机器人来应用您的策略。本系列更像是一个数字货币自动交易机器人框架。 我们将使用python 3.9(3.9.2)首先创建项目文件结构。 /exchanges /strategies /models 在这里,“ exchanges”文件夹存储了Exchange API包装器,为您的策略制定策略并为我们将要使用的业务对象建模。 模型 我们将为此项目定义几个业务对象,例如价格,货币类型或订单。首先,让我们从价格开始,将其用于我们随后将要创建的策略。 在我们为业务对象编写通用抽象层之前: ./models/model.pyfrom datetime import datetime class AbstractModel: created: datetime def __init__(self, **kwargs): for key, value in kwargs.items(): setattr(self, key, value) 然后是我们的第一个Pr...
专注加密货币量化交易 , focus on crypto quant
专注加密货币量化交易 , focus on crypto quant
Share Dialog
Share Dialog

Subscribe to quantbang

Subscribe to quantbang
<100 subscribers
<100 subscribers
No activity yet