元大期貨 交易API with python

下載位置： http://easywin.yuantafutures.com.tw/api/download.html 
(行情API需要另外簽署文件才可以使用 )

下載完成之後需要先註冊 (我是使用64位元)
64位元
API_x64/YuantaOrd64.ocx
32位元
API/YuantaOrd.ocx

程式說明：
目前只有做到登入/跟 財務查詢：EasyWin 0680 畫面
如果要做下單或其他功能 應該是差不多方式

執行需求：
交易API版本：1.6.1.3
python3
帳號, 密碼需修改

免責聲明： 提供程式以供參考，請勿用來販賣或用在商業用途上，如果以本程式用來交易，亦不負任何責任

	# -*- coding: utf-8 -*-
	import wx, time
	import wx.lib.anchors as anchors
	from ctypes import byref, POINTER, windll
	from comtypes import IUnknown, GUID
	from comtypes.client import GetModule, GetBestInterface, GetEvents
	user32 = windll.user32
	atl = windll.atl
	import queue as queue
	q = queue.Queue ()
	class switch(object):
	def __init__(self, value):
	self.value = value
	self.fall = False
	def __iter__(self):
	"""Return the match method once, then stop"""
	yield self.match
	raise StopIteration
	def match(self, *args):
	"""Indicate whether or not to enter a case suite"""
	if self.fall or not args:
	return True
	elif self.value in args: # changed for v1.5, see below
	self.fall = True
	return True
	else:
	return False
	class Job:
	STOCK_LOGIN = 1
	STOCK_USERDEFINE = 2
	def __init__ (self, do_type, do_value = 0):
	self.do_type = do_type
	self.do_value = do_value
	q.put (self)
	def DoJob(Bot, x):
	for case in switch(x.do_type):
	if case(Job.STOCK_LOGIN):
	Bot.login ()
	break
	if case(Job.STOCK_USERDEFINE):
	Bot.user_define ()
	break
	class YuantaOrdEvents(object):
	def __init__(self, parent):
	self.parent = parent
	def OnLogonS(self, this, TLinkStatus, AccList, Casq, Cast):
	print ('OnLogonS {},{},{},{}'.format (TLinkStatus, AccList, Casq, Cast))
	if TLinkStatus == 2:
	datas = AccList.split(';')
	for data in datas:
	self.parent.bot.YuantaAccount.append (data)
	self.parent.bot.YuantaSN.append (Casq.split('=')[1])
	Job (Job.STOCK_USERDEFINE)
	def OnReportQuery(self, this, RowCount, Results):
	print ('OnReportQuery')
	def OnDealQuery(self, this, RowCount, Results):
	print ('OnDealQuery')
	def OnUserDefinsFuncResult(self, this, RowCount, Results, WorkID):
	print ('OnUserDefinsFuncResult {},{},{}'.format(RowCount, WorkID, Results))
	def OnOrdRptF(self, this, Omkt, Mktt, Cmbf, Statusc, Ts_Code, Ts_Msg, Bhno, AcNo,
	Suba, Symb, Scnam, O_Kind, O_Type, Buys, S_Buys, O_Prc, O_Qty, Work_Qty, Kill_Qty,
	Deal_Qty, Order_No, T_Date, O_Date, O_Time,
	O_Src, O_Lin, A_Prc, Oseq_No, Err_Code,
	Err_Msg, R_Time, D_Flag):
	print ('OnOrdRptF')
	def OnOrdMatF(self, this, Omkt, Buys, Cmbf, Bhno,
	AcNo, Suba, Symb, Scnam, O_Kind,
	S_Buys, O_Prc, A_Prc, O_Qty, Deal_Qty,
	T_Date, D_Time, Order_No, O_Src, O_Lin,
	Oseq_No):
	print ('OnOrdMatF')
	def OnOrdResult(self, this, ID, result):
	print ('OnOrdResult')
	def OnRfOrdRptRF(self, this, Exchange, Omkt, Statusc, Ts_Code,
	Ts_Msg, Bhno, AcNo, Suba, Symb,
	Scnam, O_Kind, Buys, S_Buys, PriceType,
	O_Prc1, O_Prc2, O_Qty, Work_Qty, Kill_Qty,
	Deal_Qty, Order_No, O_Date, O_Time, O_Src,
	O_Lin, A_Prc, Oseq_No, Err_Code, Err_Msg,
	R_Time, D_Flag):
	print ('OnRfOrdRptRF')
	def OnRfOrdMatRF(self, this, Exchange, Omkt, Bhno, AcNo,
	Suba, Symb, Scnam, O_Kind, Buys,
	S_Buys, PriceType, O_Prc1, O_Prc2, A_Prc,
	O_Qty, Deal_Qty, O_Date, D_Time, Order_No,
	O_Src, O_Lin, Oseq_No):
	print ('OnRfOrdMatRF')
	def OnRfOrdResult(self, this, ID, result):
	print ('OnRfOrdResult')
	def OnRfReportQuery(self, this, RowCount, Results):
	print ('OnRfReportQuery')
	def OnRfDealQuery(self, this, RowCount, Results):
	print ('OnRfReportQuery')
	class YuantaOrdWapper:
	def __init__(self, handle, bot):
	self.bot = bot
	Iwindow = POINTER(IUnknown)()
	Icontrol = POINTER(IUnknown)()
	Ievent = POINTER(IUnknown)()
	res = atl.AtlAxCreateControlEx("Yuanta.YuantaOrdCtrl.64", handle, None,
	byref(Iwindow),
	byref(Icontrol),
	byref(GUID()),
	Ievent)
	self.YuantaOrd = GetBestInterface(Icontrol)
	self.YuantaOrdEvents = YuantaOrdEvents(self)
	self.YuantaOrdEventsConnect = GetEvents(self.YuantaOrd, self.YuantaOrdEvents)
	class StockBot:
	def __init__(self, botuid, account, pwd):
	self.Yuanta = YuantaOrdWapper (botuid, self)
	self.Account = account
	self.Pwd = pwd
	self.YuantaAccount = []
	self.YuantaSN = []
	def login (self):
	self.Yuanta.YuantaOrd.SetFutOrdConnection(self.Account, self.Pwd, 'api.yuantafutures.com.tw', '80')
	print ('login')
	def user_define (self):
	vars = self.YuantaAccount[0].split('-')
	Params = "Func=FA003|bhno={}|acno={}|suba=|type=1|currency=TWD".format (vars[1], vars[2])
	ret = self.Yuanta.YuantaOrd.UserDefinsFunc (Params, "FA003")
	print ("user_define {}".format (ret))
	class MyApp(wx.App):
	def MainLoop(self, run_func):
	# Create an event loop and make it active. If you are
	# only going to temporarily have a nested event loop then
	# you should get a reference to the old one and set it as
	# the active event loop when you are done with this one...
	evtloop = wx.GUIEventLoop()
	old = wx.EventLoop.GetActive()
	wx.EventLoop.SetActive(evtloop)
	# This outer loop determines when to exit the application,
	# for this example we let the main frame reset this flag
	# when it closes.
	while self.keepGoing:
	# At this point in the outer loop you could do
	# whatever you implemented your own MainLoop for. It
	# should be quick and non-blocking, otherwise your GUI
	# will freeze.
	# call_your_code_here()
	run_func ()
	while not q.empty():
	next_job = q.get()
	DoJob (Bot, next_job)
	# This inner loop will process any GUI events
	# until there are no more waiting.
	while evtloop.Pending():
	evtloop.Dispatch()
	# Send idle events to idle handlers. You may want to
	# throttle this back a bit somehow so there is not too
	# much CPU time spent in the idle handlers. For this
	# example, I'll just snooze a little...
	time.sleep(0.10)
	evtloop.ProcessIdle()
	wx.EventLoop.SetActive(old)
	def OnInit(self):
	self.keepGoing = True
	return True
	def run_job():
	while not q.empty():
	next_job = q.get()
	DoJob (Bot, next_job)
	if __name__ == "__main__":
	app=MyApp()
	frame = wx.Frame(None,wx.ID_ANY,"Hello")
	frame.Show (False)
	Bot = StockBot(frame.Handle, 'ACCOUNT', 'PWD')
	Job(Job.STOCK_LOGIN)
	app.MainLoop (run_job)

