元大期貨 交易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