senset/vnpy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[Mod] 统一格式，隐藏内部变量

Browse Source
This commit is contained in:
nanoric 2019-03-08 07:26:04 -04:00
parent 637dabd406
commit eb5aac2595
3 changed files with 106 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
vnpy/gateway/oes/oes_gateway.py
View File

@ -38,6 +38,7 @@ class OesGateway(BaseGateway):
self.td_api = OesTdApi(self)
def connect(self, setting: dict):
""""""
return self._connect_async(setting)
def _connect_sync(self, setting: dict):
@ -89,6 +90,7 @@ class OesGateway(BaseGateway):
self.write_log(_("无法连接到交易服务器，请检查你的配置"))
def _connect_async(self, setting: dict):
""""""
Thread(target=self._connect_sync, args=(setting,)).start()
def subscribe(self, req: SubscribeRequest):

79
vnpy/gateway/oes/oes_md.py
View File

@ -9,7 +9,7 @@ from vnpy.api.oes.vnoes import MdsApiClientEnvT, MdsApi_DestoryAll, MdsApi_InitA
MdsApi_IsValidQryChannel, MdsApi_IsValidTcpChannel, MdsApi_LogoutAll, MdsApi_SetThreadPassword, \
MdsApi_SetThreadUsername, MdsApi_SubscribeMarketData, MdsApi_WaitOnMsg, MdsL2StockSnapshotBodyT, \
MdsMktDataRequestEntryT, MdsMktDataRequestReqT, MdsMktRspMsgBodyT, MdsStockSnapshotBodyT, \
SGeneralClientChannelT, SMsgHeadT, SPlatform_IsNegEpipe, SPlatform_IsNegEtimeout, cast, \
SGeneralClientChannelT, SMsgHeadT, SPlatform_IsNegEpipe, cast, \
eMdsExchangeIdT, eMdsMktSubscribeFlagT, eMdsMsgTypeT, eMdsSecurityTypeT, eMdsSubscribeDataTypeT, \
eMdsSubscribeModeT, eMdsSubscribedTickExpireTypeT, eSMsgProtocolTypeT
@ -28,11 +28,12 @@ EXCHANGE_VT2MDS = {v: k for k, v in EXCHANGE_MDS2VT.items()}
class OesMdMessageLoop:
def __init__(self, gateway: BaseGateway, md: "OesMdApi", env: MdsApiClientEnvT):
""""""
self.gateway = gateway
self.env = env
self.alive = False
self.md = md
self.th = Thread(target=self.message_loop)
self.th = Thread(target=self._message_loop)
self.message_handlers: Dict[int, Callable[[dict], None]] = {
# tick & orderbook
@ -58,9 +59,29 @@ class OesMdMessageLoop:
self.symbol_to_exchange: Dict[str, Exchange] = {}
def register_symbol(self, symbol: str, exchange: Exchange):
""""""
self.symbol_to_exchange[symbol] = exchange
def get_last_tick(self, symbol):
def start(self):
""""""
self.alive = True
self.th.start()
def stop(self):
""""""
self.alive = False
def join(self):
""""""
self.th.join()
def reconnect(self):
""""""
self.gateway.write_log(_("正在尝试重新连接到行情服务器。"))
return self.md.connect()
def _get_last_tick(self, symbol):
""""""
try:
return self.last_tick[symbol]
except KeyError:
@ -68,25 +89,15 @@ class OesMdMessageLoop:
gateway_name=self.gateway.gateway_name,
symbol=symbol,
exchange=self.symbol_to_exchange[symbol],
# todo: use cache of something else to resolve exchange
datetime=datetime.utcnow()
)
self.last_tick[symbol] = tick
return tick
def start(self):
self.alive = True
self.th.start()
def stop(self):
self.alive = False
def join(self):
self.th.join()
def on_message(self, session_info: SGeneralClientChannelT,
head: SMsgHeadT,
body: Any):
def _on_message(self, session_info: SGeneralClientChannelT,
head: SMsgHeadT,
body: Any):
""""""
if session_info.protocolType == eSMsgProtocolTypeT.SMSG_PROTO_BINARY:
b = cast.toMdsMktRspMsgBodyT(body)
if head.msgId in self.message_handlers:
@ -100,19 +111,16 @@ class OesMdMessageLoop:
self.gateway.write_log(f"unknown prototype : {session_info.protocolType}")
return 1
def reconnect(self):
self.gateway.write_log(_("正在尝试重新连接到行情服务器。"))
return self.md.connect()
def message_loop(self):
def _message_loop(self):
""""""
tcp_channel = self.env.tcpChannel
timeout_ms = 1000
is_timeout = SPlatform_IsNegEtimeout
# is_timeout = SPlatform_IsNegEtimeout
is_disconnected = SPlatform_IsNegEpipe
while self.alive:
ret = MdsApi_WaitOnMsg(tcp_channel,
timeout_ms,
self.on_message)
self._on_message)
if ret < 0:
# if is_timeout(ret):
# pass # just no message
@ -123,9 +131,10 @@ class OesMdMessageLoop:
return
def on_l2_market_data_snapshot(self, d: MdsMktRspMsgBodyT):
""""""
data: MdsL2StockSnapshotBodyT = d.mktDataSnapshot.l2Stock
symbol = str(data.SecurityID)
tick = self.get_last_tick(symbol)
tick = self._get_last_tick(symbol)
tick.open_price = data.OpenPx / 10000
tick.pre_close = data.ClosePx / 10000
tick.high_price = data.HighPx / 10000
@ -138,9 +147,10 @@ class OesMdMessageLoop:
self.gateway.on_tick(tick)
def on_market_full_refresh(self, d: MdsMktRspMsgBodyT):
""""""
data: MdsStockSnapshotBodyT = d.mktDataSnapshot.stock
symbol = data.SecurityID
tick = self.get_last_tick(symbol)
tick = self._get_last_tick(symbol)
tick.open_price = data.OpenPx / 10000
tick.pre_close = data.ClosePx / 10000
tick.high_price = data.HighPx / 10000
@ -153,42 +163,52 @@ class OesMdMessageLoop:
self.gateway.on_tick(tick)
def on_l2_trade(self, d: MdsMktRspMsgBodyT):
""""""
data = d.trade
symbol = data.SecurityID
tick = self.get_last_tick(symbol)
tick = self._get_last_tick(symbol)
tick.datetime = datetime.utcnow()
tick.volume = data.TradeQty
tick.last_price = data.TradePrice / 10000
self.gateway.on_tick(tick)
def on_market_data_request(self, d: MdsMktRspMsgBodyT):
""""""
pass
def on_trading_session_status(self, d: MdsMktRspMsgBodyT):
""""""
pass
def on_l2_market_overview(self, d: MdsMktRspMsgBodyT):
""""""
pass
def on_index_snapshot_full_refresh(self, d: MdsMktRspMsgBodyT):
""""""
pass
def on_option_snapshot_ful_refresh(self, d: MdsMktRspMsgBodyT):
""""""
pass
def on_best_orders_snapshot(self, d: MdsMktRspMsgBodyT):
""""""
pass
def on_l2_order(self, d: MdsMktRspMsgBodyT):
""""""
pass
def on_security_status(self, d: MdsMktRspMsgBodyT):
""""""
pass
class OesMdApi:
def __init__(self, gateway: BaseGateway):
""""""
self.gateway = gateway
self.config_path: str = ''
self.username: str = ''
@ -198,6 +218,7 @@ class OesMdApi:
self._message_loop = OesMdMessageLoop(gateway, self, self._env)
def connect(self) -> bool:
""""""
"""Connect to trading server.
:note set config_path before calling this function
"""
@ -214,18 +235,22 @@ class OesMdApi:
return True
def start(self):
""""""
self._message_loop.start()
def stop(self):
""""""
self._message_loop.stop()
MdsApi_LogoutAll(self._env, True)
MdsApi_DestoryAll(self._env)
def join(self):
""""""
self._message_loop.join()
# why isn't arg a ContractData?
def subscribe(self, req: SubscribeRequest):
""""""
mds_req = MdsMktDataRequestReqT()
entry = MdsMktDataRequestEntryT()
mds_req.subMode = eMdsSubscribeModeT.MDS_SUB_MODE_APPEND