view raw gistfile1.txt hosted with ❤ by GitHub

Read More...

元大期貨 行情API with python

元大這一陣子出了 SmartAPI 後來發現需要先登入後，再以讀檔的方式進行串接
就來研究之前就有推出的元大API

下載位置：
http://easywin.yuantafutures.com.tw/api/download.html
 (行情API需要另外簽署文件才可以使用 )

下載完成之後需要先註冊
YuantaQuote_v2.1.2.3.ocx

這邊提供一個批次檔, 放在 QAPI 目錄中,

@echo off
SET Dir=%~dp0
regsvr32.exe "%Dir%YuantaQuote_v2.1.2.3.ocx"

使用系統管理員權限執行, 就可以不需要安裝在 C:\Yuanta\


程式說明：

一開始使用 comtypes, pythonnet, win32com 都無法使用
後來看到
http://algomarket.wikidot.com/comtypes#toc12
ActiveX Control Hosting (+ code example要點開)
YuantaQuote_v2.1.2.3.ocx使用需要UI的 ActiveX , 所以我使用了 wxPython 產生 Handle

 
執行需求：
行情API版本：2.1.2.3
python3 32位元
 帳號, 密碼需修改

免責聲明： 提供程式以供參考，請勿用來販賣或用在商業用途上，如果以本程式用來交易，亦不負任何責任

	# -*- coding: utf-8 -*-
	import wx, time
	import wx.lib.anchors as anchors
	from ctypes import byref, POINTER, windll
	from comtypes import IUnknown, GUID
	from comtypes.client import GetModule, GetBestInterface, GetEvents
	user32 = windll.user32
	atl = windll.atl
	import queue as queue
	q = queue.Queue ()
	class switch(object):
	def __init__(self, value):
	self.value = value
	self.fall = False
	def __iter__(self):
	"""Return the match method once, then stop"""
	yield self.match
	raise StopIteration
	def match(self, *args):
	"""Indicate whether or not to enter a case suite"""
	if self.fall or not args:
	return True
	elif self.value in args: # changed for v1.5, see below
	self.fall = True
	return True
	else:
	return False
	class Job:
	STOCK_LOGIN = 1
	STOCK_WATCH = 2
	def __init__ (self, do_type, do_value = 0):
	self.do_type = do_type
	self.do_value = do_value
	q.put (self)
	def DoJob(Bot, x):
	for case in switch(x.do_type):
	if case(Job.STOCK_LOGIN):
	Bot.login ()
	break
	if case(Job.STOCK_WATCH):
	Bot.watch (x.do_value)
	break
	class YuantaQuoteEvents(object):
	def __init__(self, parent):
	self.parent = parent
	def OnMktStatusChange (self, this, Status, Msg, ReqType):
	print ('OnMktStatusChange {},{},{}'.format (ReqType, Msg, Status))
	if Status == 2:
	Job(Job.STOCK_WATCH, ReqType)
	def OnRegError(self, this, symbol, updmode, ErrCode, ReqType):
	print ('OnRegError {},{},{},{}'.format (ReqType, ErrCode, symbol, updmode))
	def OnGetMktData(self, this, PriType, symbol, Qty, Pri, ReqType):
	print ('OnGetMktData')
	def OnGetMktQuote(self, this, symbol, DisClosure, Duration, ReqType):
	print ('OnGetMktQuote')
	def OnGetMktAll(self, this, symbol, RefPri, OpenPri, HighPri, LowPri, UpPri, DnPri, MatchTime, MatchPri, MatchQty, TolMatchQty,
	BestBuyQty, BestBuyPri, BestSellQty,BestSellPri, FDBPri, FDBQty, FDSPri, FDSQty, ReqType):
	#print ('OnGetMktAll\n')
	print ('{} {} c:{} o:{} h:{} l:{} v:{}'.format (ReqType, MatchTime, MatchPri, OpenPri, HighPri, LowPri, TolMatchQty))
	def OnGetDelayClose(self, this, symbol, DelayClose, ReqType):
	print ('OnGetDelayClose')
	def OnGetBreakResume(self, this, symbol, BreakTime, ResumeTime, ReqType):
	print ('OnGetBreakResume')
	def OnGetTradeStatus(self, this, symbol, TradeStatus, ReqType):
	print ('OnGetTradeStatus')
	def OnTickRegError(self, this, strSymbol, lMode, lErrCode, ReqType):
	print ('OnTickRegError')
	def OnGetTickData(self, this, strSymbol, strTickSn, strMatchTime, strBuyPri, strSellPri, strMatchPri, strMatchQty, strTolMatQty,
	strMatchAmt, strTolMatAmt, ReqType):
	print ('OnGetTickData')
	def OnTickRangeDataError(self, this, strSymbol, lErrCode, ReqType):
	print ('OnTickRangeDataError')
	def OnGetTickRangeData(self, this, strSymbol, strStartTime, strEndTime, strTolMatQty, strTolMatAmt, ReqType):
	print ('OnGetTickRangeData')
	def OnGetTimePack(self, this, strTradeType, strTime, ReqType):
	print ('OnGetTimePack {},{}'.format (strTradeType, strTime))
	def OnGetDelayOpen(self, this, symbol, DelayOpen, ReqType):
	print ('OnGetDelayOpen')
	def OnGetFutStatus(self, this, symbol, FunctionCode, BreakTime, StartTime, ReopenTime, ReqType):
	print ('OnGetFutStatus')
	def OnGetLimitChange(self, this, symbol, FunctionCode, StatusTime, Level, ExpandType, ReqType):
	print ('OnGetLimitChange')
	class YuantaQuoteWapper:
	def __init__(self, handle, bot):
	self.bot = bot
	Iwindow = POINTER(IUnknown)()
	Icontrol = POINTER(IUnknown)()
	Ievent = POINTER(IUnknown)()
	res = atl.AtlAxCreateControlEx("YUANTAQUOTE.YuantaQuoteCtrl.1", handle, None,
	byref(Iwindow),
	byref(Icontrol),
	byref(GUID()),
	Ievent)
	self.YuantaQuote = GetBestInterface(Icontrol)
	self.YuantaQuoteEvents = YuantaQuoteEvents(self)
	self.YuantaQuoteEventsConnect = GetEvents(self.YuantaQuote, self.YuantaQuoteEvents)
	class StockBot:
	def __init__(self, botuid, account, pwd):
	self.Yuanta = YuantaQuoteWapper (botuid, self)
	self.Account = account
	self.Pwd = pwd
	def login (self):
	#T port 80/443 , T+1 port 82/442 , reqType=1 T盤 , reqType=2 T+1盤
	self.Yuanta.YuantaQuote.SetMktLogon(self.Account, self.Pwd, '203.66.93.84', '80', 1, 0)
	self.Yuanta.YuantaQuote.SetMktLogon(self.Account, self.Pwd, '203.66.93.84', '82', 2, 1)
	print ('login')
	def watch (self, ret_type):
	ret = self.Yuanta.YuantaQuote.AddMktReg ('1101', "4", ret_type, 0)
	print ("AddMktReg {}".format (ret))
	class MyApp(wx.App):
	def MainLoop(self, run_func):
	# Create an event loop and make it active. If you are
	# only going to temporarily have a nested event loop then
	# you should get a reference to the old one and set it as
	# the active event loop when you are done with this one...
	evtloop = wx.GUIEventLoop()
	old = wx.EventLoop.GetActive()
	wx.EventLoop.SetActive(evtloop)
	# This outer loop determines when to exit the application,
	# for this example we let the main frame reset this flag
	# when it closes.
	while self.keepGoing:
	# At this point in the outer loop you could do
	# whatever you implemented your own MainLoop for. It
	# should be quick and non-blocking, otherwise your GUI
	# will freeze.
	# call_your_code_here()
	run_func ()
	while not q.empty():
	next_job = q.get()
	DoJob (Bot, next_job)
	# This inner loop will process any GUI events
	# until there are no more waiting.
	while evtloop.Pending():
	evtloop.Dispatch()
	# Send idle events to idle handlers. You may want to
	# throttle this back a bit somehow so there is not too
	# much CPU time spent in the idle handlers. For this
	# example, I'll just snooze a little...
	time.sleep(0.10)
	evtloop.ProcessIdle()
	wx.EventLoop.SetActive(old)
	def OnInit(self):
	self.keepGoing = True
	return True
	def run_job():
	while not q.empty():
	next_job = q.get()
	DoJob (Bot, next_job)
	if __name__ == "__main__":
	app=MyApp()
	frame = wx.Frame(None,wx.ID_ANY,"Hello")
	frame.Show (False)
	Bot = StockBot(frame.Handle, ACCOUNT, PWD)
	Job(Job.STOCK_LOGIN)
	app.MainLoop (run_job)

view raw gistfile1.txt hosted with ❤ by GitHub

Read More...