直接伪造数据包不成功,只能用pywinauto实现交易接口了,用easyocr识别登陆验证码,话说这个模块的识别率不是很高,而且启动还廷耗时,识别模型还要自己下载(我也是好不容易才下载下来的,有需要的同学可以留邮箱),运行中偶尔还会有unknown c++的错误出现,个人感觉不是很稳定,但我加了一个if..else和try..except来处理不稳定因素,总体运行还算OK,用PIL.ImageGrab.grab来截取验证码图片,其实pywinauto也有截取图片的函数。我用的是平安证券,证券软件文件夹里面有xiadan.exe这个模块,我就是操作这个模块来实现自动交易的。其他一些小细节我在代码旁边都有注释,虽然基本功能实现了,但真正上线的时候还是要有很多改进的,现在跟大家分享,希望有错误的地方能够指出,共同学习。下一步就专心于量化策略的开发了。
import easyocr
from PIL.ImageGrab import grab
from pywinauto.application import Application
import time
#交易软件可以电脑和手机同时登陆
#print(easyocr.model_storage_directory)
#模型保存路径Windows:C:\Users\<用户名>\.EasyOCR\model
#Linux:~/ .EasyOCR / model
class exchange():
def __init__(self):
self.reader = easyocr.Reader(["ch_sim","en"],gpu=False)
self.app = Application(backend="win32").start(r"C:\xxx\xiadan.exe").Dailog
self.jpgpath=r"C:\Users\xxx\1.jpg"
self.numindex=(xx,xx,xx,xx) #验证码的位置,可以用微信截屏来定位。
def grabnum(self):
img=grab(bbox=(self.numindex))
img.save(self.jpgpath)
def loadfirst(self):
self.grabnum()
while True: # 这个识别软件老是出错,所以只能这样了。
try:
valinum = self.reader.readtext(self.jpgpath)
valicode = int(valinum[0][1].replace(" ", "")) # 有时候识别的字符间有空格
except Exception as e:
print(e)
# app.Button10.click() #重新刷新验证码,其实这里可以不用刷新
continue
else:
break
self.app.Edit3.type_keys(valicode)
self.app.Botton.click()
def loadlast(self):
loadflag = 1
self.app.Edit2.type_keys("your_password")
while loadflag:
self.loadfirst()
time.sleep(1)
if self.app[u"验证码错误,请重新输入!"].is_visible():
self.app[u"确定"].click() # 这里会自动刷新验证码
else:
loadflag = 0
# app["Button10"].get_properties()
def load(self):
try:
self.loadlast()
except Exception as e:
pass
time.sleep(1)
# app2=Application(backend="win32").connect(process=7xx4) #通过进程连接不靠谱,重启进程就变了
# app2.top_window().print_control_identifiers()
def link_new_window(self):
app2 = Application(backend="win32").connect(title="网上股票交易系统5.0", visible_only=True) # 其实这里可以用uia的
try:
app2[u"新股申购Dialog"].is_visible()
except Exception as e:
pass
else:
app2[u"新股申购Dialog"].close()
return app2
def mybuy(self,bcode,bprice,bnum):
app2=self.link_new_window()
time.sleep(0.1)
app2.top_window().type_keys("{F1}")
time.sleep(0.1)
app2.top_window().Edit0.type_keys(bcode)
app2.top_window().Edit2.type_keys(bprice)
app2.top_window().Edit3.type_keys(bnum)
app2.top_window().set_focus() #要加这个才得
app2.top_window().type_keys('^~')
time.sleep(0.5) #这里要暂停一下
app2.top_window().Button2.click() #是的话就是Button1
def mysell(self,scode,sprice,snum):
app2 = self.link_new_window()
time.sleep(0.1)
app2.top_window().type_keys("{F2}")
time.sleep(0.1)
app2.top_window().Edit0.type_keys(scode)
app2.top_window().Edit2.type_keys(sprice)
app2.top_window().Edit3.type_keys(snum)
app2.top_window().set_focus() # 要加这个才得
app2.top_window().type_keys('^~')
time.sleep(0.5) # 这里要暂停一下
app2.top_window().Button2.click() # 是的话就是Button1,测试环境,所以点击Button1(否)
def cancel_order(self,how):
app2 = self.link_new_window()
time.sleep(0.1)
app2.top_window().type_keys("{F3}")
time.sleep(0.1)
if how=="all":
app2.top_window()[u"全撤(Z /)"].click() # 全撤
elif how=="cbuy":
app2.top_window()[u"撤买(X)"].click() # 撤买
else:
app2.top_window()[u"撤卖(C)"].click() # 撤卖
time.sleep(0.5) # 这里要暂停一下
try:
app2.top_window()[u"无可撤委托"].is_visible()
except Exception as e:
app2.top_window().Button2.click() # 是的话就是Button1
else:
app2.top_window().Button.click()
def myquery(self):
app2 = self.link_new_window()
time.sleep(0.1)
app2.top_window().type_keys("{F4}")
time.sleep(0.1)
total_assets=app2.top_window().Static12.get_properties()['texts']
useful_money=app2.top_window().Static6.get_properties()['texts']
return total_assets[0],useful_money[0]
if __name__=="__main__":
my=exchange()
my.load()
my.mybuy("0000","12","30")
my.mysell("0000","12","30")
my.cancel_order("all")
to,us=my.myquery()
print(to,us)