65
vnpy/gateway/oes/oes_td.py
View File

@ -3,7 +3,7 @@ from datetime import datetime, timedelta, timezone
from gettext import gettext as _
from threading import Lock, Thread
# noinspection PyUnresolvedReferences
from typing import Any, Callable, Dict, Tuple
from typing import Any, Callable, Dict
from vnpy.api.oes.vnoes import OesApiClientEnvT, OesApi_DestoryAll, OesApi_InitAllByConvention, \
OesApi_IsValidOrdChannel, OesApi_IsValidQryChannel, OesApi_IsValidRptChannel, OesApi_LogoutAll, \
@ -95,6 +95,7 @@ class InternalOrder:
def parse_oes_datetime(date: int, time: int):
"""convert oes datetime to python datetime"""
# YYYYMMDD
year = int(date / 10000)
month = int((date % 10000) / 100)
@ -115,13 +116,14 @@ class OesTdMessageLoop:
env: OesApiClientEnvT,
td: "OesTdApi"
):
""""""
self.gateway = gateway
self._env = env
self._td = td
self._alive = False
self._th = Thread(target=self.message_loop)
self._th = Thread(target=self._message_loop)
self.message_handlers: Dict[int, Callable[[dict], None]] = {
eOesMsgTypeT.OESMSG_RPT_BUSINESS_REJECT: self.on_order_rejected,
@ -137,18 +139,27 @@ class OesTdMessageLoop:
}
def start(self):
""""""
self._alive = True
self._th.start()
def stop(self):
""""""
self._alive = False
def join(self):
""""""
self._th.join()
def on_message(self, session_info: SGeneralClientChannelT,
head: SMsgHeadT,
body: Any):
def reconnect(self):
""""""
self.gateway.write_log(_("正在尝试重新连接到交易服务器。"))
self._td.connect()
def _on_message(self, session_info: SGeneralClientChannelT,
head: SMsgHeadT,
body: Any):
""""""
if session_info.protocolType == SMSG_PROTO_BINARY:
b = cast.toOesRspMsgBodyT(body)
if head.msgId in self.message_handlers:
@ -160,11 +171,8 @@ class OesTdMessageLoop:
self.gateway.write_log(f"unknown prototype : {session_info.protocolType}")
return 1
def reconnect(self):
self.gateway.write_log(_("正在尝试重新连接到交易服务器。"))
self._td.connect()
def message_loop(self):
def _message_loop(self):
""""""
rpt_channel = self._env.rptChannel
timeout_ms = 1000
is_disconnected = SPlatform_IsNegEpipe
@ -172,7 +180,7 @@ class OesTdMessageLoop:
while self._alive:
ret = OesApi_WaitReportMsg(rpt_channel,
timeout_ms,
self.on_message)
self._on_message)
if ret < 0:
# if is_timeout(ret):
# pass # just no message
@ -183,6 +191,7 @@ class OesTdMessageLoop:
return
def on_order_rejected(self, d: OesRspMsgBodyT):
""""""
error_code = d.rptMsg.rptHead.ordRejReason
error_string = error_to_str(error_code)
data: OesOrdRejectT = d.rptMsg.rptBody.ordRejectRsp
@ -202,6 +211,7 @@ class OesTdMessageLoop:
self.gateway.write_log(f"撤单失败，订单号： {data.origClSeqNo}。原因：{error_string}")
def on_order_inserted(self, d: OesRspMsgBodyT):
""""""
data = d.rptMsg.rptBody.ordInsertRsp
if not data.origClSeqNo:
@ -217,6 +227,7 @@ class OesTdMessageLoop:
self.gateway.on_order(vt_order)
def on_order_report(self, d: OesRspMsgBodyT):
""""""
data: OesOrdCnfmT = d.rptMsg.rptBody.ordCnfm
if not data.origClSeqNo:
@ -233,6 +244,7 @@ class OesTdMessageLoop:
self.gateway.on_order(vt_order)
def on_trade_report(self, d: OesRspMsgBodyT):
""""""
data: OesTrdCnfmT = d.rptMsg.rptBody.trdCnfm
i = self._td.get_order(data.clSeqNo)
@ -258,9 +270,11 @@ class OesTdMessageLoop:
self.gateway.on_order(vt_order)
def on_option_holding(self, d: OesRspMsgBodyT):
""""""
pass
def on_stock_holding(self, d: OesRspMsgBodyT):
""""""
data = d.rptMsg.rptBody.stkHoldingRpt
position = PositionData(
gateway_name=self.gateway.gateway_name,
@ -277,6 +291,7 @@ class OesTdMessageLoop:
self.gateway.on_position(position)
def on_cash(self, d: OesRspMsgBodyT):
""""""
data = d.rptMsg.rptBody.cashAssetRpt
balance = data.currentTotalBal
@ -296,6 +311,7 @@ class OesTdMessageLoop:
class OesTdApi:
def __init__(self, gateway: BaseGateway):
""""""
self.config_path: str = None
self.username: str = ''
self.password: str = ''
@ -309,7 +325,7 @@ class OesTdApi:
self._last_seq_lock = Lock()
self._last_seq_index = 1000000 # 0 has special manning for oes
self._orders: Dict[int, InternalOrder] = {}
def connect(self):
@ -333,14 +349,17 @@ class OesTdApi:
return True
def start(self):
""""""
self._message_loop.start()
def stop(self):
""""""
self._message_loop.stop()
OesApi_LogoutAll(self._env, True)
OesApi_DestoryAll(self._env)
def join(self):
""""""
self._message_loop.join()
def _get_new_seq_index(self):
@ -351,6 +370,7 @@ class OesTdApi:
return index
def query_account(self):
""""""
OesApi_QueryCashAsset(self._env.qryChannel,
OesQryCashAssetFilterT(),
self.on_query_asset
@ -362,6 +382,7 @@ class OesTdApi:
body: Any,
cursor: OesQryCursorT,
):
""""""
data = cast.toOesCashAssetItemT(body)
balance = data.currentTotalBal / 10000
availiable = data.currentAvailableBal / 10000
@ -377,10 +398,12 @@ class OesTdApi:
return 1
def query_stock(self, ) -> bool:
""""""
# Thread(target=self._query_stock, ).start()
return self._query_stock()
def _query_stock(self, ) -> bool:
""""""
f = OesQryStockFilterT()
ret = OesApi_QueryStock(self._env.qryChannel, f, self.on_query_stock)
return ret >= 0
@ -391,6 +414,7 @@ class OesTdApi:
body: Any,
cursor: OesQryCursorT,
):
""""""
data: OesStockBaseInfoT = cast.toOesStockItemT(body)
contract = ContractData(
gateway_name=self.gateway.gateway_name,
@ -405,6 +429,7 @@ class OesTdApi:
return 1
def query_option(self) -> bool:
""""""
f = OesQryOptionFilterT()
ret = OesApi_QueryOption(self._env.qryChannel,
f,
@ -418,6 +443,7 @@ class OesTdApi:
body: Any,
cursor: OesQryCursorT,
):
""""""
data = cast.toOesOptionItemT(body)
contract = ContractData(
gateway_name=self.gateway.gateway_name,
@ -432,6 +458,7 @@ class OesTdApi:
return 1
def query_stock_holding(self) -> bool:
""""""
f = OesQryStkHoldingFilterT()
ret = OesApi_QueryStkHolding(self._env.qryChannel,
f,
@ -445,6 +472,7 @@ class OesTdApi:
body: Any,
cursor: OesQryCursorT,
):
""""""
data = cast.toOesStkHoldingItemT(body)
position = PositionData(
@ -463,6 +491,7 @@ class OesTdApi:
return 1
def query_option_holding(self) -> bool:
""""""
f = OesQryStkHoldingFilterT()
f.mktId = eOesMarketIdT.OES_MKT_ID_UNDEFINE
f.userInfo = 0
@ -478,6 +507,7 @@ class OesTdApi:
body: Any,
cursor: OesQryCursorT,
):
""""""
data = cast.toOesOptHoldingItemT(body)
# 权利
@ -513,15 +543,18 @@ class OesTdApi:
return 1
def query_contracts(self):
""""""
self.query_stock()
# self.query_option()
# self.query_issue()
def query_position(self):
""""""
self.query_stock_holding()
self.query_option_holding()
def send_order(self, vt_req: OrderRequest):
""""""
seq_id = self._get_new_seq_index()
order_id = seq_id
@ -551,6 +584,7 @@ class OesTdApi:
return order.vt_orderid
def cancel_order(self, vt_req: CancelRequest):
""""""
seq_id = self._get_new_seq_index()
oes_req = OesOrdCancelReqT()
@ -565,11 +599,13 @@ class OesTdApi:
oes_req)
def schedule_query_order(self, internal_order: InternalOrder) -> Thread:
""""""
th = Thread(target=self.query_order, args=(internal_order,))
th.start()
return th
def query_order(self, internal_order: InternalOrder) -> bool:
""""""
f = OesQryOrdFilterT()
f.mktId = EXCHANGE_VT2OES[internal_order.vt_order.exchange]
f.clSeqNo = internal_order.order_id
@ -584,6 +620,7 @@ class OesTdApi:
head: SMsgHeadT,
body: Any,
cursor: OesQryCursorT):
""""""
data: OesOrdCnfmT = cast.toOesOrdItemT(body)
i = self._td.get_order(data.clSeqNo)
@ -612,6 +649,7 @@ class OesTdApi:
body: Any,
cursor: OesQryCursorT,
):
""""""
data: OesOrdCnfmT = cast.toOesOrdItemT(body)
try:
i = self.get_order(data.clSeqNo)
@ -649,11 +687,12 @@ class OesTdApi:
return 1
def save_order(self, order_id: int, order: OrderData):
""""""
self._orders[order_id] = InternalOrder(
order_id=order_id,
vt_order=order,
)
def get_order(self, order_id: int):
""""""
return self._orders[order_id]