验证中...
AutoThs
Raw Copy
# -*- coding: UTF-8 -*-
from pywinauto import application
import time
from pywinauto import clipboard
import win32con
from collections import namedtuple
SymbolPosition = namedtuple('SymbolPosition', """
symbol total available pnl avg_cost pnl_ratio
""".split())
TradeInfo = namedtuple('TradeInfo', """
trade_time symbol bs qty price turnover order_id trade_id
""")
OrderInfo =namedtuple('OrderInfo', """
order_time symbol bs ststus order_qty order_price traded_qty order_id
""")
class AutoThs():
def __init__(self,
thspath='',
account='',
trade_pw='',
communication_pw=''):
self.path = thspath
self.account = account
self.trade_pw = trade_pw
self.communication_pw = communication_pw
self.app = None
self.trade_win = None
self.info_win = None
def attach(self):
try:
self.app = application.Application().connect(path=self.path)
except:
self._do_login()
self.trade_win = self.app.window(title_re=u'网上股票交易系统5.0')
self.info_win = self.trade_win.window(title_re='Custom2', class_name='CVirtualGridCtrl')
self.trade_win.type_keys('{F2}')
def _do_login(self):
self.app = application.Application().start(self.path)
win = self.app.window(title_re=u'用户登录', class_name='#32770')
account_edit = win[u'帐\u3000\u3000号(&A):Edit']
trade_pw_edit = win[u'交易密码(&G):Edit']
communication_pw_dit = win[u'通讯密码(&K):Edit']
account_edit.type_keys(self.account)
trade_pw_edit.type_keys(self.trade_pw)
communication_pw_dit.type_keys(self.communication_pw)
win.type_keys('{ENTER}')
time.sleep(1)
def _set_order(self, symbol='', price='', qty=0):
symbol_edit = self.trade_win[u'Edit1']
price_edit = self.trade_win[u'Edit2']
qyt_edit = self.trade_win[u'Edit3']
#symbol_edit.set_focus()
symbol_edit.set_edit_text(symbol)
price_edit.set_edit_text(price)
qyt_edit.set_edit_text(qty)
def _close_popup_window(self):
popup_hwnd = self.trade_win.popup_window()
if popup_hwnd:
popup_window = self.app.window(handle=popup_hwnd)
popup_window.set_focus()
popup_window.Button.click()
return True
return False
def _close_popup_windows(self):
while self._close_popup_window():
time.sleep(0.5)
def _refresh_window(self):
self.trade_win.type_keys('{F5}')
time.sleep(0.2)
def _select_window(self, win=''):
if win == 'W':
self.info_win.post_message(win32con.WM_KEYDOWN, ord('W'))
self.info_win.post_message(win32con.WM_KEYDOWN, ord('W'))
elif win == 'E':
self.info_win.post_message(win32con.WM_KEYDOWN, ord('E'))
self.info_win.post_message(win32con.WM_KEYDOWN, ord('E'))
elif win == 'R':
self.info_win.post_message(win32con.WM_KEYDOWN, ord('R'))
self.info_win.post_message(win32con.WM_KEYDOWN, ord('R'))
else:
assert 0
time.sleep(0.1)
def buy(self, symbol='', price='', qty=0):
assert qty > 0
self.trade_win.type_keys('{F2}')
time.sleep(0.01)
self.trade_win.type_keys('{F1}')
self._set_order(symbol, price, qty)
time.sleep(0.2)
self.trade_win[u'买入[B]'].click()
self._close_popup_windows()
def sell(self, symbol='', price='', qty=0):
assert qty > 0
self.trade_win.type_keys('{F1}')
time.sleep(0.01)
self.trade_win.type_keys('{F2}')
self._set_order(symbol, price, qty)
time.sleep(0.2)
self.trade_win[u'卖出[S]'].click()
self._close_popup_windows()
def get_available_fund(self):
fund = self.trade_win.window(best_match=u'StatusBar')
return float(fund.texts()[5])
def get_position(self):
self._select_window('W')
self.trade_win.click_input(button='left', coords=(332, 491))
self.trade_win.type_keys('^c')
data = clipboard.GetData().split('\r\n')[1:]
poss = []
for i in data:
pos_item = i.split('\t')
if int(pos_item[2]) == 0 or not pos_item[0].isdigit() or pos_item[0] == '': continue
poss.append(SymbolPosition(symbol=pos_item[0],
total=int(pos_item[2]),
available=int(pos_item[3]),
pnl=float(pos_item[5]),
avg_cost=float(pos_item[6]),
pnl_ratio=float(pos_item[7])))
return poss
def get_trade(self):
self._select_window('E')
self.trade_win.click_input(button='left', coords=(332, 491))
self.trade_win.type_keys('^c')
data = clipboard.GetData()
data = data.split('\r\n')[1:]
trades = []
for i in data:
trade_item = i.split('\t')
if trade_item[0] == '': continue
trades.append(TradeInfo(trade_time=trade_item[0],
symbol=trade_item[1],
bs= 'B' if trade_item[3] == u'买' else 'S',
qty=int(trade_item[4]),
price=float(trade_item[5]),
turnover=float(trade_item[6]),
order_id=int(trade_item[7]),
trade_id=int(trade_item[8])))
return trades
def get_alive_order(self):
self._select_window('R')
self.trade_win.click_input(button='left', coords=(332, 491))
self.trade_win.type_keys('^c')
data = clipboard.GetData()
data = data.split('\r\n')[1:]
orders = []
for i in data:
order_item = i.split('\t')
if order_item[0] == '': continue
orders.append(OrderInfo(order_time=order_item[8],
symbol=order_item[0],
bs= 'B' if order_item[7] == u'买' else 'S',
ststus=order_item[2],
order_qty=int(order_item[3]),
order_price=float(order_item[5]),
traded_qty=int(order_item[4]),
order_id=int(order_item[10])))
return orders
def cancel_order(self, order_id=0):
self.alive_orders = self.get_alive_order()
order_idx = 1
for order in self.alive_orders:
if order.order_id == order_id:
self.info_win.double_click_input(coords=(100, order_idx * 17 + 7))
self._close_popup_windows()
order_idx += 1
def cancel_all(self):
self.trade_win[u'全撤(Z /)Button'].click()
self._close_popup_windows()
if __name__ == '__main__':
aths = AutoThs(u'D:/海通融资融券5.0/xiadan.exe','xxxx', 'xxxx', 'xxxx')
aths.attach()
#aths.buy('510050', '2.90', 100)
#aths.sell('510500', '5.70', 100)
print(aths.get_available_fund())
#aths._refresh_window()
aths.cancel_order(2530)
#aths.cancel_all()

Comment list( 0 )

You need to Sign in for post a comment

Help Search