用 Python 和幣安 API 構(gòu)建數(shù)字貨幣交易機(jī)器人（二）
./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)
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
./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()
評(píng)論
圖片
表情