From 8cf23cea4ba0392fad791e0d02f3f77fb58d1e82 Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Wed, 5 Jun 2019 23:33:53 +0800 Subject: [PATCH 01/16] add query_account and query_positions function , debug OK --- vnpy/gateway/alpaca/__init__.py | 1 + vnpy/gateway/alpaca/alpaca_gateway.py | 426 ++++++++++++++++++++++++++ 2 files changed, 427 insertions(+) create mode 100644 vnpy/gateway/alpaca/__init__.py create mode 100644 vnpy/gateway/alpaca/alpaca_gateway.py diff --git a/vnpy/gateway/alpaca/__init__.py b/vnpy/gateway/alpaca/__init__.py new file mode 100644 index 00000000..16a9a0b9 --- /dev/null +++ b/vnpy/gateway/alpaca/__init__.py @@ -0,0 +1 @@ +from .alpaca_gateway import AlpacaGateway diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py new file mode 100644 index 00000000..d102403b --- /dev/null +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -0,0 +1,426 @@ +# encoding: UTF-8 +""" + Author: vigarbuaa +""" + +import hashlib +import hmac +import sys +import time +from copy import copy +from threading import Lock +from datetime import datetime +from urllib.parse import urlencode +from vnpy.api.rest import Request, RestClient +from vnpy.api.websocket import WebsocketClient + +from vnpy.trader.constant import ( + Direction, + Exchange, + OrderType, + Product, + Status, + Offset, + Interval +) +from vnpy.trader.gateway import BaseGateway +from vnpy.trader.object import ( + TickData, + OrderData, + TradeData, + PositionData, AccountData, + ContractData, + OrderRequest, + CancelRequest, + SubscribeRequest, +) + +REST_HOST = "https://api.alpaca.markets" +PAPER_REST_HOST = "https://paper-api.alpaca.markets" +KEY = "" +SECRET = "" + +STATUS_ALPACA2VT = { + "new": Status.SUBMITTING, + "open": Status.NOTTRADED, + "part_filled": Status.PARTTRADED, + "filled": Status.ALLTRADED, + "cancelled": Status.CANCELLED, + "cancelling": Status.CANCELLED, + "failure": Status.REJECTED, +} + +DIRECTION_VT2ALPACA = {Direction.LONG: "buy", Direction.SHORT: "sell"} +DIRECTION_ALPACA2VT = {"buy": Direction.LONG, "sell": Direction.SHORT, + "long": Direction.LONG, "short": Direction.SHORT} + +ORDERTYPE_VT2ALPACA = { + OrderType.LIMIT: "limit", + OrderType.MARKET: "market" +} +ORDERTYPE_ALPACA2VT = {v: k for k, v in ORDERTYPE_VT2ALPACA.items()} + +class AlpacaGateway(BaseGateway): + """ + VN Trader Gateway for Alpaca connection. + """ + + default_setting = { + "key": "", + "secret": "", + "session": 3, + "proxy_host": "127.0.0.1", + "proxy_port": 1080, + } + + def __init__(self, event_engine): + """Constructor""" + super(AlpacaGateway, self).__init__(event_engine, "ALPACA") + + self.rest_api = AlpacaRestApi(self) + #self.ws_api = AlpacaWebsocketApi(self) + self.order_map = {} + + def connect(self, setting: dict): + """""" + key = setting["key"] + secret = setting["secret"] + session = setting["session"] + proxy_host = setting["proxy_host"] + proxy_port = setting["proxy_port"] + + self.rest_api.connect(key, secret, session, proxy_host, proxy_port) + + #self.ws_api.connect(key, secret, proxy_host, proxy_port) + + def subscribe(self, req: SubscribeRequest): + """""" + #self.ws_api.subscribe(req) + pass + + def send_order(self, req: OrderRequest): + """""" + return self.rest_api.send_order(req) + + def cancel_order(self, req: CancelRequest): + """""" + #self.ws_api.cancel_order(req) + pass + + def query_account(self): + """""" + self.rest_api.query_account() + + def query_position(self): + """""" + self.rest_api.query_position() + + def close(self): + """""" + self.rest_api.stop() + self.ws_api.stop() + + +class AlpacaRestApi(RestClient): + """ + Alpaca REST API + """ + + def __init__(self, gateway: BaseGateway): + """""" + super(AlpacaRestApi, self).__init__() + + self.gateway = gateway + self.gateway_name = gateway.gateway_name + + self.key = "" + self.secret = "" + + self.order_count = 1_000_000 + self.order_count_lock = Lock() + self.connect_time = 0 + print( + self.gateway_name, + self.key, + self.secret, + self.order_count, + self.connect_time) + + def query_account(self): + print("call query_account") + path = f"/v1/account" + self.add_request( + method="GET", + path=path, + callback=self.on_query_account + ) + + def on_query_account(self, data, request): + print("debug on_query_account: ", data, request) + account = AccountData( + accountid=data['id'], + balance=float(data['cash']), + frozen=float(data['cash']) - float(data['buying_power']), + gateway_name=self.gateway_name + ) + self.gateway.on_account(account) + + def query_position(self): + print("call query_position") + path = f"/v1/positions" + self.add_request( + method="GET", + path=path, + callback=self.on_query_position + ) + + def on_query_position(self, data, request): + print("debug on_query_position: ", data,"!!!", request) + for d in data: + position = PositionData( + symbol=d['symbol'], + exchange=Exchange.ALPACA, + direction=DIRECTION_ALPACA2VT[d['side']], + volume=d['qty'], + price=d['avg_entry_price'], + pnl=d['unrealized_pl'], + gateway_name=self.gateway_name, + ) + self.gateway.on_position(position) + + + + def sign(self, request): + """ + Generate Alpaca signature. + """ + #msg = request.method + \ + # "/api/v1/{}{}{}".format(path, nonce, request.data) + headers = { + "APCA-API-KEY-ID": self.key, + "APCA-API-SECRET-KEY": self.secret, + } + + request.headers = headers + request.allow_redirects=False + return request + + def _new_order_id(self): + with self.order_count_lock: + self.order_count += 1 + return self.order_count + + def connect( + self, + key: str, + secret: str, + session_num: int, + proxy_host: str, + proxy_port: int, + ): + """ + Initialize connection to REST server. + """ + self.key = key + self.secret = secret + + self.init(PAPER_REST_HOST, proxy_host, proxy_port) + print("rest connect: ", PAPER_REST_HOST, proxy_host, proxy_port) + self.start(session_num) + self.connect_time = ( + int(datetime.now().strftime("%y%m%d%H%M%S")) * self.order_count + ) + print("rest client connected",self.connect_time) + self.gateway.write_log("ALPACA REST API启动成功") + self.query_account() + self.query_position() + + def on_send_order(self, status_code: int, request: Request): + print("debug on send order: ", status_code, request) + pass + + def on_failed(self, status_code: int, request: Request): + """ + Callback to handle request failed. + """ + msg = f"请求失败，状态码：{status_code}，信息：{request.response.text}" + print('debug on_failed', msg) + self.gateway.write_log(msg) + + def on_error( + self, exception_type: type, exception_value: Exception, tb, request: Request + ): + """ + Callback to handler request exception. + """ + msg = f"触发异常，状态码：{exception_type}，信息：{exception_value}" + print('debug on_error', msg) + self.gateway.write_log(msg) + + sys.stderr.write( + self.exception_detail(exception_type, exception_value, tb, request) + ) + + def send_order(self, req: OrderRequest): + orderid = str(self.connect_time + self._new_order_id()) + data = { + "symbol": req.symbol, + "qty":2, + "side":'buy', + "type":'limit', + "time_in_force":'opg', + "limit_price":20.50 +# "side": DIRECTION_VT2BITMEX[req.direction], +# "ordType": ORDERTYPE_VT2BITMEX[req.type], +# "orderQty": int(req.volume), +# "clOrdID": orderid, + } + order = req.create_order_data(orderid, self.gateway_name) + print("debug detail order: ",order) + self.add_request( + "POST", + "/v1/orders", + callback=self.on_send_order, + data=data, + extra=order, + on_failed=self.on_failed, + on_error=self.on_error, + ) + self.gateway.on_order(order) + return order.vt_orderid + + +class AlpacaWebsocketApi(WebsocketClient): + """""" + + def __init__(self, gateway): + """""" + super(AlpacaWebsocketApi, self).__init__() + + self.gateway = gateway + self.gateway_name = gateway.gateway_name + self.order_id = 1_000_000 + # self.date = int(datetime.now().strftime('%y%m%d%H%M%S')) * self.orderId + self.key = "" + self.secret = "" + + self.callbacks = { + "trade": self.on_tick, + "orderBook10": self.on_depth, + "execution": self.on_trade, + "order": self.on_order, + "position": self.on_position, + "margin": self.on_account, + "instrument": self.on_contract, + } + + self.ticks = {} + self.accounts = {} + self.orders = {} + self.trades = set() + self.tickDict = {} + self.bidDict = {} + self.askDict = {} + self.orderLocalDict = {} + self.channelDict = {} # ChannelID : (Channel, Symbol) + + def connect( + self, key: str, secret: str, proxy_host: str, proxy_port: int + ): + """""" + self.key = key + self.secret = secret.encode() + + self.init(WEBSOCKET_HOST, proxy_host, proxy_port) + + self.start() + + def subscribe(self, req: SubscribeRequest): + pass + + def send_order(self, req: OrderRequest): + pass + + # ---------------------------------------------------------------------- + def cancel_order(self, req: CancelRequest): + """""" + pass + + def on_connected(self): + """""" + self.gateway.write_log("Websocket API连接成功") + self.authenticate() + + def on_disconnected(self): + """""" + self.gateway.write_log("Websocket API连接断开") + + # need debug 20190404 + def on_packet(self, packet: dict): + """""" + print("debug on_packet: " , packet) + + # ---------------------------------------------------------------------- + def on_response(self, data): + """""" + pass + # ---------------------------------------------------------------------- + def on_update(self, data): + """""" + pass + + # ---------------------------------------------------------------------- + def on_wallet(self, data): + """""" + pass + + # ---------------------------------------------------------------------- + def on_trade_update(self, data): + """""" + pass + + def on_error(self, exception_type: type, exception_value: Exception, tb): + """""" + print("on_error: " ,type,Exception,tb) + pass + + # debug OK , 0405 + def authenticate(self): + pass + + def subscribe_topic(self): + pass + + def on_tick(self, d): + """""" + pass + + def on_depth(self, d): + """""" + pass + + def on_trade(self, d): + """""" + pass + + def generateDateTime(self, s): + """生成时间""" + dt = datetime.fromtimestamp(s / 1000.0) + time = dt.strftime("%H:%M:%S.%f") + return time + + def on_order(self, data): + """""" + pass + + def on_position(self, d): + """""" + pass + + def on_account(self, d): + """""" + pass + + def on_contract(self, d): + """""" + pass \ No newline at end of file From 21cc72ac3f43deab545f0bbcae70f357c16aca82 Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Thu, 6 Jun 2019 23:13:39 +0800 Subject: [PATCH 02/16] add send_order function 1. add send_order function 2. post by json param, need to modify rest_client.py to support json field --- vnpy/api/rest/rest_client.py | 11 ++++++++++- vnpy/gateway/alpaca/alpaca_gateway.py | 16 ++++++---------- 2 files changed, 16 insertions(+), 11 deletions(-) diff --git a/vnpy/api/rest/rest_client.py b/vnpy/api/rest/rest_client.py index 7bd4568e..ead3992f 100644 --- a/vnpy/api/rest/rest_client.py +++ b/vnpy/api/rest/rest_client.py @@ -49,6 +49,10 @@ class Request(object): self.response = None self.status = RequestStatus.ready + self.json=None + + def set_json(self, json_str: dict): + self.json=json_str def __str__(self): if self.response is None: @@ -61,6 +65,7 @@ class Request(object): "headers: {}\n" "params: {}\n" "data: {}\n" + "json: {}\n" "response:" "{}\n".format( self.method, @@ -70,6 +75,7 @@ class Request(object): self.headers, self.params, self.data, + self.json, "" if self.response is None else self.response.text, ) ) @@ -145,6 +151,7 @@ class RestClient(object): on_failed: Callable = None, on_error: Callable = None, extra: Any = None, + json_str: dict = None, ): """ Add a new request. @@ -170,6 +177,8 @@ class RestClient(object): on_error, extra, ) + if json_str is not None: + request.set_json(json_str) self._queue.put(request) return request @@ -253,10 +262,10 @@ class RestClient(object): headers=request.headers, params=request.params, data=request.data, + json=request.json, proxies=self.proxies, ) request.response = response - status_code = response.status_code if status_code // 100 == 2: # 2xx都算成功，尽管交易所都用200 jsonBody = response.json() diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index d102403b..3d938b44 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -265,26 +265,22 @@ class AlpacaRestApi(RestClient): orderid = str(self.connect_time + self._new_order_id()) data = { "symbol": req.symbol, - "qty":2, - "side":'buy', - "type":'limit', + "qty": int(req.volume), + "side": DIRECTION_VT2ALPACA[req.direction], + "type": ORDERTYPE_VT2ALPACA[req.type], "time_in_force":'opg', - "limit_price":20.50 -# "side": DIRECTION_VT2BITMEX[req.direction], -# "ordType": ORDERTYPE_VT2BITMEX[req.type], -# "orderQty": int(req.volume), -# "clOrdID": orderid, + "limit_price": float(req.price) } order = req.create_order_data(orderid, self.gateway_name) - print("debug detail order: ",order) self.add_request( "POST", "/v1/orders", callback=self.on_send_order, - data=data, + #data=data, extra=order, on_failed=self.on_failed, on_error=self.on_error, + json_str=data, ) self.gateway.on_order(order) return order.vt_orderid From 4e30095ac37870e5f00617b9c89bab2da8dd622e Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Thu, 6 Jun 2019 23:33:17 +0800 Subject: [PATCH 03/16] modify send_order func to fill limit price order --- vnpy/gateway/alpaca/alpaca_gateway.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index 3d938b44..aa58dc98 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -263,14 +263,17 @@ class AlpacaRestApi(RestClient): def send_order(self, req: OrderRequest): orderid = str(self.connect_time + self._new_order_id()) - data = { + raw_dict={ "symbol": req.symbol, "qty": int(req.volume), "side": DIRECTION_VT2ALPACA[req.direction], "type": ORDERTYPE_VT2ALPACA[req.type], - "time_in_force":'opg', - "limit_price": float(req.price) + "time_in_force":'day', } + if raw_dict['type'] == "limit": + raw_dict['limit_price'] = float(req.price) + + data = raw_dict order = req.create_order_data(orderid, self.gateway_name) self.add_request( "POST", From 9fbac9dfba89dc39b1d387a7e3b8cb12736fa3bd Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Sat, 8 Jun 2019 06:40:09 +0800 Subject: [PATCH 04/16] format --- vnpy/gateway/alpaca/alpaca_gateway.py | 71 ++++++++++++++------------- 1 file changed, 36 insertions(+), 35 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index aa58dc98..8b2b5aa1 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -29,7 +29,7 @@ from vnpy.trader.object import ( OrderData, TradeData, PositionData, AccountData, - ContractData, + ContractData, OrderRequest, CancelRequest, SubscribeRequest, @@ -60,6 +60,7 @@ ORDERTYPE_VT2ALPACA = { } ORDERTYPE_ALPACA2VT = {v: k for k, v in ORDERTYPE_VT2ALPACA.items()} + class AlpacaGateway(BaseGateway): """ VN Trader Gateway for Alpaca connection. @@ -95,7 +96,7 @@ class AlpacaGateway(BaseGateway): def subscribe(self, req: SubscribeRequest): """""" - #self.ws_api.subscribe(req) + # self.ws_api.subscribe(req) pass def send_order(self, req: OrderRequest): @@ -104,8 +105,7 @@ class AlpacaGateway(BaseGateway): def cancel_order(self, req: CancelRequest): """""" - #self.ws_api.cancel_order(req) - pass + self.rest_api.cancel_order(req) def query_account(self): """""" @@ -139,12 +139,7 @@ class AlpacaRestApi(RestClient): self.order_count = 1_000_000 self.order_count_lock = Lock() self.connect_time = 0 - print( - self.gateway_name, - self.key, - self.secret, - self.order_count, - self.connect_time) + self.order_dict={} def query_account(self): print("call query_account") @@ -154,7 +149,7 @@ class AlpacaRestApi(RestClient): path=path, callback=self.on_query_account ) - + def on_query_account(self, data, request): print("debug on_query_account: ", data, request) account = AccountData( @@ -164,7 +159,7 @@ class AlpacaRestApi(RestClient): gateway_name=self.gateway_name ) self.gateway.on_account(account) - + def query_position(self): print("call query_position") path = f"/v1/positions" @@ -175,7 +170,7 @@ class AlpacaRestApi(RestClient): ) def on_query_position(self, data, request): - print("debug on_query_position: ", data,"!!!", request) + print("debug on_query_position: ", data, "!!!", request) for d in data: position = PositionData( symbol=d['symbol'], @@ -188,28 +183,24 @@ class AlpacaRestApi(RestClient): ) self.gateway.on_position(position) - - def sign(self, request): """ Generate Alpaca signature. """ - #msg = request.method + \ - # "/api/v1/{}{}{}".format(path, nonce, request.data) headers = { "APCA-API-KEY-ID": self.key, "APCA-API-SECRET-KEY": self.secret, } request.headers = headers - request.allow_redirects=False + request.allow_redirects = False return request def _new_order_id(self): with self.order_count_lock: self.order_count += 1 return self.order_count - + def connect( self, key: str, @@ -230,29 +221,36 @@ class AlpacaRestApi(RestClient): self.connect_time = ( int(datetime.now().strftime("%y%m%d%H%M%S")) * self.order_count ) - print("rest client connected",self.connect_time) + print("rest client connected", self.connect_time) self.gateway.write_log("ALPACA REST API启动成功") self.query_account() self.query_position() def on_send_order(self, status_code: int, request: Request): print("debug on send order: ", status_code, request) - pass + order = request.extra + self.gateway.on_order(order) - def on_failed(self, status_code: int, request: Request): + def on_failed_order(self, status_code: int, request: Request): """ Callback to handle request failed. """ + order = request.extra + order.status = Status.REJECTED + self.gateway.on_order(order) msg = f"请求失败，状态码：{status_code}，信息：{request.response.text}" print('debug on_failed', msg) self.gateway.write_log(msg) - def on_error( + def on_error_order( self, exception_type: type, exception_value: Exception, tb, request: Request ): """ Callback to handler request exception. """ + order = request.extra + order.status = Status.REJECTED + self.gateway.on_order(order) msg = f"触发异常，状态码：{exception_type}，信息：{exception_value}" print('debug on_error', msg) self.gateway.write_log(msg) @@ -261,14 +259,19 @@ class AlpacaRestApi(RestClient): self.exception_detail(exception_type, exception_value, tb, request) ) + def cancel_order(self, req: CancelRequest): + """""" + order_id = req.orderid + print("come to cancel_order", order_id) + def send_order(self, req: OrderRequest): orderid = str(self.connect_time + self._new_order_id()) - raw_dict={ + raw_dict = { "symbol": req.symbol, "qty": int(req.volume), "side": DIRECTION_VT2ALPACA[req.direction], "type": ORDERTYPE_VT2ALPACA[req.type], - "time_in_force":'day', + "time_in_force": 'day', } if raw_dict['type'] == "limit": raw_dict['limit_price'] = float(req.price) @@ -279,15 +282,14 @@ class AlpacaRestApi(RestClient): "POST", "/v1/orders", callback=self.on_send_order, - #data=data, + # data=data, extra=order, - on_failed=self.on_failed, - on_error=self.on_error, + on_failed=self.on_failed_order, + on_error=self.on_error_order, json_str=data, ) - self.gateway.on_order(order) return order.vt_orderid - + class AlpacaWebsocketApi(WebsocketClient): """""" @@ -329,9 +331,7 @@ class AlpacaWebsocketApi(WebsocketClient): """""" self.key = key self.secret = secret.encode() - self.init(WEBSOCKET_HOST, proxy_host, proxy_port) - self.start() def subscribe(self, req: SubscribeRequest): @@ -357,13 +357,14 @@ class AlpacaWebsocketApi(WebsocketClient): # need debug 20190404 def on_packet(self, packet: dict): """""" - print("debug on_packet: " , packet) + print("debug on_packet: ", packet) # ---------------------------------------------------------------------- def on_response(self, data): """""" pass # ---------------------------------------------------------------------- + def on_update(self, data): """""" pass @@ -380,7 +381,7 @@ class AlpacaWebsocketApi(WebsocketClient): def on_error(self, exception_type: type, exception_value: Exception, tb): """""" - print("on_error: " ,type,Exception,tb) + print("on_error: ", type, Exception, tb) pass # debug OK , 0405 @@ -422,4 +423,4 @@ class AlpacaWebsocketApi(WebsocketClient): def on_contract(self, d): """""" - pass \ No newline at end of file + pass From 5e5137f9abac82e3826703e0c9fb5bc4b4b93f84 Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Sat, 8 Jun 2019 07:17:25 +0800 Subject: [PATCH 05/16] add order_dict for cancel_order function order_dict {local_order_id: remote_order_id} --- vnpy/gateway/alpaca/alpaca_gateway.py | 28 +++++++++++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index 8b2b5aa1..e82caf4f 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -226,9 +226,13 @@ class AlpacaRestApi(RestClient): self.query_account() self.query_position() - def on_send_order(self, status_code: int, request: Request): - print("debug on send order: ", status_code, request) + def on_send_order(self, data, request: Request): + print("debug on send order: ", data, request) order = request.extra + lcoal_order_id= order.orderId + remote_order_id = data['id'] + self.order_dict[local_order_id]=remote_order_id + print('debug on_send_order: ', local_order_id, remote_order_id) self.gateway.on_order(order) def on_failed_order(self, status_code: int, request: Request): @@ -259,11 +263,31 @@ class AlpacaRestApi(RestClient): self.exception_detail(exception_type, exception_value, tb, request) ) + # need debug 0608 def cancel_order(self, req: CancelRequest): """""" order_id = req.orderid + remote_order_id = self.order_dict['order_id'] + if remote_order_id is None: + print("[error]: can not get remote_order_id from local dict!") + return + path="/v1/orders/"+str(remote_order_id) + self.add_request( + "DELETE", + path, + callback=self.on_cancel_order, + on_error=self.on_cancel_order_error, + extra=req + ) print("come to cancel_order", order_id) + def on_cancel_order(self, data, request): + """Websocket will push a new order status""" + pass + + def on_cancel_order_error(self, data, request): + pass + def send_order(self, req: OrderRequest): orderid = str(self.connect_time + self._new_order_id()) raw_dict = { From 7be2a90e4c76e1cca0864cc3b770ac184b5b06f2 Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Sat, 15 Jun 2019 11:19:15 +0800 Subject: [PATCH 06/16] add websocket into alpaca --- vnpy/gateway/alpaca/alpaca_gateway.py | 97 +++++++++++++++++++++------ 1 file changed, 77 insertions(+), 20 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index e82caf4f..528457e5 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -37,6 +37,7 @@ from vnpy.trader.object import ( REST_HOST = "https://api.alpaca.markets" PAPER_REST_HOST = "https://paper-api.alpaca.markets" +WEBSOCKET_HOST = "wss://paper-api.alpaca.markets/stream" # Market Data KEY = "" SECRET = "" @@ -79,7 +80,7 @@ class AlpacaGateway(BaseGateway): super(AlpacaGateway, self).__init__(event_engine, "ALPACA") self.rest_api = AlpacaRestApi(self) - #self.ws_api = AlpacaWebsocketApi(self) + self.ws_api = AlpacaWebsocketApi(self) self.order_map = {} def connect(self, setting: dict): @@ -92,7 +93,7 @@ class AlpacaGateway(BaseGateway): self.rest_api.connect(key, secret, session, proxy_host, proxy_port) - #self.ws_api.connect(key, secret, proxy_host, proxy_port) + self.ws_api.connect(key, secret, proxy_host, proxy_port) def subscribe(self, req: SubscribeRequest): """""" @@ -139,7 +140,7 @@ class AlpacaRestApi(RestClient): self.order_count = 1_000_000 self.order_count_lock = Lock() self.connect_time = 0 - self.order_dict={} + self.order_list = [] def query_account(self): print("call query_account") @@ -151,7 +152,6 @@ class AlpacaRestApi(RestClient): ) def on_query_account(self, data, request): - print("debug on_query_account: ", data, request) account = AccountData( accountid=data['id'], balance=float(data['cash']), @@ -170,7 +170,6 @@ class AlpacaRestApi(RestClient): ) def on_query_position(self, data, request): - print("debug on_query_position: ", data, "!!!", request) for d in data: position = PositionData( symbol=d['symbol'], @@ -225,14 +224,14 @@ class AlpacaRestApi(RestClient): self.gateway.write_log("ALPACA REST API启动成功") self.query_account() self.query_position() + #self.query_contracts() - def on_send_order(self, data, request: Request): - print("debug on send order: ", data, request) - order = request.extra - lcoal_order_id= order.orderId + def on_send_order(self, data, request ): + print("debug on_send_order data is {} request is {} ---".format( data, request)) remote_order_id = data['id'] - self.order_dict[local_order_id]=remote_order_id - print('debug on_send_order: ', local_order_id, remote_order_id) + order = request.extra + self.order_list.append(remote_order_id) + print('debug on_send_order: local is {} ---'.format(self.order_list)) self.gateway.on_order(order) def on_failed_order(self, status_code: int, request: Request): @@ -266,12 +265,13 @@ class AlpacaRestApi(RestClient): # need debug 0608 def cancel_order(self, req: CancelRequest): """""" + """ order_id = req.orderid remote_order_id = self.order_dict['order_id'] if remote_order_id is None: print("[error]: can not get remote_order_id from local dict!") return - path="/v1/orders/"+str(remote_order_id) + path = "/v1/orders/" + str(remote_order_id) self.add_request( "DELETE", path, @@ -280,11 +280,13 @@ class AlpacaRestApi(RestClient): extra=req ) print("come to cancel_order", order_id) +""" + pass def on_cancel_order(self, data, request): """Websocket will push a new order status""" pass - + def on_cancel_order_error(self, data, request): pass @@ -314,6 +316,38 @@ class AlpacaRestApi(RestClient): ) return order.vt_orderid + def on_query_contracts(self, data, request: Request): + for instrument_data in data: + symbol = instrument_data['symbol'] + contract = ContractData( + symbol=symbol, + exchange=Exchange.ALPACA, + name=symbol, + product=Product.SPOT, + size=1, # need debug + pricetick=0.01, # need debug + gateway_name=self.gateway_name + ) + self.on_contract(contract) + + def on_failed_query_contracts(self, status_code: int, request: Request): + pass + + def on_error_query_contracts(self, exception_type: type, exception_value: Exception, tb, request: Request): + pass + + def query_contracts(self): + params = {"status": "active"} + self.add_request( + "GET", + "/v1/assets", + params = params, + callback=self.on_query_contracts, + on_failed=self.on_failed_query_contracts, + on_error=self.on_error_query_contracts, + # data=data, + ) + class AlpacaWebsocketApi(WebsocketClient): """""" @@ -354,10 +388,26 @@ class AlpacaWebsocketApi(WebsocketClient): ): """""" self.key = key - self.secret = secret.encode() + self.secret = secret self.init(WEBSOCKET_HOST, proxy_host, proxy_port) self.start() + def login(self): + """""" + params={"action":"authenticate", "data": { + "key_id":self.key,"secret_key":self.secret + }} + print("string is {} ===".format(params)) + self.send_packet(params) + + def on_login(self, packet): + """""" + print("websocket authenticate success!!! msg: ", packet) + params={"action":"listen", "data": { + "streams":["account_updates", "trade_updates"] + }} + self.send_packet(params) + def subscribe(self, req: SubscribeRequest): pass @@ -372,7 +422,7 @@ class AlpacaWebsocketApi(WebsocketClient): def on_connected(self): """""" self.gateway.write_log("Websocket API连接成功") - self.authenticate() + self.login() def on_disconnected(self): """""" @@ -382,6 +432,17 @@ class AlpacaWebsocketApi(WebsocketClient): def on_packet(self, packet: dict): """""" print("debug on_packet: ", packet) + if "ping" in packet: + # has ping??? + req = {"pong": packet["ping"]} + print("ping response: ", req) + self.send_packet(req) + elif "stream" in packet and "data" in packet: + # on_data + pass + else: + print("unrecognize mesg", packet) + #parse authenticate/trade/other data here # ---------------------------------------------------------------------- def on_response(self, data): @@ -408,10 +469,6 @@ class AlpacaWebsocketApi(WebsocketClient): print("on_error: ", type, Exception, tb) pass - # debug OK , 0405 - def authenticate(self): - pass - def subscribe_topic(self): pass @@ -447,4 +504,4 @@ class AlpacaWebsocketApi(WebsocketClient): def on_contract(self, d): """""" - pass + pass \ No newline at end of file From e8e2e1121d91bac59c710bf390d2481dc3b73d63 Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Sat, 15 Jun 2019 13:45:50 +0800 Subject: [PATCH 07/16] websocket stream and auth and listening stream --- vnpy/gateway/alpaca/alpaca_gateway.py | 53 +++++++++++++++++++-------- 1 file changed, 37 insertions(+), 16 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index 528457e5..a033a384 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -392,7 +392,7 @@ class AlpacaWebsocketApi(WebsocketClient): self.init(WEBSOCKET_HOST, proxy_host, proxy_port) self.start() - def login(self): + def authenticate(self): """""" params={"action":"authenticate", "data": { "key_id":self.key,"secret_key":self.secret @@ -400,12 +400,13 @@ class AlpacaWebsocketApi(WebsocketClient): print("string is {} ===".format(params)) self.send_packet(params) - def on_login(self, packet): + def on_authenticate(self): """""" - print("websocket authenticate success!!! msg: ", packet) + print("websocket authenticate success!!! msg: " ) params={"action":"listen", "data": { "streams":["account_updates", "trade_updates"] }} + print("on_authenticate func", params) self.send_packet(params) def subscribe(self, req: SubscribeRequest): @@ -422,27 +423,45 @@ class AlpacaWebsocketApi(WebsocketClient): def on_connected(self): """""" self.gateway.write_log("Websocket API连接成功") - self.login() + self.authenticate() def on_disconnected(self): """""" self.gateway.write_log("Websocket API连接断开") - # need debug 20190404 def on_packet(self, packet: dict): """""" print("debug on_packet: ", packet) - if "ping" in packet: - # has ping??? - req = {"pong": packet["ping"]} - print("ping response: ", req) - self.send_packet(req) - elif "stream" in packet and "data" in packet: - # on_data - pass + if "stream" in packet and "data" in packet: + stream_ret = packet['stream'] + data_ret = packet['data'] + if(stream_ret == "authorization"): + self.handle_auth(packet) + elif(stream_ret == "listening"): + self.gateway.write_log("listening {}".format(data_ret)) + else: + self.ondata(packet) else: - print("unrecognize mesg", packet) - #parse authenticate/trade/other data here + print("unrecognize msg", packet) + + # ---------------------------------------------------------------------- + def on_data(self, data): + pass + + # ---------------------------------------------------------------------- + def handle_auth(self, data): + stream_ret = data['stream'] + data_ret = data['data'] + print("stream is {}, data is {}".format(stream_ret,data_ret)) + if (data_ret['status'] == "authorized"): + print("authorization success!!!") + self.gateway.write_log("authorization success!!!") + self.on_authenticate() + elif (data_ret['status'] == "unauthorized"): + print("authorization failed!!!") + self.gateway.write_log("authorization failed!!!") + else: + print("??unhandled status: ",data) # ---------------------------------------------------------------------- def on_response(self, data): @@ -467,7 +486,9 @@ class AlpacaWebsocketApi(WebsocketClient): def on_error(self, exception_type: type, exception_value: Exception, tb): """""" print("on_error: ", type, Exception, tb) - pass + sys.stderr.write( + self.exception_detail(exception_type, exception_value, tb ) + ) def subscribe_topic(self): pass From 1e0248ce5cc858e8a44e93d5b28a242155d938de Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Sat, 15 Jun 2019 15:07:07 +0800 Subject: [PATCH 08/16] add swicht {real, paper} url function --- vnpy/gateway/alpaca/alpaca_gateway.py | 69 +++++++++++++++++++-------- 1 file changed, 49 insertions(+), 20 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index a033a384..20bc75f6 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -36,19 +36,19 @@ from vnpy.trader.object import ( ) REST_HOST = "https://api.alpaca.markets" +WEBSOCKET_HOST = "wss://api.alpaca.markets/stream" # Market Data PAPER_REST_HOST = "https://paper-api.alpaca.markets" -WEBSOCKET_HOST = "wss://paper-api.alpaca.markets/stream" # Market Data +PAPER_WEBSOCKET_HOST = "wss://paper-api.alpaca.markets/stream" # Market Data KEY = "" SECRET = "" STATUS_ALPACA2VT = { "new": Status.SUBMITTING, - "open": Status.NOTTRADED, - "part_filled": Status.PARTTRADED, - "filled": Status.ALLTRADED, + "partial_fill": Status.PARTTRADED, + "fill": Status.ALLTRADED, "cancelled": Status.CANCELLED, - "cancelling": Status.CANCELLED, - "failure": Status.REJECTED, + #"done_for_day": Status.CANCELLED, + "expired": Status.NOTTRADED } DIRECTION_VT2ALPACA = {Direction.LONG: "buy", Direction.SHORT: "sell"} @@ -71,6 +71,7 @@ class AlpacaGateway(BaseGateway): "key": "", "secret": "", "session": 3, + "服务器": ["REAL", "PAPER"], "proxy_host": "127.0.0.1", "proxy_port": 1080, } @@ -85,15 +86,17 @@ class AlpacaGateway(BaseGateway): def connect(self, setting: dict): """""" + print("[debug] gateway setting: ",setting) key = setting["key"] secret = setting["secret"] session = setting["session"] proxy_host = setting["proxy_host"] proxy_port = setting["proxy_port"] - - self.rest_api.connect(key, secret, session, proxy_host, proxy_port) - - self.ws_api.connect(key, secret, proxy_host, proxy_port) + env=setting['服务器'] + rest_url = REST_HOST if env == "REAL" else PAPER_REST_HOST + websocket_url = WEBSOCKET_HOST if env == "REAL" else PAPER_WEBSOCKET_HOST + self.rest_api.connect(key, secret, session, proxy_host, proxy_port,rest_url) + self.ws_api.connect(key, secret, proxy_host, proxy_port,websocket_url) def subscribe(self, req: SubscribeRequest): """""" @@ -152,6 +155,7 @@ class AlpacaRestApi(RestClient): ) def on_query_account(self, data, request): + print("on_query_account debug: " , data) account = AccountData( accountid=data['id'], balance=float(data['cash']), @@ -207,15 +211,15 @@ class AlpacaRestApi(RestClient): session_num: int, proxy_host: str, proxy_port: int, + url: str, ): """ Initialize connection to REST server. """ self.key = key self.secret = secret - - self.init(PAPER_REST_HOST, proxy_host, proxy_port) - print("rest connect: ", PAPER_REST_HOST, proxy_host, proxy_port) + self.init(url, proxy_host, proxy_port) + print("rest connect: ", url, proxy_host, proxy_port) self.start(session_num) self.connect_time = ( int(datetime.now().strftime("%y%m%d%H%M%S")) * self.order_count @@ -227,7 +231,6 @@ class AlpacaRestApi(RestClient): #self.query_contracts() def on_send_order(self, data, request ): - print("debug on_send_order data is {} request is {} ---".format( data, request)) remote_order_id = data['id'] order = request.extra self.order_list.append(remote_order_id) @@ -384,12 +387,12 @@ class AlpacaWebsocketApi(WebsocketClient): self.channelDict = {} # ChannelID : (Channel, Symbol) def connect( - self, key: str, secret: str, proxy_host: str, proxy_port: int + self, key: str, secret: str, proxy_host: str, proxy_port: int,url:str ): """""" self.key = key self.secret = secret - self.init(WEBSOCKET_HOST, proxy_host, proxy_port) + self.init(url, proxy_host, proxy_port) self.start() def authenticate(self): @@ -397,16 +400,13 @@ class AlpacaWebsocketApi(WebsocketClient): params={"action":"authenticate", "data": { "key_id":self.key,"secret_key":self.secret }} - print("string is {} ===".format(params)) self.send_packet(params) def on_authenticate(self): """""" - print("websocket authenticate success!!! msg: " ) params={"action":"listen", "data": { "streams":["account_updates", "trade_updates"] }} - print("on_authenticate func", params) self.send_packet(params) def subscribe(self, req: SubscribeRequest): @@ -446,7 +446,36 @@ class AlpacaWebsocketApi(WebsocketClient): # ---------------------------------------------------------------------- def on_data(self, data): - pass + print("on_data is {}".format(data)) + stream_ret = data['stream'] + data_ret = data['data'] + if(stream_ret == "account_updates"): + #handle account + account = AccountData( + accountid=data_ret['id'], + balance=float(data_ret['cash']), + frozen=float(data_ret['cash']) - float(data_ret['cash_withdrawable']), + gateway_name=self.gateway_name + ) + self.gateway.on_account(account) + elif(stream_ret == "trade_updates"): + d=data_ret['order'] + trade = TradeData( + symbol=d["symbol"], + exchange=Exchange.ALPACA, + orderid=d['id'], + tradeid=None, + direction=DIRECTION_ALPACA2VT[d["side"]], + price=data_ret["price"], + volume=data_ret["qty"], + time=data_ret["timestamp"][11:19], + gateway_name=self.gateway_name, + ) + self.gateway.on_trade(trade) + #self.gateway.on_order(order) # udpate order status + else: + pass + # ---------------------------------------------------------------------- def handle_auth(self, data): From eb691525fc78cd00408cb66bea06df440a60f237 Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Tue, 18 Jun 2019 23:13:41 +0800 Subject: [PATCH 09/16] cancel order successful send cancel req by rest_client receive reponse by websocket ... need to debug --- vnpy/gateway/alpaca/alpaca_gateway.py | 50 ++++++++++++++++----------- 1 file changed, 29 insertions(+), 21 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index 20bc75f6..02d8709b 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -143,7 +143,7 @@ class AlpacaRestApi(RestClient): self.order_count = 1_000_000 self.order_count_lock = Lock() self.connect_time = 0 - self.order_list = [] + self.order_dict ={} def query_account(self): print("call query_account") @@ -230,11 +230,14 @@ class AlpacaRestApi(RestClient): self.query_position() #self.query_contracts() - def on_send_order(self, data, request ): + def on_send_order(self, data, request: Request ): + print("debug on_send_order data: ", data) + print("debug on_send_order request: ", request) + print("***debug on_send_order request: ", request.extra) remote_order_id = data['id'] order = request.extra - self.order_list.append(remote_order_id) - print('debug on_send_order: local is {} ---'.format(self.order_list)) + self.order_dict[order.orderid]=remote_order_id + print("+++debug on_send_order request: ", self.order_dict) self.gateway.on_order(order) def on_failed_order(self, status_code: int, request: Request): @@ -268,9 +271,9 @@ class AlpacaRestApi(RestClient): # need debug 0608 def cancel_order(self, req: CancelRequest): """""" - """ order_id = req.orderid - remote_order_id = self.order_dict['order_id'] + remote_order_id = self.order_dict[order_id] + print("debug cancel order: order id ", order_id, "---", remote_order_id) if remote_order_id is None: print("[error]: can not get remote_order_id from local dict!") return @@ -283,7 +286,6 @@ class AlpacaRestApi(RestClient): extra=req ) print("come to cancel_order", order_id) -""" pass def on_cancel_order(self, data, request): @@ -307,6 +309,7 @@ class AlpacaRestApi(RestClient): data = raw_dict order = req.create_order_data(orderid, self.gateway_name) + print("debug send_order orderBody extra: ",order) self.add_request( "POST", "/v1/orders", @@ -440,7 +443,7 @@ class AlpacaWebsocketApi(WebsocketClient): elif(stream_ret == "listening"): self.gateway.write_log("listening {}".format(data_ret)) else: - self.ondata(packet) + self.on_data(packet) else: print("unrecognize msg", packet) @@ -460,23 +463,28 @@ class AlpacaWebsocketApi(WebsocketClient): self.gateway.on_account(account) elif(stream_ret == "trade_updates"): d=data_ret['order'] - trade = TradeData( - symbol=d["symbol"], - exchange=Exchange.ALPACA, - orderid=d['id'], - tradeid=None, - direction=DIRECTION_ALPACA2VT[d["side"]], - price=data_ret["price"], - volume=data_ret["qty"], - time=data_ret["timestamp"][11:19], - gateway_name=self.gateway_name, - ) - self.gateway.on_trade(trade) + if (data_ret['event'] == "fill"): + trade = TradeData( + symbol=d["symbol"], + exchange=Exchange.ALPACA, + orderid=d['id'], + tradeid=None, + direction=DIRECTION_ALPACA2VT[d["side"]], + price=d["filled_avg_price"], + volume=d["filled_qty"], + time=data_ret["timestamp"][11:19], + gateway_name=self.gateway_name, + ) + self.gateway.on_trade(trade) + elif (data_ret['event'] == "canceled"): + + self.gateway.on_order(order) + else: + print("unhandled trade_update msg, ", data_ret['event']) #self.gateway.on_order(order) # udpate order status else: pass - # ---------------------------------------------------------------------- def handle_auth(self, data): stream_ret = data['stream'] From 04b0da5ce53a825126f132bfcefa7bf817eeb43f Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Thu, 20 Jun 2019 22:24:03 +0800 Subject: [PATCH 10/16] add try/catch to json parse func MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit rest_client认为每个返回都可解为json,在处理alpaca的撤单操作时报错 --- vnpy/api/rest/rest_client.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/vnpy/api/rest/rest_client.py b/vnpy/api/rest/rest_client.py index ead3992f..1af29631 100644 --- a/vnpy/api/rest/rest_client.py +++ b/vnpy/api/rest/rest_client.py @@ -268,9 +268,12 @@ class RestClient(object): request.response = response status_code = response.status_code if status_code // 100 == 2: # 2xx都算成功，尽管交易所都用200 - jsonBody = response.json() - request.callback(jsonBody, request) - request.status = RequestStatus.success + try: + jsonBody = response.json() + request.callback(jsonBody, request) + request.status = RequestStatus.success + except: + print("not json body from request, check body: ",response.text) else: request.status = RequestStatus.failed From 0753e3cfe2689fab49972c025ae05c75228f7a2a Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Thu, 20 Jun 2019 22:24:18 +0800 Subject: [PATCH 11/16] add cancel order fun for alpaca --- vnpy/gateway/alpaca/alpaca_gateway.py | 20 ++++++++++++++------ 1 file changed, 14 insertions(+), 6 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index 02d8709b..95c5e28a 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -60,7 +60,7 @@ ORDERTYPE_VT2ALPACA = { OrderType.MARKET: "market" } ORDERTYPE_ALPACA2VT = {v: k for k, v in ORDERTYPE_VT2ALPACA.items()} - +GLOBAL_ORDER={} class AlpacaGateway(BaseGateway): """ @@ -239,6 +239,8 @@ class AlpacaRestApi(RestClient): self.order_dict[order.orderid]=remote_order_id print("+++debug on_send_order request: ", self.order_dict) self.gateway.on_order(order) + GLOBAL_ORDER[remote_order_id]=order + print("===+++ debug update global_order_dict: ", GLOBAL_ORDER) def on_failed_order(self, status_code: int, request: Request): """ @@ -286,14 +288,15 @@ class AlpacaRestApi(RestClient): extra=req ) print("come to cancel_order", order_id) - pass def on_cancel_order(self, data, request): """Websocket will push a new order status""" pass - def on_cancel_order_error(self, data, request): - pass + def on_cancel_order_error( self, exception_type: type, exception_value: Exception, tb, request: Request): + # Record exception if not ConnectionError + if not issubclass(exception_type, ConnectionError): + self.on_error(exception_type, exception_value, tb, request) def send_order(self, req: OrderRequest): orderid = str(self.connect_time + self._new_order_id()) @@ -320,6 +323,7 @@ class AlpacaRestApi(RestClient): on_error=self.on_error_order, json_str=data, ) + print("debug send_order ret val : ", order.vt_orderid) return order.vt_orderid def on_query_contracts(self, data, request: Request): @@ -341,7 +345,8 @@ class AlpacaRestApi(RestClient): def on_error_query_contracts(self, exception_type: type, exception_value: Exception, tb, request: Request): pass - + + # need debug def query_contracts(self): params = {"status": "active"} self.add_request( @@ -477,8 +482,11 @@ class AlpacaWebsocketApi(WebsocketClient): ) self.gateway.on_trade(trade) elif (data_ret['event'] == "canceled"): - + order_id = d['id'] + order=GLOBAL_ORDER[order_id] + order.status = Status.CANCELLED self.gateway.on_order(order) + print("^^^^debug cancel order id, ",order_id," body: ",order) else: print("unhandled trade_update msg, ", data_ret['event']) #self.gateway.on_order(order) # udpate order status From 2dd33b5ba04326dbbe421adc72b02180e6be760c Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Fri, 21 Jun 2019 13:26:40 +0800 Subject: [PATCH 12/16] add subscribe func --- vnpy/gateway/alpaca/alpaca_gateway.py | 21 ++++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index 95c5e28a..08bb6fb7 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -36,9 +36,9 @@ from vnpy.trader.object import ( ) REST_HOST = "https://api.alpaca.markets" -WEBSOCKET_HOST = "wss://api.alpaca.markets/stream" # Market Data +WEBSOCKET_HOST = "wss://api.alpaca.markets/stream" # Market Data PAPER_REST_HOST = "https://paper-api.alpaca.markets" -PAPER_WEBSOCKET_HOST = "wss://paper-api.alpaca.markets/stream" # Market Data +PAPER_WEBSOCKET_HOST = "wss://paper-api.alpaca.markets/stream" # Market Data KEY = "" SECRET = "" @@ -100,7 +100,7 @@ class AlpacaGateway(BaseGateway): def subscribe(self, req: SubscribeRequest): """""" - # self.ws_api.subscribe(req) + self.ws_api.subscribe(req) pass def send_order(self, req: OrderRequest): @@ -393,6 +393,7 @@ class AlpacaWebsocketApi(WebsocketClient): self.askDict = {} self.orderLocalDict = {} self.channelDict = {} # ChannelID : (Channel, Symbol) + self.channels=["account_updates", "trade_updates"] def connect( self, key: str, secret: str, proxy_host: str, proxy_port: int,url:str @@ -413,12 +414,16 @@ class AlpacaWebsocketApi(WebsocketClient): def on_authenticate(self): """""" params={"action":"listen", "data": { - "streams":["account_updates", "trade_updates"] + "streams":self.channels }} self.send_packet(params) def subscribe(self, req: SubscribeRequest): - pass + self.channels.append(req.symbol) + params={"action":"listen", "data": { + "streams":self.channels + }} + self.send_packet(params) def send_order(self, req: OrderRequest): pass @@ -468,6 +473,8 @@ class AlpacaWebsocketApi(WebsocketClient): self.gateway.on_account(account) elif(stream_ret == "trade_updates"): d=data_ret['order'] + order_id = d['id'] + order=GLOBAL_ORDER[order_id] if (data_ret['event'] == "fill"): trade = TradeData( symbol=d["symbol"], @@ -481,9 +488,9 @@ class AlpacaWebsocketApi(WebsocketClient): gateway_name=self.gateway_name, ) self.gateway.on_trade(trade) + order.status = Status.ALLTRADED + self.gateway.on_order(order) elif (data_ret['event'] == "canceled"): - order_id = d['id'] - order=GLOBAL_ORDER[order_id] order.status = Status.CANCELLED self.gateway.on_order(order) print("^^^^debug cancel order id, ",order_id," body: ",order) From 9124302ccad9147595c6f3770a5168a888354893 Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Fri, 21 Jun 2019 22:31:30 +0800 Subject: [PATCH 13/16] add process_timer_event to query account and position --- vnpy/gateway/alpaca/alpaca_gateway.py | 63 +++++++++++++++++---------- 1 file changed, 40 insertions(+), 23 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index 08bb6fb7..21e26b2b 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -13,6 +13,8 @@ from datetime import datetime from urllib.parse import urlencode from vnpy.api.rest import Request, RestClient from vnpy.api.websocket import WebsocketClient +from vnpy.event import Event +from vnpy.trader.event import EVENT_TIMER from vnpy.trader.constant import ( Direction, @@ -97,11 +99,11 @@ class AlpacaGateway(BaseGateway): websocket_url = WEBSOCKET_HOST if env == "REAL" else PAPER_WEBSOCKET_HOST self.rest_api.connect(key, secret, session, proxy_host, proxy_port,rest_url) self.ws_api.connect(key, secret, proxy_host, proxy_port,websocket_url) + self.init_query() def subscribe(self, req: SubscribeRequest): """""" self.ws_api.subscribe(req) - pass def send_order(self, req: OrderRequest): """""" @@ -123,6 +125,32 @@ class AlpacaGateway(BaseGateway): """""" self.rest_api.stop() self.ws_api.stop() + + def process_timer_event(self, event: Event): + """""" + self.count += 1 + if self.count < 5: + return + + self.query_account() + self.query_position() + + def init_query(self): + """""" + self.count = 0 + self.event_engine.register(EVENT_TIMER, self.process_timer_event) + + + def process_timer_event(self, event: Event): + """""" + self.count += 1 + if self.count < 5: + return + else: + self.count = 0 + + self.query_account() + self.query_position() class AlpacaRestApi(RestClient): @@ -419,7 +447,9 @@ class AlpacaWebsocketApi(WebsocketClient): self.send_packet(params) def subscribe(self, req: SubscribeRequest): - self.channels.append(req.symbol) + self.channels.append("A."+req.symbol) + self.channels.append("Q."+req.symbol) + print("debug subscribe: {} ".format(self.channels)) params={"action":"listen", "data": { "streams":self.channels }} @@ -435,7 +465,7 @@ class AlpacaWebsocketApi(WebsocketClient): def on_connected(self): """""" - self.gateway.write_log("Websocket API连接成功") + self.gateway.write_log("Websocket API连接成功") self.authenticate() def on_disconnected(self): @@ -464,6 +494,7 @@ class AlpacaWebsocketApi(WebsocketClient): data_ret = data['data'] if(stream_ret == "account_updates"): #handle account + print("!!!!!!!@@@@@@@@ account_udpate is {}".format(data)) account = AccountData( accountid=data_ret['id'], balance=float(data_ret['cash']), @@ -475,12 +506,15 @@ class AlpacaWebsocketApi(WebsocketClient): d=data_ret['order'] order_id = d['id'] order=GLOBAL_ORDER[order_id] - if (data_ret['event'] == "fill"): + local_order_id = order.orderid + order.status = STATUS_ALPACA2VT[data_ret['event']] + self.gateway.on_order(order) + if (data_ret['event'] == "fill" or data_ret['event'] == "partial_fill"): trade = TradeData( symbol=d["symbol"], exchange=Exchange.ALPACA, - orderid=d['id'], - tradeid=None, + orderid=local_order_id, + tradeid=d['id'], direction=DIRECTION_ALPACA2VT[d["side"]], price=d["filled_avg_price"], volume=d["filled_qty"], @@ -488,15 +522,6 @@ class AlpacaWebsocketApi(WebsocketClient): gateway_name=self.gateway_name, ) self.gateway.on_trade(trade) - order.status = Status.ALLTRADED - self.gateway.on_order(order) - elif (data_ret['event'] == "canceled"): - order.status = Status.CANCELLED - self.gateway.on_order(order) - print("^^^^debug cancel order id, ",order_id," body: ",order) - else: - print("unhandled trade_update msg, ", data_ret['event']) - #self.gateway.on_order(order) # udpate order status else: pass @@ -526,14 +551,6 @@ class AlpacaWebsocketApi(WebsocketClient): pass # ---------------------------------------------------------------------- - def on_wallet(self, data): - """""" - pass - - # ---------------------------------------------------------------------- - def on_trade_update(self, data): - """""" - pass def on_error(self, exception_type: type, exception_value: Exception, tb): """""" From 08d3b26a5d5060a0b2a4f5c43ef23e799d3b3368 Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Sat, 22 Jun 2019 10:22:04 +0800 Subject: [PATCH 14/16] remove debug print info --- vnpy/gateway/alpaca/alpaca_gateway.py | 28 ++++++--------------------- 1 file changed, 6 insertions(+), 22 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index 21e26b2b..a34e8529 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -88,7 +88,6 @@ class AlpacaGateway(BaseGateway): def connect(self, setting: dict): """""" - print("[debug] gateway setting: ",setting) key = setting["key"] secret = setting["secret"] session = setting["session"] @@ -174,7 +173,6 @@ class AlpacaRestApi(RestClient): self.order_dict ={} def query_account(self): - print("call query_account") path = f"/v1/account" self.add_request( method="GET", @@ -183,7 +181,6 @@ class AlpacaRestApi(RestClient): ) def on_query_account(self, data, request): - print("on_query_account debug: " , data) account = AccountData( accountid=data['id'], balance=float(data['cash']), @@ -193,7 +190,6 @@ class AlpacaRestApi(RestClient): self.gateway.on_account(account) def query_position(self): - print("call query_position") path = f"/v1/positions" self.add_request( method="GET", @@ -208,7 +204,7 @@ class AlpacaRestApi(RestClient): exchange=Exchange.ALPACA, direction=DIRECTION_ALPACA2VT[d['side']], volume=d['qty'], - price=d['avg_entry_price'], + price=round(d['avg_entry_price'],3), pnl=d['unrealized_pl'], gateway_name=self.gateway_name, ) @@ -221,6 +217,7 @@ class AlpacaRestApi(RestClient): headers = { "APCA-API-KEY-ID": self.key, "APCA-API-SECRET-KEY": self.secret, + 'Content-Type': 'application/json' } request.headers = headers @@ -247,28 +244,21 @@ class AlpacaRestApi(RestClient): self.key = key self.secret = secret self.init(url, proxy_host, proxy_port) - print("rest connect: ", url, proxy_host, proxy_port) self.start(session_num) self.connect_time = ( int(datetime.now().strftime("%y%m%d%H%M%S")) * self.order_count ) - print("rest client connected", self.connect_time) self.gateway.write_log("ALPACA REST API启动成功") self.query_account() self.query_position() #self.query_contracts() def on_send_order(self, data, request: Request ): - print("debug on_send_order data: ", data) - print("debug on_send_order request: ", request) - print("***debug on_send_order request: ", request.extra) remote_order_id = data['id'] order = request.extra self.order_dict[order.orderid]=remote_order_id - print("+++debug on_send_order request: ", self.order_dict) self.gateway.on_order(order) GLOBAL_ORDER[remote_order_id]=order - print("===+++ debug update global_order_dict: ", GLOBAL_ORDER) def on_failed_order(self, status_code: int, request: Request): """ @@ -278,7 +268,6 @@ class AlpacaRestApi(RestClient): order.status = Status.REJECTED self.gateway.on_order(order) msg = f"请求失败，状态码：{status_code}，信息：{request.response.text}" - print('debug on_failed', msg) self.gateway.write_log(msg) def on_error_order( @@ -298,12 +287,10 @@ class AlpacaRestApi(RestClient): self.exception_detail(exception_type, exception_value, tb, request) ) - # need debug 0608 def cancel_order(self, req: CancelRequest): """""" order_id = req.orderid remote_order_id = self.order_dict[order_id] - print("debug cancel order: order id ", order_id, "---", remote_order_id) if remote_order_id is None: print("[error]: can not get remote_order_id from local dict!") return @@ -345,11 +332,11 @@ class AlpacaRestApi(RestClient): "POST", "/v1/orders", callback=self.on_send_order, - # data=data, + data=data, extra=order, on_failed=self.on_failed_order, on_error=self.on_error_order, - json_str=data, + #json_str=data, ) print("debug send_order ret val : ", order.vt_orderid) return order.vt_orderid @@ -362,8 +349,8 @@ class AlpacaRestApi(RestClient): exchange=Exchange.ALPACA, name=symbol, product=Product.SPOT, - size=1, # need debug - pricetick=0.01, # need debug + size=1, + pricetick=0.01, gateway_name=self.gateway_name ) self.on_contract(contract) @@ -374,7 +361,6 @@ class AlpacaRestApi(RestClient): def on_error_query_contracts(self, exception_type: type, exception_value: Exception, tb, request: Request): pass - # need debug def query_contracts(self): params = {"status": "active"} self.add_request( @@ -384,7 +370,6 @@ class AlpacaRestApi(RestClient): callback=self.on_query_contracts, on_failed=self.on_failed_query_contracts, on_error=self.on_error_query_contracts, - # data=data, ) @@ -529,7 +514,6 @@ class AlpacaWebsocketApi(WebsocketClient): def handle_auth(self, data): stream_ret = data['stream'] data_ret = data['data'] - print("stream is {}, data is {}".format(stream_ret,data_ret)) if (data_ret['status'] == "authorized"): print("authorization success!!!") self.gateway.write_log("authorization success!!!") From 17fe9bf914a98612cd80bff328d958b259531903 Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Sat, 22 Jun 2019 10:51:50 +0800 Subject: [PATCH 15/16] remove not used rest_client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. "json的内容可以通过设置Content-Type为application/json达成，没必要单独增加一个json字段。" remove not used code 2. 在alpaca撤单调试过程中，发生返回值200，但body.text是json的情况，致rest_client解析json报错，加了try/catch做处理 --- vnpy/api/rest/rest_client.py | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/vnpy/api/rest/rest_client.py b/vnpy/api/rest/rest_client.py index 1af29631..ce2fb60a 100644 --- a/vnpy/api/rest/rest_client.py +++ b/vnpy/api/rest/rest_client.py @@ -49,10 +49,6 @@ class Request(object): self.response = None self.status = RequestStatus.ready - self.json=None - - def set_json(self, json_str: dict): - self.json=json_str def __str__(self): if self.response is None: @@ -65,7 +61,6 @@ class Request(object): "headers: {}\n" "params: {}\n" "data: {}\n" - "json: {}\n" "response:" "{}\n".format( self.method, @@ -75,7 +70,6 @@ class Request(object): self.headers, self.params, self.data, - self.json, "" if self.response is None else self.response.text, ) ) @@ -151,7 +145,6 @@ class RestClient(object): on_failed: Callable = None, on_error: Callable = None, extra: Any = None, - json_str: dict = None, ): """ Add a new request. @@ -177,8 +170,6 @@ class RestClient(object): on_error, extra, ) - if json_str is not None: - request.set_json(json_str) self._queue.put(request) return request @@ -262,7 +253,6 @@ class RestClient(object): headers=request.headers, params=request.params, data=request.data, - json=request.json, proxies=self.proxies, ) request.response = response From 985c06254973f48f0dddac10ab0791b22161673e Mon Sep 17 00:00:00 2001 From: vigarbuaa Date: Sat, 22 Jun 2019 10:56:03 +0800 Subject: [PATCH 16/16] format code --- vnpy/gateway/alpaca/alpaca_gateway.py | 90 +++++++++++++-------------- 1 file changed, 45 insertions(+), 45 deletions(-) diff --git a/vnpy/gateway/alpaca/alpaca_gateway.py b/vnpy/gateway/alpaca/alpaca_gateway.py index a34e8529..31b37d24 100644 --- a/vnpy/gateway/alpaca/alpaca_gateway.py +++ b/vnpy/gateway/alpaca/alpaca_gateway.py @@ -49,7 +49,7 @@ STATUS_ALPACA2VT = { "partial_fill": Status.PARTTRADED, "fill": Status.ALLTRADED, "cancelled": Status.CANCELLED, - #"done_for_day": Status.CANCELLED, + # "done_for_day": Status.CANCELLED, "expired": Status.NOTTRADED } @@ -62,7 +62,8 @@ ORDERTYPE_VT2ALPACA = { OrderType.MARKET: "market" } ORDERTYPE_ALPACA2VT = {v: k for k, v in ORDERTYPE_VT2ALPACA.items()} -GLOBAL_ORDER={} +GLOBAL_ORDER = {} + class AlpacaGateway(BaseGateway): """ @@ -93,11 +94,12 @@ class AlpacaGateway(BaseGateway): session = setting["session"] proxy_host = setting["proxy_host"] proxy_port = setting["proxy_port"] - env=setting['服务器'] - rest_url = REST_HOST if env == "REAL" else PAPER_REST_HOST - websocket_url = WEBSOCKET_HOST if env == "REAL" else PAPER_WEBSOCKET_HOST - self.rest_api.connect(key, secret, session, proxy_host, proxy_port,rest_url) - self.ws_api.connect(key, secret, proxy_host, proxy_port,websocket_url) + env = setting['服务器'] + rest_url = REST_HOST if env == "REAL" else PAPER_REST_HOST + websocket_url = WEBSOCKET_HOST if env == "REAL" else PAPER_WEBSOCKET_HOST + self.rest_api.connect(key, secret, session, + proxy_host, proxy_port, rest_url) + self.ws_api.connect(key, secret, proxy_host, proxy_port, websocket_url) self.init_query() def subscribe(self, req: SubscribeRequest): @@ -124,7 +126,7 @@ class AlpacaGateway(BaseGateway): """""" self.rest_api.stop() self.ws_api.stop() - + def process_timer_event(self, event: Event): """""" self.count += 1 @@ -139,7 +141,6 @@ class AlpacaGateway(BaseGateway): self.count = 0 self.event_engine.register(EVENT_TIMER, self.process_timer_event) - def process_timer_event(self, event: Event): """""" self.count += 1 @@ -170,7 +171,7 @@ class AlpacaRestApi(RestClient): self.order_count = 1_000_000 self.order_count_lock = Lock() self.connect_time = 0 - self.order_dict ={} + self.order_dict = {} def query_account(self): path = f"/v1/account" @@ -204,7 +205,7 @@ class AlpacaRestApi(RestClient): exchange=Exchange.ALPACA, direction=DIRECTION_ALPACA2VT[d['side']], volume=d['qty'], - price=round(d['avg_entry_price'],3), + price=round(d['avg_entry_price'], 3), pnl=d['unrealized_pl'], gateway_name=self.gateway_name, ) @@ -251,14 +252,14 @@ class AlpacaRestApi(RestClient): self.gateway.write_log("ALPACA REST API启动成功") self.query_account() self.query_position() - #self.query_contracts() + # self.query_contracts() - def on_send_order(self, data, request: Request ): + def on_send_order(self, data, request: Request): remote_order_id = data['id'] order = request.extra - self.order_dict[order.orderid]=remote_order_id + self.order_dict[order.orderid] = remote_order_id self.gateway.on_order(order) - GLOBAL_ORDER[remote_order_id]=order + GLOBAL_ORDER[remote_order_id] = order def on_failed_order(self, status_code: int, request: Request): """ @@ -308,7 +309,7 @@ class AlpacaRestApi(RestClient): """Websocket will push a new order status""" pass - def on_cancel_order_error( self, exception_type: type, exception_value: Exception, tb, request: Request): + def on_cancel_order_error(self, exception_type: type, exception_value: Exception, tb, request: Request): # Record exception if not ConnectionError if not issubclass(exception_type, ConnectionError): self.on_error(exception_type, exception_value, tb, request) @@ -327,7 +328,7 @@ class AlpacaRestApi(RestClient): data = raw_dict order = req.create_order_data(orderid, self.gateway_name) - print("debug send_order orderBody extra: ",order) + print("debug send_order orderBody extra: ", order) self.add_request( "POST", "/v1/orders", @@ -336,20 +337,20 @@ class AlpacaRestApi(RestClient): extra=order, on_failed=self.on_failed_order, on_error=self.on_error_order, - #json_str=data, + # json_str=data, ) print("debug send_order ret val : ", order.vt_orderid) return order.vt_orderid def on_query_contracts(self, data, request: Request): - for instrument_data in data: + for instrument_data in data: symbol = instrument_data['symbol'] contract = ContractData( symbol=symbol, - exchange=Exchange.ALPACA, + exchange=Exchange.ALPACA, name=symbol, product=Product.SPOT, - size=1, + size=1, pricetick=0.01, gateway_name=self.gateway_name ) @@ -360,13 +361,13 @@ class AlpacaRestApi(RestClient): def on_error_query_contracts(self, exception_type: type, exception_value: Exception, tb, request: Request): pass - + def query_contracts(self): params = {"status": "active"} self.add_request( "GET", "/v1/assets", - params = params, + params=params, callback=self.on_query_contracts, on_failed=self.on_failed_query_contracts, on_error=self.on_error_query_contracts, @@ -406,10 +407,10 @@ class AlpacaWebsocketApi(WebsocketClient): self.askDict = {} self.orderLocalDict = {} self.channelDict = {} # ChannelID : (Channel, Symbol) - self.channels=["account_updates", "trade_updates"] + self.channels = ["account_updates", "trade_updates"] def connect( - self, key: str, secret: str, proxy_host: str, proxy_port: int,url:str + self, key: str, secret: str, proxy_host: str, proxy_port: int, url: str ): """""" self.key = key @@ -419,24 +420,24 @@ class AlpacaWebsocketApi(WebsocketClient): def authenticate(self): """""" - params={"action":"authenticate", "data": { - "key_id":self.key,"secret_key":self.secret + params = {"action": "authenticate", "data": { + "key_id": self.key, "secret_key": self.secret }} self.send_packet(params) def on_authenticate(self): """""" - params={"action":"listen", "data": { - "streams":self.channels + params = {"action": "listen", "data": { + "streams": self.channels }} self.send_packet(params) - + def subscribe(self, req: SubscribeRequest): - self.channels.append("A."+req.symbol) - self.channels.append("Q."+req.symbol) + self.channels.append("A." + req.symbol) + self.channels.append("Q." + req.symbol) print("debug subscribe: {} ".format(self.channels)) - params={"action":"listen", "data": { - "streams":self.channels + params = {"action": "listen", "data": { + "streams": self.channels }} self.send_packet(params) @@ -450,7 +451,7 @@ class AlpacaWebsocketApi(WebsocketClient): def on_connected(self): """""" - self.gateway.write_log("Websocket API连接成功") + self.gateway.write_log("Websocket API连接成功") self.authenticate() def on_disconnected(self): @@ -463,9 +464,9 @@ class AlpacaWebsocketApi(WebsocketClient): if "stream" in packet and "data" in packet: stream_ret = packet['stream'] data_ret = packet['data'] - if(stream_ret == "authorization"): + if(stream_ret == "authorization"): self.handle_auth(packet) - elif(stream_ret == "listening"): + elif(stream_ret == "listening"): self.gateway.write_log("listening {}".format(data_ret)) else: self.on_data(packet) @@ -478,19 +479,18 @@ class AlpacaWebsocketApi(WebsocketClient): stream_ret = data['stream'] data_ret = data['data'] if(stream_ret == "account_updates"): - #handle account - print("!!!!!!!@@@@@@@@ account_udpate is {}".format(data)) account = AccountData( accountid=data_ret['id'], balance=float(data_ret['cash']), - frozen=float(data_ret['cash']) - float(data_ret['cash_withdrawable']), + frozen=float(data_ret['cash']) - + float(data_ret['cash_withdrawable']), gateway_name=self.gateway_name ) self.gateway.on_account(account) elif(stream_ret == "trade_updates"): - d=data_ret['order'] + d = data_ret['order'] order_id = d['id'] - order=GLOBAL_ORDER[order_id] + order = GLOBAL_ORDER[order_id] local_order_id = order.orderid order.status = STATUS_ALPACA2VT[data_ret['event']] self.gateway.on_order(order) @@ -522,7 +522,7 @@ class AlpacaWebsocketApi(WebsocketClient): print("authorization failed!!!") self.gateway.write_log("authorization failed!!!") else: - print("??unhandled status: ",data) + print("??unhandled status: ", data) # ---------------------------------------------------------------------- def on_response(self, data): @@ -540,7 +540,7 @@ class AlpacaWebsocketApi(WebsocketClient): """""" print("on_error: ", type, Exception, tb) sys.stderr.write( - self.exception_detail(exception_type, exception_value, tb ) + self.exception_detail(exception_type, exception_value, tb) ) def subscribe_topic(self): @@ -578,4 +578,4 @@ class AlpacaWebsocketApi(WebsocketClient): def on_contract(self, d): """""" - pass \ No newline at end of file + pass