[Add] OandaGateway
This commit is contained in:
nanoric
parent
afb079c9ca
commit
d805ed043c
@ -1,3 +1,4 @@
|
||||
import json
|
||||
import multiprocessing
|
||||
import os
|
||||
import sys
|
||||
@ -5,8 +6,9 @@ import traceback
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
from multiprocessing.dummy import Pool
|
||||
from threading import Lock
|
||||
from typing import Any, Callable, List, Optional, Union
|
||||
from threading import Lock, Thread
|
||||
from types import TracebackType
|
||||
from typing import Any, Callable, List, Optional, Type, Union
|
||||
|
||||
import requests
|
||||
|
||||
@ -20,6 +22,11 @@ class RequestStatus(Enum):
|
||||
|
||||
pool: multiprocessing.pool.Pool = Pool(os.cpu_count() * 20)
|
||||
|
||||
CALLBACK_TYPE = Callable[[dict, "Request"], Any]
|
||||
ON_FAILED_TYPE = Callable[[int, "Request"], Any]
|
||||
ON_ERROR_TYPE = Callable[[Type, Exception, TracebackType, "Request"], Any]
|
||||
CONNECTED_TYPE = Callable[["Request"], Any]
|
||||
|
||||
|
||||
class Request(object):
|
||||
"""
|
||||
@ -33,10 +40,13 @@ class Request(object):
|
||||
params: dict,
|
||||
data: Union[dict, str, bytes],
|
||||
headers: dict,
|
||||
callback: Callable = None,
|
||||
on_failed: Callable = None,
|
||||
on_error: Callable = None,
|
||||
callback: CALLBACK_TYPE = None,
|
||||
on_failed: ON_FAILED_TYPE = None,
|
||||
on_error: ON_ERROR_TYPE = None,
|
||||
stream: bool = False,
|
||||
on_connected: CONNECTED_TYPE = None, # for streaming request
|
||||
extra: Any = None,
|
||||
client: "RestClient" = None,
|
||||
):
|
||||
""""""
|
||||
self.method = method
|
||||
@ -46,12 +56,17 @@ class Request(object):
|
||||
self.data = data
|
||||
self.headers = headers
|
||||
|
||||
self.stream = stream
|
||||
self.on_connected = on_connected
|
||||
self.processing_line: Optional[str] = ''
|
||||
|
||||
self.on_failed = on_failed
|
||||
self.on_error = on_error
|
||||
self.extra = extra
|
||||
|
||||
self.response = None
|
||||
self.response: Optional[requests.Response] = None
|
||||
self.status = RequestStatus.ready
|
||||
self.client: "RestClient" = client
|
||||
|
||||
def __str__(self):
|
||||
if self.response is None:
|
||||
@ -59,21 +74,31 @@ class Request(object):
|
||||
else:
|
||||
status_code = self.response.status_code
|
||||
|
||||
if self.stream:
|
||||
response = self.processing_line
|
||||
else:
|
||||
if self.response is None:
|
||||
response = None
|
||||
else:
|
||||
response = self.response.text
|
||||
|
||||
return (
|
||||
"request : {} {} {} because {}: \n"
|
||||
"headers: {}\n"
|
||||
"params: {}\n"
|
||||
"data: {}\n"
|
||||
"response:"
|
||||
"{}\n".format(
|
||||
self.method,
|
||||
self.path,
|
||||
self.status.name,
|
||||
status_code,
|
||||
self.headers,
|
||||
self.params,
|
||||
self.data,
|
||||
"" if self.response is None else self.response.text,
|
||||
"request: {method} {path} {http_code}: \n"
|
||||
"full_url: {full_url}\n"
|
||||
"status: {status}\n"
|
||||
"headers: {headers}\n"
|
||||
"params: {params}\n"
|
||||
"data: {data}\n"
|
||||
"response: {response}\n".format(
|
||||
full_url=self.client.make_full_url(self.path),
|
||||
status=self.status.name,
|
||||
method=self.method,
|
||||
path=self.path,
|
||||
http_code=status_code,
|
||||
headers=self.headers,
|
||||
params=self.params,
|
||||
data=self.data,
|
||||
response=response,
|
||||
)
|
||||
)
|
||||
|
||||
@ -114,6 +139,13 @@ class RestClient(object):
|
||||
self._sessions_lock = Lock()
|
||||
self._sessions: List[requests.Session] = []
|
||||
|
||||
self._streams_lock = Lock()
|
||||
self._streams: List[Thread] = []
|
||||
|
||||
@property
|
||||
def alive(self):
|
||||
return self._active
|
||||
|
||||
def init(self, url_base: str, proxy_host: str = "", proxy_port: int = 0):
|
||||
"""
|
||||
Init rest client with url_base which is the API root address.
|
||||
@ -150,16 +182,55 @@ class RestClient(object):
|
||||
for task in self._tasks:
|
||||
task.wait()
|
||||
|
||||
def add_streaming_request(
|
||||
self,
|
||||
method: str,
|
||||
path: str,
|
||||
callback: CALLBACK_TYPE,
|
||||
params: dict = None,
|
||||
data: Union[dict, str, bytes] = None,
|
||||
headers: dict = None,
|
||||
on_connected: CONNECTED_TYPE = None,
|
||||
on_failed: ON_FAILED_TYPE = None,
|
||||
on_error: ON_ERROR_TYPE = None,
|
||||
extra: Any = None,
|
||||
):
|
||||
"""
|
||||
See add_request for usage.
|
||||
"""
|
||||
request = Request(
|
||||
method=method,
|
||||
path=path,
|
||||
params=params,
|
||||
data=data,
|
||||
headers=headers,
|
||||
callback=callback,
|
||||
on_failed=on_failed,
|
||||
on_error=on_error,
|
||||
extra=extra,
|
||||
client=self,
|
||||
stream=True,
|
||||
on_connected=on_connected,
|
||||
)
|
||||
# stream task should not push to thread pool
|
||||
# because it is usually no return.
|
||||
th = Thread(
|
||||
target=self._process_stream_request,
|
||||
args=[request, ],
|
||||
)
|
||||
th.start()
|
||||
return request
|
||||
|
||||
def add_request(
|
||||
self,
|
||||
method: str,
|
||||
path: str,
|
||||
callback: Callable,
|
||||
callback: CALLBACK_TYPE,
|
||||
params: dict = None,
|
||||
data: Union[dict, str, bytes] = None,
|
||||
headers: dict = None,
|
||||
on_failed: Callable = None,
|
||||
on_error: Callable = None,
|
||||
on_failed: ON_FAILED_TYPE = None,
|
||||
on_error: ON_ERROR_TYPE = None,
|
||||
extra: Any = None,
|
||||
):
|
||||
"""
|
||||
@ -170,7 +241,7 @@ class RestClient(object):
|
||||
:param params: dict for query string
|
||||
:param data: Http body. If it is a dict, it will be converted to form-data. Otherwise, it will be converted to bytes.
|
||||
:param headers: dict for headers
|
||||
:param on_failed: callback function if Non-2xx status, type, type: (code, dict, Request)
|
||||
:param on_failed: callback function if Non-2xx status, type, type: (code, Request)
|
||||
:param on_error: callback function when catching Python exception, type: (etype, evalue, tb, Request)
|
||||
:param extra: Any extra data which can be used when handling callback
|
||||
:return: Request
|
||||
@ -185,6 +256,7 @@ class RestClient(object):
|
||||
on_failed=on_failed,
|
||||
on_error=on_error,
|
||||
extra=extra,
|
||||
client=self,
|
||||
)
|
||||
task = pool.apply_async(
|
||||
self._process_request,
|
||||
@ -204,6 +276,11 @@ class RestClient(object):
|
||||
not_finished_tasks = [i for i in self._tasks if not i.ready()]
|
||||
self._tasks = not_finished_tasks
|
||||
|
||||
def _clean_finished_streams(self):
|
||||
with self._streams_lock:
|
||||
not_finished_streams = [i for i in self._streams if i.is_alive()]
|
||||
self._streams = not_finished_streams
|
||||
|
||||
def _get_session(self):
|
||||
with self._sessions_lock:
|
||||
if self._sessions:
|
||||
@ -257,6 +334,21 @@ class RestClient(object):
|
||||
)
|
||||
return text
|
||||
|
||||
def is_request_success(self, data: dict, request: "Request"):
|
||||
"""
|
||||
check if a request succeed
|
||||
default behavior is returning True
|
||||
:return True if succeed.
|
||||
"""
|
||||
return True
|
||||
|
||||
def _process_stream_request(self, request: Request):
|
||||
"""
|
||||
Sending request to server and get result.
|
||||
"""
|
||||
self._process_request(request)
|
||||
self._clean_finished_streams()
|
||||
|
||||
def _process_request(
|
||||
self, request: Request
|
||||
):
|
||||
@ -269,6 +361,7 @@ class RestClient(object):
|
||||
|
||||
url = self.make_full_url(request.path)
|
||||
|
||||
stream = request.stream
|
||||
response = session.request(
|
||||
request.method,
|
||||
url,
|
||||
@ -276,24 +369,35 @@ class RestClient(object):
|
||||
params=request.params,
|
||||
data=request.data,
|
||||
proxies=self.proxies,
|
||||
stream=stream,
|
||||
)
|
||||
request.response = response
|
||||
status_code = response.status_code
|
||||
if status_code // 100 == 2: # 2xx codes are all successful
|
||||
if status_code == 204:
|
||||
json_body = None
|
||||
else:
|
||||
json_body = response.json()
|
||||
|
||||
request.callback(json_body, request)
|
||||
if not stream: # normal API:
|
||||
# just call callback with all contents received.
|
||||
if status_code // 100 == 2: # 2xx codes are all successful
|
||||
if status_code == 204:
|
||||
json_body = None
|
||||
else:
|
||||
json_body = response.json()
|
||||
self._process_json_body(json_body, request)
|
||||
else:
|
||||
if request.on_failed:
|
||||
request.status = RequestStatus.failed
|
||||
request.on_failed(status_code, request)
|
||||
else:
|
||||
self.on_failed(status_code, request)
|
||||
else: # streaming API:
|
||||
if request.on_connected:
|
||||
request.on_connected(request)
|
||||
# split response by lines, and call one callback for each line.
|
||||
for line in response.iter_lines(chunk_size=None):
|
||||
if line:
|
||||
request.processing_line = line
|
||||
json_body = json.loads(line)
|
||||
self._process_json_body(json_body, request)
|
||||
request.status = RequestStatus.success
|
||||
else:
|
||||
request.status = RequestStatus.failed
|
||||
|
||||
if request.on_failed:
|
||||
request.on_failed(status_code, request)
|
||||
else:
|
||||
self.on_failed(status_code, request)
|
||||
except Exception:
|
||||
request.status = RequestStatus.error
|
||||
t, v, tb = sys.exc_info()
|
||||
@ -302,6 +406,18 @@ class RestClient(object):
|
||||
else:
|
||||
self.on_error(t, v, tb, request)
|
||||
|
||||
def _process_json_body(self, json_body: Optional[dict], request: "Request"):
|
||||
status_code = request.response.status_code
|
||||
if self.is_request_success(json_body, request):
|
||||
request.status = RequestStatus.success
|
||||
request.callback(json_body, request)
|
||||
else:
|
||||
request.status = RequestStatus.failed
|
||||
if request.on_failed:
|
||||
request.on_failed(status_code, request)
|
||||
else:
|
||||
self.on_failed(status_code, request)
|
||||
|
||||
def make_full_url(self, path: str):
|
||||
"""
|
||||
Make relative api path into full url.
|
||||
@ -332,7 +448,8 @@ class RestClient(object):
|
||||
path,
|
||||
params,
|
||||
data,
|
||||
headers
|
||||
headers,
|
||||
client=self,
|
||||
)
|
||||
request = self.sign(request)
|
||||
|
||||
|
||||
1
vnpy/gateway/oanda/__init__.py
Normal file
1
vnpy/gateway/oanda/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
from .oanda_gateway import OandaGateway
|
||||
65
vnpy/gateway/oanda/oanda_api_base.py
Normal file
65
vnpy/gateway/oanda/oanda_api_base.py
Normal file
@ -0,0 +1,65 @@
|
||||
import json
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from threading import Lock
|
||||
from typing import Dict, List, Optional, TYPE_CHECKING, Tuple
|
||||
|
||||
from requests import ConnectionError
|
||||
|
||||
from vnpy.api.rest import Request
|
||||
from vnpy.trader.constant import Direction, Exchange, Interval, Offset, Product, Status
|
||||
from vnpy.trader.object import AccountData, BarData, CancelRequest, ContractData, OrderRequest, \
|
||||
PositionData, OrderData
|
||||
from .oanda_common import (INTERVAL_VT2OANDA, ORDER_TYPE_VT2OANDA,
|
||||
STATUS_OANDA2VT, parse_time, INTERVAL_VT2OANDA_DELTA, parse_datetime)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from vnpy.gateway.oanda import OandaGateway
|
||||
from vnpy.api.rest import RestClient
|
||||
|
||||
_ = lambda x: x # noqa
|
||||
|
||||
|
||||
class OandaApiBase(RestClient):
|
||||
"""
|
||||
Oanda Base API
|
||||
"""
|
||||
def __init__(self, gateway: "OandaGateway"):
|
||||
super().__init__()
|
||||
self.gateway = gateway
|
||||
self.gateway_name = gateway.gateway_name
|
||||
|
||||
self.key = ""
|
||||
self.secret = b""
|
||||
|
||||
def sign(self, request):
|
||||
"""
|
||||
Generate BitMEX signature.
|
||||
"""
|
||||
if request.data:
|
||||
request.data = json.dumps(request.data)
|
||||
request.headers = {
|
||||
"Authorization": f"Bearer {self.key}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
return request
|
||||
|
||||
def is_request_success(self, data: dict, request: "Request"):
|
||||
# message should be check too
|
||||
# but checking only this is enough for us.
|
||||
return super().is_request_success(data, request) and 'errorMessage' not in data
|
||||
|
||||
def on_error(
|
||||
self, exception_type: type, exception_value: Exception, tb, request: Request
|
||||
):
|
||||
"""
|
||||
Callback to handler request exception.
|
||||
"""
|
||||
msg = f"触发异常，状态码：{exception_type}，信息：{exception_value}"
|
||||
self.gateway.write_log(msg)
|
||||
|
||||
sys.stderr.write(
|
||||
self.exception_detail(exception_type, exception_value, tb, request)
|
||||
)
|
||||
|
||||
86
vnpy/gateway/oanda/oanda_common.py
Normal file
86
vnpy/gateway/oanda/oanda_common.py
Normal file
@ -0,0 +1,86 @@
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from vnpy.trader.constant import Direction, Interval, OrderType, Status
|
||||
|
||||
if TYPE_CHECKING:
|
||||
# noinspection PyUnresolvedReferences
|
||||
from vnpy.gateway.oanda import OandaGateway
|
||||
|
||||
STATUS_OANDA2VT = {
|
||||
"PENDING": Status.NOTTRADED,
|
||||
"FILLED": Status.ALLTRADED,
|
||||
"CANCELLED": Status.CANCELLED,
|
||||
# "TRIGGERED": Status.REJECTED,
|
||||
}
|
||||
STOP_ORDER_STATUS_OANDA2VT = {
|
||||
"Untriggered": Status.NOTTRADED,
|
||||
"Triggered": Status.NOTTRADED,
|
||||
# Active: triggered and placed.
|
||||
# since price is market price, placed == AllTraded?
|
||||
"Active": Status.ALLTRADED,
|
||||
"Cancelled": Status.CANCELLED,
|
||||
"Rejected": Status.REJECTED,
|
||||
}
|
||||
DIRECTION_VT2OANDA = {Direction.LONG: "Buy", Direction.SHORT: "Sell"}
|
||||
DIRECTION_OANDA2VT = {v: k for k, v in DIRECTION_VT2OANDA.items()}
|
||||
DIRECTION_OANDA2VT.update({
|
||||
"None": Direction.LONG
|
||||
})
|
||||
|
||||
OPPOSITE_DIRECTION = {
|
||||
Direction.LONG: Direction.SHORT,
|
||||
Direction.SHORT: Direction.LONG,
|
||||
}
|
||||
|
||||
ORDER_TYPE_VT2OANDA = {
|
||||
OrderType.LIMIT: "LIMIT",
|
||||
OrderType.MARKET: "MARKET",
|
||||
OrderType.STOP: "STOP",
|
||||
}
|
||||
|
||||
ORDER_TYPE_OANDA2VT = {v: k for k, v in ORDER_TYPE_VT2OANDA.items()}
|
||||
ORDER_TYPE_OANDA2VT.update({
|
||||
'LIMIT_ORDER': OrderType.LIMIT,
|
||||
'MARKET_ORDER': OrderType.MARKET,
|
||||
'STOP_ORDER': OrderType.STOP,
|
||||
})
|
||||
|
||||
INTERVAL_VT2OANDA = {
|
||||
Interval.MINUTE: "M1",
|
||||
Interval.HOUR: "H1",
|
||||
Interval.DAILY: "D",
|
||||
Interval.WEEKLY: "W",
|
||||
}
|
||||
INTERVAL_VT2OANDA_INT = {
|
||||
Interval.MINUTE: 1,
|
||||
Interval.HOUR: 60,
|
||||
Interval.DAILY: 60 * 24,
|
||||
Interval.WEEKLY: 60 * 24 * 7,
|
||||
}
|
||||
INTERVAL_VT2OANDA_DELTA = {
|
||||
Interval.MINUTE: timedelta(minutes=1),
|
||||
Interval.HOUR: timedelta(hours=1),
|
||||
Interval.DAILY: timedelta(days=1),
|
||||
Interval.WEEKLY: timedelta(days=7),
|
||||
}
|
||||
|
||||
utc_tz = timezone.utc
|
||||
local_tz = datetime.now(timezone.utc).astimezone().tzinfo
|
||||
|
||||
|
||||
def generate_timestamp(expire_after: float = 30) -> int:
|
||||
"""
|
||||
:param expire_after: expires in seconds.
|
||||
:return: timestamp in milliseconds
|
||||
"""
|
||||
return int(time.time() * 1000 + expire_after * 1000)
|
||||
|
||||
|
||||
def parse_datetime(dt: str) -> datetime:
|
||||
return datetime.fromisoformat(dt[:-4])
|
||||
|
||||
|
||||
def parse_time(dt: str) -> str:
|
||||
return dt[11:26]
|
||||
208
vnpy/gateway/oanda/oanda_gateway.py
Normal file
208
vnpy/gateway/oanda/oanda_gateway.py
Normal file
@ -0,0 +1,208 @@
|
||||
"""
|
||||
author: nanoric
|
||||
todo:
|
||||
* query history
|
||||
"""
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import Dict, List, Any, Tuple, Optional
|
||||
|
||||
from vnpy.event import Event
|
||||
from vnpy.trader.constant import (Exchange, OrderType, Interval, Direction, Offset, Status)
|
||||
from vnpy.trader.event import EVENT_TIMER
|
||||
from vnpy.trader.gateway import BaseGateway
|
||||
from vnpy.trader.object import (CancelRequest, HistoryRequest, OrderData, OrderRequest,
|
||||
PositionData, SubscribeRequest, TickData, BarData)
|
||||
from .oanda_common import DIRECTION_OANDA2VT, OPPOSITE_DIRECTION, ORDER_TYPE_OANDA2VT, \
|
||||
STOP_ORDER_STATUS_OANDA2VT, parse_datetime, STATUS_OANDA2VT, INTERVAL_VT2OANDA, local_tz, \
|
||||
utc_tz, INTERVAL_VT2OANDA_INT, parse_time, INTERVAL_VT2OANDA_DELTA
|
||||
from .oanda_rest_api import OandaRestApi, HistoryDataNextInfo
|
||||
from .oanda_stream_api import OandaStreamApi
|
||||
|
||||
|
||||
_ = lambda x:x # noqa
|
||||
|
||||
|
||||
@dataclass()
|
||||
class HistoryDataInfo:
|
||||
bars: List[BarData]
|
||||
extra: Any
|
||||
|
||||
|
||||
class OandaGateway(BaseGateway):
|
||||
"""
|
||||
VN Trader Gateway for BitMEX connection.
|
||||
"""
|
||||
|
||||
default_setting = {
|
||||
"APIKey": "",
|
||||
"会话数": 3,
|
||||
"服务器": ["REAL", "TESTNET"],
|
||||
"代理地址": "",
|
||||
"代理端口": "",
|
||||
}
|
||||
HISTORY_RECORD_PER_REQUEST = 5000 # # of records per history request
|
||||
|
||||
exchanges = [Exchange.OANDA]
|
||||
|
||||
def __init__(self, event_engine):
|
||||
"""Constructor"""
|
||||
super(OandaGateway, self).__init__(event_engine, "OANDA")
|
||||
|
||||
self.rest_api = OandaRestApi(self)
|
||||
self.stream_api = OandaStreamApi(self)
|
||||
|
||||
self.account_id: Optional[str] = None
|
||||
|
||||
self.orders: Dict[str, OrderData] = {}
|
||||
self.local2sys_map: Dict[str, str] = {}
|
||||
event_engine.register(EVENT_TIMER, self.process_timer_event)
|
||||
|
||||
def connect(self, setting: dict):
|
||||
""""""
|
||||
key = setting["APIKey"]
|
||||
session_number = setting["会话数"]
|
||||
server = setting["服务器"]
|
||||
proxy_host = setting["代理地址"]
|
||||
proxy_port = setting["代理端口"]
|
||||
|
||||
if proxy_port.isdigit():
|
||||
proxy_port = int(proxy_port)
|
||||
else:
|
||||
proxy_port = 0
|
||||
|
||||
self.stream_api.connect(key, session_number,
|
||||
server, proxy_host, proxy_port)
|
||||
self.rest_api.connect(key, session_number,
|
||||
server, proxy_host, proxy_port)
|
||||
self.query_account()
|
||||
|
||||
def subscribe(self, req: SubscribeRequest):
|
||||
""""""
|
||||
assert self.account_id is not None, _("请先初始化并连接RestAPI")
|
||||
self.stream_api.subscribe(req)
|
||||
|
||||
def send_order(self, req: OrderRequest):
|
||||
""""""
|
||||
return self.rest_api.send_order(req)
|
||||
|
||||
def cancel_order(self, req: CancelRequest):
|
||||
""""""
|
||||
self.rest_api.cancel_order(req)
|
||||
|
||||
def query_account(self):
|
||||
""""""
|
||||
self.rest_api.query_accounts()
|
||||
|
||||
def query_position(self):
|
||||
""""""
|
||||
self.rest_api.query_positions()
|
||||
|
||||
def query_first_history(self,
|
||||
symbol: str,
|
||||
interval: Interval,
|
||||
start: datetime,
|
||||
) -> Tuple[List[BarData], "HistoryDataNextInfo"]:
|
||||
|
||||
# datetime for a bar is close_time
|
||||
# we got open_time from API.
|
||||
adjustment = INTERVAL_VT2OANDA_DELTA[interval]
|
||||
|
||||
utc_time = start.replace(tzinfo=local_tz).astimezone(tz=utc_tz)
|
||||
return self.rest_api.query_history(
|
||||
symbol=symbol,
|
||||
interval=interval,
|
||||
|
||||
# todo: vnpy: shall all datetime object use tzinfo?
|
||||
start=utc_time - adjustment,
|
||||
)
|
||||
|
||||
def query_next_history(self,
|
||||
next_info: Any,
|
||||
):
|
||||
data: "HistoryDataNextInfo" = next_info
|
||||
return self.rest_api.query_history(
|
||||
symbol=data.symbol,
|
||||
interval=data.interval,
|
||||
start=data.end,
|
||||
)
|
||||
|
||||
def query_history(self, req: HistoryRequest):
|
||||
"""
|
||||
todo: vnpy: download in parallel
|
||||
todo: vnpy: use yield to simplify logic
|
||||
:raises RequestFailedException: if server reply an error.
|
||||
Any Exception might be raised from requests.request: network error.
|
||||
"""
|
||||
history = []
|
||||
|
||||
symbol = req.symbol
|
||||
interval = req.interval
|
||||
start = req.start
|
||||
|
||||
bars, next_info = self.query_first_history(
|
||||
symbol=symbol,
|
||||
interval=interval,
|
||||
start=start,
|
||||
)
|
||||
msg = f"获取历史数据成功，{req.symbol} - {req.interval.value}，{bars[0].datetime} - {bars[-1].datetime}"
|
||||
self.write_log(msg)
|
||||
history.extend(bars)
|
||||
|
||||
end = req.end
|
||||
if end is None:
|
||||
end = datetime.now()
|
||||
if bars[-1].datetime >= end or len(bars) < self.HISTORY_RECORD_PER_REQUEST:
|
||||
return history
|
||||
while True:
|
||||
bars, next_info = self.query_next_history(next_info)
|
||||
history.extend(bars)
|
||||
|
||||
msg = f"获取历史数据成功，{req.symbol} - {req.interval.value}，{bars[0].datetime} - {bars[-1].datetime}"
|
||||
self.write_log(msg)
|
||||
|
||||
# Break if total data count less than <count> (latest date collected)
|
||||
if bars[-1].datetime >= end or len(bars) < self.HISTORY_RECORD_PER_REQUEST:
|
||||
break
|
||||
return history
|
||||
|
||||
def close(self):
|
||||
""""""
|
||||
self.rest_api.stop()
|
||||
self.stream_api.stop()
|
||||
|
||||
def process_timer_event(self, event: Event):
|
||||
""""""
|
||||
if self.rest_api.fully_initialized:
|
||||
self.rest_api.query_account_changes()
|
||||
|
||||
def write_log(self, msg: str):
|
||||
print(msg)
|
||||
return super().write_log(msg)
|
||||
|
||||
def parse_order_data(self, data, status: Status, time_key: str):
|
||||
client_extension = data.get('clientExtensions', None)
|
||||
if client_extension is None:
|
||||
order_id = data['id']
|
||||
else:
|
||||
order_id = client_extension['id']
|
||||
vol = int(data['units'])
|
||||
type_ = ORDER_TYPE_OANDA2VT[data['type']]
|
||||
|
||||
order = OrderData(
|
||||
gateway_name=self.gateway_name,
|
||||
symbol=data['instrument'],
|
||||
exchange=Exchange.OANDA,
|
||||
orderid=order_id,
|
||||
type=type_,
|
||||
direction=Direction.LONG if vol > 0 else Direction.SHORT,
|
||||
offset=Offset.NONE,
|
||||
price=float(data['price']) if type_ is not OrderType.MARKET else 0.0,
|
||||
volume=abs(vol),
|
||||
# status=STATUS_OANDA2VT[data['state']],
|
||||
status=status,
|
||||
time=parse_time(data[time_key]),
|
||||
)
|
||||
self.orders[order_id] = order
|
||||
return order
|
||||
|
||||
454
vnpy/gateway/oanda/oanda_rest_api.py
Normal file
454
vnpy/gateway/oanda/oanda_rest_api.py
Normal file
@ -0,0 +1,454 @@
|
||||
import json
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from threading import Lock
|
||||
from typing import Dict, List, Optional, TYPE_CHECKING, Tuple
|
||||
|
||||
from requests import ConnectionError
|
||||
|
||||
from vnpy.api.rest import Request
|
||||
from vnpy.gateway.oanda.oanda_api_base import OandaApiBase
|
||||
from vnpy.trader.constant import Direction, Exchange, Interval, Offset, Product, Status
|
||||
from vnpy.trader.object import AccountData, BarData, CancelRequest, ContractData, OrderRequest, \
|
||||
PositionData, OrderData
|
||||
from .oanda_common import (INTERVAL_VT2OANDA, ORDER_TYPE_VT2OANDA,
|
||||
STATUS_OANDA2VT, parse_time, INTERVAL_VT2OANDA_DELTA, parse_datetime,
|
||||
utc_tz, local_tz)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from vnpy.gateway.oanda import OandaGateway
|
||||
_ = lambda x: x # noqa
|
||||
|
||||
HOST = "https://api-fxtrade.oanda.com"
|
||||
TEST_HOST = "https://api-fxpractice.oanda.com"
|
||||
|
||||
# asked from official developer
|
||||
PRICE_TICKS = {
|
||||
"BTCUSD": 0.5,
|
||||
"ETHUSD": 0.05,
|
||||
"EOSUSD": 0.001,
|
||||
"XRPUSD": 0.0001,
|
||||
}
|
||||
|
||||
|
||||
@dataclass()
|
||||
class HistoryDataNextInfo:
|
||||
symbol: str
|
||||
interval: Interval
|
||||
end: datetime
|
||||
|
||||
|
||||
class RequestFailedException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class OandaRestApi(OandaApiBase):
|
||||
"""
|
||||
Oanda Rest API
|
||||
"""
|
||||
|
||||
def __init__(self, gateway: "OandaGateway"):
|
||||
""""""
|
||||
super().__init__(gateway)
|
||||
|
||||
self.order_count = 1_000_000
|
||||
self.order_count_lock = Lock()
|
||||
|
||||
self.connect_time = 0
|
||||
|
||||
self.contracts: Dict[str, ContractData] = {}
|
||||
|
||||
self.last_account_transaction_id: Optional[str] = None
|
||||
|
||||
# used for automatic tests.
|
||||
self.account_initialized = False
|
||||
self.orders_initialized = False
|
||||
|
||||
def connect(
|
||||
self,
|
||||
key: str,
|
||||
session_number: int,
|
||||
server: str,
|
||||
proxy_host: str,
|
||||
proxy_port: int,
|
||||
):
|
||||
"""
|
||||
Initialize connection to REST server.
|
||||
"""
|
||||
self.key = key
|
||||
|
||||
self.connect_time = (
|
||||
int(datetime.now().strftime("%y%m%d%H%M%S")) * self.order_count
|
||||
)
|
||||
|
||||
if server == "REAL":
|
||||
self.init(HOST, proxy_host, proxy_port)
|
||||
else:
|
||||
self.init(TEST_HOST, proxy_host, proxy_port)
|
||||
|
||||
self.start(session_number)
|
||||
|
||||
self.gateway.write_log(_("REST API启动成功"))
|
||||
|
||||
def _new_order_id(self) -> str:
|
||||
""""""
|
||||
with self.order_count_lock:
|
||||
self.order_count += 1
|
||||
return f'a{self.connect_time}{self.order_count}'
|
||||
|
||||
@staticmethod
|
||||
def is_local_order_id(order_id: str):
|
||||
return order_id[0] == 'a'
|
||||
|
||||
def send_order(self, req: OrderRequest):
|
||||
""""""
|
||||
req.offset = Offset.NONE
|
||||
order_id = self._new_order_id()
|
||||
|
||||
symbol = req.symbol
|
||||
vol = int(req.volume)
|
||||
direction = req.direction
|
||||
data = {
|
||||
"instrument": symbol,
|
||||
# positive for long , negative for short
|
||||
"units": vol if direction is Direction.LONG else -vol,
|
||||
"clientExtensions": {
|
||||
"id": order_id,
|
||||
}
|
||||
}
|
||||
|
||||
order = req.create_order_data(order_id, self.gateway_name)
|
||||
order.time = parse_time(datetime.now().isoformat())
|
||||
|
||||
# Only add price for limit order.
|
||||
str_price = str(req.price)
|
||||
data["type"] = ORDER_TYPE_VT2OANDA[req.type]
|
||||
data["price"] = str(req.price)
|
||||
self.gateway.orders[order.orderid] = order
|
||||
self.add_request(
|
||||
"POST",
|
||||
f"/v3/accounts/{self.gateway.account_id}/orders",
|
||||
callback=self.on_send_order,
|
||||
data={'order': data},
|
||||
extra=order,
|
||||
on_failed=self.on_send_order_failed,
|
||||
on_error=self.on_send_order_error,
|
||||
)
|
||||
self.gateway.on_order(order)
|
||||
return order.vt_orderid
|
||||
|
||||
def cancel_order(self, req: CancelRequest):
|
||||
""""""
|
||||
order_id = req.orderid
|
||||
order = self.gateway.orders[order_id]
|
||||
|
||||
if self.is_local_order_id(order_id):
|
||||
order_id = '@' + order_id
|
||||
self.add_request(
|
||||
"PUT",
|
||||
f"/v3/accounts/{self.gateway.account_id}/orders/{order_id}/cancel",
|
||||
callback=self.on_cancel_order,
|
||||
extra=order,
|
||||
)
|
||||
|
||||
def query_history(self,
|
||||
symbol: str,
|
||||
interval: Interval,
|
||||
start: datetime,
|
||||
limit: int = None,
|
||||
) -> Tuple[List[BarData], "HistoryDataNextInfo"]:
|
||||
"""
|
||||
Get history data synchronously.
|
||||
"""
|
||||
if limit is None:
|
||||
limit = self.gateway.HISTORY_RECORD_PER_REQUEST
|
||||
|
||||
bars = []
|
||||
|
||||
# datetime for a bar is close_time
|
||||
# we got open_time from API.
|
||||
adjustment = INTERVAL_VT2OANDA_DELTA[interval]
|
||||
|
||||
# todo: RestClient: return RestClient.Request object instead of requests.Response.
|
||||
resp = self.request(
|
||||
"GET",
|
||||
f"/v3/instruments/{symbol}/candles",
|
||||
params={
|
||||
# "price": "M", # M for mids, B for bids, A for asks
|
||||
"granularity": INTERVAL_VT2OANDA[interval],
|
||||
"count": 5000,
|
||||
"from": start.isoformat(),
|
||||
}
|
||||
)
|
||||
|
||||
# Break if request failed with other status code
|
||||
raw_data = resp.json()
|
||||
# noinspection PyTypeChecker
|
||||
if not self.is_request_success(raw_data, None):
|
||||
msg = f"获取历史数据失败，状态码：{resp.status_code}，信息：{resp.text}"
|
||||
self.gateway.write_log(msg)
|
||||
raise RequestFailedException(msg)
|
||||
result = raw_data['candles']
|
||||
for data in result:
|
||||
bar_data = data['mid']
|
||||
open_time = parse_datetime(data["time"])
|
||||
close_time = open_time + adjustment
|
||||
bar = BarData(
|
||||
symbol=symbol,
|
||||
exchange=Exchange.OANDA,
|
||||
datetime=close_time,
|
||||
interval=interval,
|
||||
volume=data["volume"],
|
||||
open_price=float(bar_data["o"]),
|
||||
high_price=float(bar_data["h"]),
|
||||
low_price=float(bar_data["l"]),
|
||||
close_price=float(bar_data["c"]),
|
||||
gateway_name=self.gateway_name
|
||||
)
|
||||
bars.append(bar)
|
||||
|
||||
end = bars[-1].datetime.replace(tzinfo=utc_tz).astimezone(tz=local_tz)
|
||||
return bars, HistoryDataNextInfo(symbol, interval, end)
|
||||
|
||||
def on_send_order_failed(self, status_code: int, request: Request):
|
||||
"""
|
||||
Callback when sending order failed on server.
|
||||
"""
|
||||
data = request.response.json()
|
||||
|
||||
order = request.extra
|
||||
order.status = Status.REJECTED
|
||||
self.gateway.on_order(order)
|
||||
|
||||
msg = f"委托失败，错误代码:{data.get('errorCode', '')}, 错误信息：{data['errorMessage']}"
|
||||
self.gateway.write_log(msg)
|
||||
|
||||
def on_send_order_error(
|
||||
self, exception_type: type, exception_value: Exception, tb, request: Request
|
||||
):
|
||||
"""
|
||||
Callback when sending order caused exception.
|
||||
"""
|
||||
order = request.extra
|
||||
order.status = Status.REJECTED
|
||||
self.gateway.on_order(order)
|
||||
|
||||
# Record exception if not ConnectionError
|
||||
if not issubclass(exception_type, ConnectionError):
|
||||
self.on_error(exception_type, exception_value, tb, request)
|
||||
|
||||
def on_send_order(self, raw_data: dict, request: Request):
|
||||
""""""
|
||||
order: OrderData = request.extra
|
||||
# creation = raw_data.get('orderCreateTransaction', None)
|
||||
# if creation is not None:
|
||||
# order.status = Status.NOTTRADED
|
||||
# order.time = creation['time'][11:19]
|
||||
# self.gateway.on_order(order)
|
||||
|
||||
# cancel = raw_data.get('orderCancelTransaction', None)
|
||||
# if cancel is not None:
|
||||
# # potential bug: stream API will generate a Status.Cancel Order for this transaction
|
||||
# order.status = Status.REJECTED
|
||||
# order.time = parse_time(cancel['time'])
|
||||
pass
|
||||
|
||||
def on_cancel_order(self, raw_data: dict, request: Request):
|
||||
""""""
|
||||
# order: OrderData = request.extra
|
||||
# order.status = Status.CANCELLED
|
||||
# order.time = raw_data['orderCancelTransaction']['time'][11:19]
|
||||
# self.gateway.on_order(order)
|
||||
pass
|
||||
|
||||
def on_failed(self, status_code: int, request: Request):
|
||||
"""
|
||||
Callback to handle request failed.
|
||||
"""
|
||||
data = request.response.json()
|
||||
self._handle_error_response(data, request)
|
||||
|
||||
def _handle_error_response(self, data, request, operation_name: str = None):
|
||||
if operation_name is None:
|
||||
operation_name = request.path
|
||||
|
||||
# todo: rate limit?
|
||||
|
||||
error_msg = data.get("message", None)
|
||||
if error_msg is None:
|
||||
error_msg = data.get("errorMessage", None)
|
||||
msg = f"请求{operation_name}失败，状态码：{request.status}，错误消息：{error_msg}"
|
||||
msg += f'\n{request}'
|
||||
self.gateway.write_log(msg)
|
||||
|
||||
def query_positions(self):
|
||||
self.add_request("GET",
|
||||
f"/v3/accounts/{self.gateway.account_id}/positions",
|
||||
self.on_query_positions
|
||||
)
|
||||
|
||||
def query_account_changes(self):
|
||||
do_nothing = lambda a, b, c, d: None # noqa
|
||||
|
||||
if self.last_account_transaction_id is not None:
|
||||
account_id = self.gateway.account_id
|
||||
self.add_request("GET",
|
||||
f"/v3/accounts/{account_id}/changes",
|
||||
params={
|
||||
"sinceTransactionID": self.last_account_transaction_id
|
||||
},
|
||||
callback=self.on_query_account_changes,
|
||||
extra=account_id,
|
||||
on_error=do_nothing,
|
||||
)
|
||||
|
||||
def on_query_account_changes(self, raw_data: dict, request: "Request"):
|
||||
account_id: str = request.extra
|
||||
|
||||
# state: we focus mainly on account balance
|
||||
state = raw_data['state']
|
||||
NAV = float(state['NAV'])
|
||||
pnl = float(state['unrealizedPL'])
|
||||
account = AccountData(
|
||||
gateway_name=self.gateway_name,
|
||||
accountid=account_id,
|
||||
balance=NAV - pnl,
|
||||
frozen=0, # no frozen
|
||||
)
|
||||
self.gateway.on_account(account)
|
||||
|
||||
# changes: we focus mainly on position changes
|
||||
changes = raw_data['changes']
|
||||
positions = changes['positions']
|
||||
unrealized_pnls: Dict[(str, Direction), float] = {}
|
||||
|
||||
# pnl in query_account_changes is different from data returned by query_account
|
||||
# we have to get pnl from 'state' record.
|
||||
for pos_state_data in state['positions']:
|
||||
symbol = pos_state_data['instrument']
|
||||
unrealized_pnls[(symbol, Direction.LONG)] = float(pos_state_data['longUnrealizedPL'])
|
||||
unrealized_pnls[(symbol, Direction.SHORT)] = float(pos_state_data['shortUnrealizedPL'])
|
||||
|
||||
for pos_data in positions:
|
||||
pos_long, pos_short = self.parse_position_data(pos_data)
|
||||
symbol = pos_long.symbol
|
||||
pos_long.pnl = unrealized_pnls[(symbol, Direction.LONG)]
|
||||
pos_short.pnl = unrealized_pnls[(symbol, Direction.SHORT)]
|
||||
self.gateway.on_position(pos_long)
|
||||
self.gateway.on_position(pos_short)
|
||||
self.last_account_transaction_id = raw_data['lastTransactionID']
|
||||
|
||||
def on_query_positions(self, raw_data: dict, request: "Request"):
|
||||
for pos_data in raw_data['positions']:
|
||||
pos_long, pos_short = self.parse_position_data(pos_data)
|
||||
self.gateway.on_position(pos_long)
|
||||
self.gateway.on_position(pos_short)
|
||||
|
||||
def parse_position_data(self, pos_data) -> Tuple[PositionData, PositionData]:
|
||||
symbol = pos_data['instrument']
|
||||
pos_long, pos_short = [PositionData(
|
||||
gateway_name=self.gateway_name,
|
||||
symbol=symbol,
|
||||
exchange=Exchange.OANDA,
|
||||
direction=direction,
|
||||
volume=int(data['units']),
|
||||
price=float(data.get('averagePrice', 0.0)),
|
||||
pnl=float(data.get('unrealizedPL', 0.0)),
|
||||
)
|
||||
for direction, data in (
|
||||
(Direction.LONG, pos_data['long']),
|
||||
(Direction.SHORT, pos_data['short'])
|
||||
)
|
||||
]
|
||||
return pos_long, pos_short
|
||||
|
||||
def query_orders(self):
|
||||
"""
|
||||
query all orders, including stop orders
|
||||
"""
|
||||
self.add_request("GET",
|
||||
f"/v3/accounts/{self.gateway.account_id}/orders?state=ALL",
|
||||
callback=self.on_query_orders,
|
||||
)
|
||||
|
||||
def on_query_orders(self, raw_data: dict, request: "Request"):
|
||||
for data in raw_data['orders']:
|
||||
order = self.gateway.parse_order_data(data,
|
||||
STATUS_OANDA2VT[data['state']],
|
||||
'createTime')
|
||||
self.gateway.on_order(order)
|
||||
self.orders_initialized = True
|
||||
|
||||
def query_accounts(self):
|
||||
self.add_request("GET",
|
||||
"/v3/accounts",
|
||||
callback=self.on_query_accounts,
|
||||
)
|
||||
|
||||
def on_query_accounts(self, raw_data: dict, request: "Request"):
|
||||
"""
|
||||
{"accounts":[{"id":"101-001-12185735-001","tags":[]}]}
|
||||
"""
|
||||
for acc in raw_data['accounts']:
|
||||
account_id: str = acc['id']
|
||||
self.gateway.account_id = account_id
|
||||
|
||||
self.query_account(account_id)
|
||||
self.query_contracts(account_id)
|
||||
self.query_orders()
|
||||
# self.query_positions()
|
||||
|
||||
def query_account(self, account_id: str):
|
||||
self.add_request("GET",
|
||||
f"/v3/accounts/{account_id}",
|
||||
callback=self.on_query_account,
|
||||
)
|
||||
|
||||
def on_query_account(self, raw_data: dict, request: "Request"):
|
||||
acc = raw_data['account']
|
||||
account_data = AccountData(
|
||||
gateway_name=self.gateway_name,
|
||||
accountid=acc['id'],
|
||||
balance=float(acc['balance']),
|
||||
frozen=0, # no frozen
|
||||
)
|
||||
self.gateway.on_account(account_data)
|
||||
for data in acc['orders']:
|
||||
# order = self.parse_order_data(data)
|
||||
# self.gateway.on_order(order)
|
||||
pass
|
||||
for pos_data in acc['positions']:
|
||||
pos_long, pos_short = self.parse_position_data(pos_data)
|
||||
self.gateway.on_position(pos_long)
|
||||
self.gateway.on_position(pos_short)
|
||||
for trade in acc['trades']:
|
||||
pass
|
||||
self.last_account_transaction_id = acc['lastTransactionID']
|
||||
self.gateway.stream_api.subscribe_transaction()
|
||||
self.account_initialized = True
|
||||
|
||||
@property
|
||||
def fully_initialized(self):
|
||||
return self.account_initialized and self.orders_initialized
|
||||
|
||||
def query_contracts(self, account_id):
|
||||
self.add_request("GET",
|
||||
f"/v3/accounts/{account_id}/instruments",
|
||||
self.on_query_contracts)
|
||||
|
||||
def on_query_contracts(self, data: dict, request: "Request"):
|
||||
for ins in data['instruments']:
|
||||
symbol = ins['name']
|
||||
contract = ContractData(
|
||||
gateway_name=self.gateway_name,
|
||||
symbol=symbol,
|
||||
exchange=Exchange.OANDA,
|
||||
name=symbol,
|
||||
product=Product.FOREX,
|
||||
size=1,
|
||||
pricetick=1.0 / pow(10, ins['displayPrecision']),
|
||||
)
|
||||
self.gateway.on_contract(contract)
|
||||
self.contracts[symbol] = contract
|
||||
self.gateway.write_log(_("合约信息查询成功"))
|
||||
228
vnpy/gateway/oanda/oanda_stream_api.py
Normal file
228
vnpy/gateway/oanda/oanda_stream_api.py
Normal file
@ -0,0 +1,228 @@
|
||||
from copy import copy
|
||||
from copy import copy
|
||||
from dataclasses import dataclass
|
||||
from functools import partial
|
||||
from http.client import IncompleteRead, RemoteDisconnected
|
||||
from typing import Callable, TYPE_CHECKING, Type
|
||||
|
||||
from urllib3.exceptions import ProtocolError
|
||||
|
||||
from vnpy.api.rest import Request
|
||||
from vnpy.trader.constant import Exchange, Interval, Offset, Status
|
||||
from vnpy.trader.object import OrderData, SubscribeRequest, TickData, TradeData
|
||||
from .oanda_api_base import OandaApiBase
|
||||
from .oanda_common import (parse_datetime, parse_time)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from vnpy.gateway.oanda import OandaGateway
|
||||
_ = lambda x: x # noqa
|
||||
|
||||
HOST = "https://stream-fxtrade.oanda.com"
|
||||
TEST_HOST = "https://stream-fxpractice.oanda.com"
|
||||
|
||||
# asked from official developer
|
||||
PRICE_TICKS = {
|
||||
"BTCUSD": 0.5,
|
||||
"ETHUSD": 0.05,
|
||||
"EOSUSD": 0.001,
|
||||
"XRPUSD": 0.0001,
|
||||
}
|
||||
|
||||
|
||||
@dataclass()
|
||||
class HistoryDataNextInfo:
|
||||
symbol: str
|
||||
interval: Interval
|
||||
end: int
|
||||
|
||||
|
||||
class RequestFailedException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class OandaStreamApi(OandaApiBase):
|
||||
"""
|
||||
Oanda Streaming API
|
||||
"""
|
||||
|
||||
def __init__(self, gateway: "OandaGateway"):
|
||||
""""""
|
||||
super().__init__(gateway)
|
||||
|
||||
self.fully_initialized = False
|
||||
|
||||
self._transaction_callbacks = {
|
||||
'ORDER_FILL': self.on_order_filled,
|
||||
'MARKET_ORDER': self.on_order,
|
||||
'LIMIT_ORDER': self.on_order,
|
||||
'STOP_ORDER': self.on_order,
|
||||
'ORDER_CANCEL': self.on_order_canceled,
|
||||
|
||||
# 'HEARTBEAT': do_nothing,
|
||||
}
|
||||
|
||||
def connect(
|
||||
self,
|
||||
key: str,
|
||||
session_number: int,
|
||||
server: str,
|
||||
proxy_host: str,
|
||||
proxy_port: int,
|
||||
):
|
||||
"""
|
||||
Initialize connection to REST server.
|
||||
"""
|
||||
self.key = key
|
||||
|
||||
if server == "REAL":
|
||||
self.init(HOST, proxy_host, proxy_port)
|
||||
else:
|
||||
self.init(TEST_HOST, proxy_host, proxy_port)
|
||||
|
||||
self.start(session_number)
|
||||
|
||||
self.gateway.write_log(_("Streaming API启动成功"))
|
||||
|
||||
def subscribe(self, req: SubscribeRequest):
|
||||
# noinspection PyTypeChecker
|
||||
self.add_streaming_request(
|
||||
"GET",
|
||||
f"/v3/accounts/{self.gateway.account_id}/pricing/stream?instruments={req.symbol}",
|
||||
callback=self.on_price,
|
||||
on_error=partial(self.on_streaming_error, partial(self.subscribe, copy(req))),
|
||||
)
|
||||
|
||||
def on_price(self, data: dict, request: Request):
|
||||
type_ = data['type']
|
||||
if type_ == 'PRICE':
|
||||
symbol = data['instrument']
|
||||
# only one level of bids/asks
|
||||
bid = data['bids'][0]
|
||||
ask = data['asks'][0]
|
||||
tick = TickData(
|
||||
gateway_name=self.gateway_name,
|
||||
symbol=symbol,
|
||||
exchange=Exchange.OANDA,
|
||||
datetime=parse_datetime(data['time']),
|
||||
name=symbol,
|
||||
bid_price_1=float(bid['price']),
|
||||
bid_volume_1=bid['liquidity'],
|
||||
ask_price_1=float(ask['price']),
|
||||
ask_volume_1=ask['liquidity'],
|
||||
)
|
||||
self.gateway.on_tick(tick)
|
||||
|
||||
def has_error(self, target_type: Type[Exception], e: Exception):
|
||||
"""check if error type \a target_error exists inside \a e"""
|
||||
if isinstance(e, target_type):
|
||||
return True
|
||||
for arg in e.args:
|
||||
if isinstance(arg, Exception) and self.has_error(target_type, arg):
|
||||
return True
|
||||
return False
|
||||
|
||||
def on_streaming_error(self,
|
||||
re_subscribe: Callable,
|
||||
exception_type: type,
|
||||
exception_value: Exception,
|
||||
tb,
|
||||
request: Request,
|
||||
):
|
||||
"""normally triggered by network error."""
|
||||
# skip known errors
|
||||
known = False
|
||||
for et in (ProtocolError, IncompleteRead, RemoteDisconnected,):
|
||||
if self.has_error(et, exception_value):
|
||||
known = True
|
||||
break
|
||||
|
||||
if known:
|
||||
# re-subscribe
|
||||
re_subscribe()
|
||||
# write log for any unknown errors
|
||||
else:
|
||||
super().on_error(exception_type, exception_value, tb, request)
|
||||
|
||||
def subscribe_transaction(self):
|
||||
# noinspection PyTypeChecker
|
||||
self.add_streaming_request(
|
||||
"GET",
|
||||
f"/v3/accounts/{self.gateway.account_id}/transactions/stream",
|
||||
callback=self.on_transaction,
|
||||
on_connected=self.on_subscribed_transaction,
|
||||
on_error=partial(self.on_streaming_error, partial(self.subscribe_transaction, )),
|
||||
)
|
||||
|
||||
def on_subscribed_transaction(self, request: "Request"):
|
||||
self.fully_initialized = True
|
||||
|
||||
def on_transaction(self, data: dict, request: "Request"):
|
||||
type_ = data['type']
|
||||
callback = self._transaction_callbacks.get(type_, None)
|
||||
if callback is not None:
|
||||
callback(data, request)
|
||||
elif type_ != "HEARTBEAT":
|
||||
print(type_)
|
||||
|
||||
def on_order(self, data: dict, request: "Request"):
|
||||
order = self.gateway.parse_order_data(data,
|
||||
Status.NOTTRADED,
|
||||
'time',
|
||||
)
|
||||
self.gateway.on_order(order)
|
||||
|
||||
def on_order_canceled(self, data: dict, request: "Request"):
|
||||
order_id = data.get('clientOrderID', None)
|
||||
if order_id is None:
|
||||
order_id = data['id']
|
||||
order = self.gateway.orders[order_id]
|
||||
order.status = Status.CANCELLED
|
||||
order.time = parse_time(data['time'])
|
||||
self.gateway.on_order(order)
|
||||
|
||||
def on_order_filled(self, data: dict, request: "Request"):
|
||||
order_id = data.get('clientOrderID', None)
|
||||
if order_id is None:
|
||||
order_id = data['orderID']
|
||||
|
||||
order: OrderData = self.gateway.orders[order_id]
|
||||
|
||||
# # new API:
|
||||
# price = 0.0
|
||||
# if 'tradeOpened' in data:
|
||||
# price += float(data['tradeOpened']['price'])
|
||||
# if 'tradeReduced' in data:
|
||||
# price += float(data['tradeReduced']['price'])
|
||||
# if 'tradeClosed' in data:
|
||||
# price += sum([float(i['price']) for i in data['tradeClosed']])
|
||||
|
||||
# note: 'price' record is Deprecated
|
||||
# but since this is faster and easier, we use this record.
|
||||
price = float(data['price'])
|
||||
|
||||
# for Oanda, one order fill is a single trade.
|
||||
trade = TradeData(
|
||||
gateway_name=self.gateway_name,
|
||||
symbol=order.symbol,
|
||||
exchange=Exchange.OANDA,
|
||||
orderid=order_id,
|
||||
tradeid=order_id,
|
||||
direction=order.direction,
|
||||
offset=Offset.NONE,
|
||||
price=price,
|
||||
volume=order.volume,
|
||||
time=parse_time(data['time']),
|
||||
)
|
||||
self.gateway.on_trade(trade)
|
||||
|
||||
# this message indicate that this order is full filled.
|
||||
# ps: oanda's order has only two state: NOTTRADED, ALLTRADED. It it my settings error?
|
||||
order.traded = order.volume
|
||||
order.status = Status.ALLTRADED
|
||||
# order.time = trade.time
|
||||
order.time = parse_time(data['time'])
|
||||
self.gateway.on_order(order)
|
||||
|
||||
# quick references
|
||||
on_tick = on_price
|
||||
on_trade = on_order_filled
|
||||
@ -109,6 +109,8 @@ class Exchange(Enum):
|
||||
EUNX = "EUNX" # Euronext Exchange
|
||||
KRX = "KRX" # Korean Exchange
|
||||
|
||||
OANDA = "OANDA" # oanda.com
|
||||
|
||||
# CryptoCurrency
|
||||
BITMEX = "BITMEX"
|
||||
OKEX = "OKEX"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user