maxinxin/zhitou_trade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
zhitou_trade/autobasic/hualong_clienttrader.py

746 lines
26 KiB
Python
Raw Normal View History

first
2025-05-22 16:47:45 +08:00
# -*- coding: utf-8 -*-
"""
Created on Fri Aug 5 15:29:12 2022
@author: MINGL
"""
import pandas as pd
import pyautogui
import pywinauto
from PIL import ImageGrab
xinxin
2025-06-11 11:01:37 +08:00
from pywinauto import clipboard,keyboard
first
2025-05-22 16:47:45 +08:00
from skimage import io
import time
import io as mio
import pytesseract
import warnings
import datetime
from tools import *
warnings.filterwarnings('ignore')
mxx
2025-05-26 09:19:39 +08:00
from pathlib import Path
root_path = Path(__file__).parent.parent
path = r'C:\trade_software1\trade_software\hualong\Tc.exe'
first
2025-05-22 16:47:45 +08:00
engine = create_engine(
'mysql+pymysql://cn_ainvest_db:cn_ainvest_sd3a1@rm-2zewagytttzk6f24xno.mysql.rds.aliyuncs.com:3306/',
encoding="utf-8", echo=False)
class HLClientTrader:
"""
基于"开源证券客户端"自动交易程序
"""
def __init__(self, account_name, exe_path=path):
self.path = exe_path
self.account_name = account_name
self.log = Logger(f'{root_path}/logs',self.account_name)
# 用户券商信息
sql = f"select * from ainvest_usercount where username='{account_name}'"
df_count = download_data_from_db(sql, 'ai_strategy_update_iddb')
self.securities_name = str(df_count['securities_username'].tolist()[0])
self.securities_password = str(df_count['securities_password'].tolist()[0])
self.communication_password = str(df_count['communication_password'].tolist()[0])
def exit(self):
'''
结束进程
Returns
-------
None.
'''
self._app.kill()
def _close_prompt_windows(self):
time.sleep(0.1)
for window in self._app.windows(class_name="#32770"):
if window.window_text() != '网上股票交易系统*':
window.close()
time.sleep(0.1)
def login(self):
'''
启动软件并登录
Returns
-------
None.
'''
account_name = self.account_name
broker = '华龙证券'
#软件启动判断，若有进行中进程，先关闭再重启
try:
self._app = pywinauto.Application().connect(
path=self.path, timeout=1
)
self.exit()
self.log.warn('There are running programs, and there are processes in retreat')
except Exception:
pass
# 启动软件进程，输入账号密码
self.log.info('Start login')
start_time = time.time()
reset_proxy_to_default() # 初始化proxy信息
# 重置mac
set_mac = SetMac()
set_mac.run()
mxx
2025-05-26 09:19:39 +08:00
if not connect_wifi('ZTE_A5DA76', '1234567890'): # 'Redmi K40','123456789'
first
2025-05-22 16:47:45 +08:00
self.log.error("无法建立网络连接，请检查配置")
else:
self.log.info("网络已就绪")
get_ip_times = 0
writer = ExcelDataWriter() # 初始化ip_records 表格
used_ip = writer.get_unavailable_ips(broker, account_name)
while get_ip_times < 3:
try:
item = get_ip_data() # 获取动态IP
if item['ip'] in used_ip:
get_ip_times += 1
else:
self.log.info(f'代理IP已经获取{item}')
break
except Exception as e:
self.log.error(f'获取IP失败，请检查网络连接{e}')
time.sleep(5)
# raise ConnectionError(f"获取IP失败，请检查网络连接。错误详情: {e}")
# 设置全局代理
proxy_ip = item['ip']
proxy_port = item['port']
set_global_proxy(proxy_ip, proxy_port, enable=True)
# exit_ip = get_proxy_ip(proxy_ip,proxy_port)
time_with_change_proxy = time.time() - start_time
self.log.info('全局代理设置成功')
insert_data = {'as_of_date': datetime.datetime.fromtimestamp(start_time).strftime('%Y-%m-%d %H:%M:%S.%f'),
'broker': broker,
'account': account_name,
'proxy_ip': proxy_ip,
'proxy_port': proxy_port,
'ip_location': item['prov'] + ':' + item['city'],
# 'exit_ip': exit_ip,
'ip_cross_check_result': '一致',
'ip_switch_time': time_with_change_proxy,
# 'ip_expire_time':item['expire']
'ip_survival_time': '3min'
}
# 写入数据表
writer = ExcelDataWriter() # 初始化ip_records 表格
writer.write_data(insert_data)
#启动软件进程，输入账号密码
self.log.info('Start login')
self._app = pywinauto.Application().start(self.path)
time.sleep(1)
while True:
try:
self._app.window(title_re='华龙证券网上交易.*').wait("ready")
break
except RuntimeError:
pass
time.sleep(1)
self._app.window(title_re='华龙证券网上交易.*').window(control_id=0x3E9, class_name='Edit',found_index = 0).click()
pyautogui.typewrite(self.securities_name)
time.sleep(0.1)
self._app.window(title_re='华龙证券网上交易.*').window(control_id=0x450, class_name='AfxWnd42',found_index = 0).click()
time.sleep(0.5)
pyautogui.typewrite(self.securities_password)
time.sleep(0.5)
#切换为验证码
self._app.window(title_re='华龙证券网上交易.*').window(control_id=0x457, class_name='ComboBox',
found_index=0).click()
pyautogui.click(885, 577)
# time.sleep(0.5)
# try:
while True:
id_code = self.verify_code()
print('#############',id_code)
time.sleep(0.2)
self._app.window(title_re='华龙证券网上交易.*').window(control_id=0x458, class_name='Edit').type_keys('{BACKSPACE}' * 6)
time.sleep(0.2)
self._app.window(title_re='华龙证券网上交易.*').window(control_id=0x458, class_name='Edit').type_keys(id_code)
time.sleep(0.2)
# pywinauto.keyboard.send_keys("{ENTER}")
self._app.window(title_re='华龙证券网上交易.*').window(control_id=0x1, class_name='Button').click()
time.sleep(0.5)
try:
self._app.window(title_re='华龙证券网上交易.*').set_focus()
result_idcode = self.result_text()
time.sleep(1)
print('(((((((((',result_idcode)
self.log.softInfo(result_idcode)
if '验证码' in result_idcode:
# print('111')
self.log.warn('Verification code error')
self._app.window(title_re='华龙证券网上交易.*').window(control_id=0x458, class_name='Edit').click()
time.sleep(0.5)
else:
break
except Exception:
break
time.sleep(10)
# 关闭弹窗
# self._close_prompt_windows()
tryrun2(self._app.window(title_re='通达信网上交易.*').wait("ready"))
#设置主窗口及菜单窗口
self.main_wnd = self._app.window(title_re='通达信网上交易*')
self.left_wnd = self.main_wnd.window(class_name='SysTreeView32', control_id=0xE900)
xinxin
2025-06-11 11:01:37 +08:00
first
2025-05-22 16:47:45 +08:00
def _turn_off_advertising(self):
'''
关闭弹窗
Returns
-------
None.
'''
try:
self._app.top_window().set_focus()
time.sleep(0.5)
pywinauto.keyboard.send_keys("{ESC}")
pywinauto.keyboard.send_keys("{ENTER}")
pywinauto.keyboard.send_keys("{ESC}")
except Exception as e:
pass
def result_text(self):
time.sleep(0.5)
time.sleep(0.2)
ImageGrab.grab(bbox=(750, 600, 1010, 630)).save(f'{root_path}/imgs/temp123.png')
im = io.imread(f'{root_path}/imgs/temp123.png')
# print(')))))',im)
id_code = pytesseract.image_to_string(im, lang='chi_sim',config='--psm 7').strip()
id_code = id_code.replace(' ', '')
return id_code
def verify_code(self):
'''
获取登录验证码
Returns
-------
id_code : TYPE
DESCRIPTION.
'''
time.sleep(0.5)
time.sleep(0.2)
ImageGrab.grab(bbox=(1080, 510, 1170, 600)).save(f'{root_path}/imgs/temp1231.png')
im = io.imread(f'{root_path}/imgs/temp1231.png')
id_code = pytesseract.image_to_string(im, lang='eng',
config='--psm 6 --oem 3 -c tessedit_char_whitelist=0123456789').strip()
id_code = id_code.replace(' ', '')
return id_code
def buy(self, stock_no, price, num):
'''
买入股票
Parameters
----------
stock_no : TYPE
DESCRIPTION.
price : TYPE
DESCRIPTION.
num : TYPE
DESCRIPTION.
Returns
-------
None.
'''
self.left_wnd.get_item([0]).click_input()
self.__trade(stock_no, price, num)
def sell(self, stock_no, price, num):
'''
卖出股票
Parameters
----------
stock_no : TYPE
DESCRIPTION.
price : TYPE
DESCRIPTION.
num : TYPE
DESCRIPTION.
Returns
-------
None.
'''
# print(self.left_wnd.select())
self.left_wnd.get_item([1]).click_input()
self.__trade(stock_no, price, num)
def etf_buy(self,ticker,num):
self.left_wnd.select(['场内基金','基金认购'])
# 设置股票代码
self.main_wnd.window(control_id=0x408, class_name="Edit").click()
self.main_wnd.window(control_id=0x408, class_name="Edit").set_text(str(ticker)[2:])
time.sleep(1)
# 设置金额
self.main_wnd.window(control_id=0x40A, class_name="Edit").set_text(str(num))
time.sleep(1)
# 点击确认
self.main_wnd.window(control_id=0x3EE, class_name="Button").click()
time.sleep(1)
# 获取弹窗返回的信息
# self._app.top_window().set_focus()
one_text = self._app.window(control_id=0x0, class_name='#32770',found_index = 0).window(control_id=0x3EC, class_name='Static',found_index = 0).window_text()
# print(one_text)
# 摘取合同信息
self.log.info(one_text.replace('/n', ' '))
time.sleep(0.2)
#确认申购委托
# self._app.window(control_id=0x0, class_name="#32770", found_index=0).window(control_id=0x2,class_name='Button').click()
def get_cancel_entrust(self, ticker):
'''
按股票撤单
Parameters
----------
ticker : TYPE
DESCRIPTION.
Returns
-------
None.
'''
self.left_wnd.select(['撤单[F3]'])
time.sleep(1)
# 删除框内ticker
self.main_wnd.window(control_id=0xD14, class_name='Edit').click()
for i in range(7):
pywinauto.keyboard.send_keys("{DELETE}")
# 输入ticker
self.main_wnd.window(control_id=0xD14, class_name='Edit').type_keys(ticker)
time.sleep(0.2)
# 选中ticker
pywinauto.keyboard.send_keys("{ENTER}")
time.sleep(0.1)
# 点击撤单
self.main_wnd.window(control_id=0x44B, class_name='Button').click()
time.sleep(0.1)
# 确认撤单
pywinauto.keyboard.send_keys("{ENTER}")
time.sleep(0.2)
#获取弹窗信息
result = self._app.top_window().window(control_id=0x3EC, class_name='Static').window_text()
self.log.info(result)
pywinauto.keyboard.send_keys("{ENTER}")
time.sleep(0.2)
pywinauto.keyboard.send_keys("{ESC}")
def cancel_entrust(self):
'''
全撤
Returns
-------
None.
'''
# # 进入撤单页面
self.left_wnd.select(['撤单[F3]'])
time.sleep(1)
# 一键全撤
self.main_wnd.window(control_id=0x7531, class_name='Button').click()
self._app.top_window().set_focus()
self.main_wnd.window(control_id=0x44B, class_name='Button').click()
time.sleep(1)
pywinauto.keyboard.send_keys("{ENTER}")
time.sleep(0.5)
pywinauto.keyboard.send_keys("{ENTER}")
def get_unsettled_trades(self):
'''
获取未成交单
Returns
-------
one_df : TYPE
DESCRIPTION.
'''
self.left_wnd.select(['撤单[F3]'])
self._copy_data()
one_df =self._data_to_df()
if len(one_df) == 0:
pass
else:
one_df['证券代码'] = one_df['证券代码'].apply(lambda x: tranTicker(x.strip('=').strip('"')))
return one_df
def get_balance(self):
'''
获取资金情况
Returns
-------
record : TYPE
DESCRIPTION.
'''
self.left_wnd.get_item([5, 0]).click_input()
# self.left_wnd.select(['查询','资金股票']).click_input()
time.sleep(1)
# self.main_wnd.window(control_id=0xC8, class_name="AfxWnd42s", found_index=0).click()
self._app.top_window().set_focus()
time.sleep(1)
self._output_data()
date = str(datetime.date.today()).replace('-', '')
balance = pd.read_csv(fr'C:\Users\EDY\Documents\{date} 资金股份查询.xls', encoding='gbk', sep='\t', nrows=2,
error_bad_lines=False)
result = balance[['="参考市值"', '="可用"', '="资产"']]
balance['Position_amount'] = float(result['="参考市值"'].values[0].replace('"', '').replace('=', ''))
balance['Cash'] = float(result['="可用"'].values[0].replace('"', '').replace('=', ''))
balance['Total_account'] = float(result['="资产"'].values[0].replace('"', '').replace('=', ''))
df_every_money = balance[['Position_amount', 'Cash', 'Total_account']]
df_every_money['Account_Name'] = self.account_name
df_every_money['As_Of_Date'] = datetime.date.today()
print(df_every_money)
return df_every_money
def _output_data(self):
self._app.top_window().set_focus()
time.sleep(1)
self.main_wnd.window(control_id=0x0, class_name='#32770',found_index = 0).window(control_id=0x47F, class_name='Button').click()
print("5")
self._app.top_window().set_focus()
# while True:
# try:
# self._app.top_window().window(title='确认另存为').wait("ready")
# break
# except RuntimeError:
# pass
self._app.top_window().window(control_id=0xE9, class_name='Button',found_index = 0).click()
self._app.top_window().window(control_id=0x1, class_name='Button', found_index=0).click()
# self._app.top_window().window(control_id=0x2, class_name='Button', found_index=0).click()
time.sleep(2)
# self._app.top_window().set_focus()
# pywinauto.keyboard.send_keys("{Y}")
# time.sleep(1)
# pywinauto.keyboard.send_keys("%{F4}")
# print("6")
def get_position(self):
'''
获取持仓情况
Returns
-------
df_position : TYPE
DESCRIPTION.
'''
# print(self.left_wnd)
self.left_wnd.get_item([5,0]).click_input()
# self.left_wnd.select(['查询','资金股票']).click_input()
time.sleep(1)
# self.main_wnd.window(control_id=0xC8, class_name="AfxWnd42s", found_index=0).click()
self._app.top_window().set_focus()
time.sleep(1)
self._output_data()
date = str(datetime.date.today()).replace('-','')
position = pd.read_csv(fr'C:\Users\EDY\Documents\{date} 资金股份查询.xls', encoding='gbk', sep='\t',
skiprows=3, error_bad_lines=False).applymap(
lambda x: x.replace('"', '').replace('=', '') if isinstance(x, str) else x)
position.columns = [i.replace('"', '').replace('=', '') for i in list(position.columns)]
position.rename(columns={'证券代码': "Ticker",
'证券名称': "Ticker_name",
"库存数量": "Number_transactions",
"可卖数量": "cash",
"成本价": "cost_price",
"当前价": "market_price",
"最新市值": "market_value",
"浮动盈亏": "profit_and_loss",
"盈亏比例(%)": "profit_loss_ratio",
}, inplace=True)
position['Ticker'] = position['Ticker'].apply(lambda x: tranTicker(x))
position['Account_Name'] = self.account_name
position['As_Of_Date'] = datetime.date.today()
position = position[
['As_Of_Date', 'Account_Name', 'Ticker', 'Ticker_name', 'Number_transactions', 'cash', 'cost_price',
'market_price', 'market_value', 'profit_and_loss', 'profit_loss_ratio']]
# print(position)
return position
def get_order(self):
'''
获取委托情况
Returns
-------
TYPE
DESCRIPTION.
'''
self.left_wnd.select(['查询[F4]','当日委托'])
self._copy_data()
return self._data_to_df()
def get_today_trades(self):
'''
获取当日成交
Returns
-------
TYPE
DESCRIPTION.
'''
self.left_wnd.get_item([5, 2]).click_input()
# self.left_wnd.select(['查询','资金股票']).click_input()
time.sleep(1)
# self.main_wnd.window(control_id=0xC8, class_name="AfxWnd42s", found_index=0).click()
self._app.top_window().set_focus()
time.sleep(1)
self._output_data()
date = str(datetime.date.today()).replace('-','')
trades = pd.read_csv(fr'C:\Users\EDY\Documents\{date} 当日成交查询.xls', encoding='gbk', sep='\t',
error_bad_lines=False).applymap(
lambda x: x.replace('"', '').replace('=', '') if isinstance(x, str) else x)
trades.columns = [i.replace('"', '').replace('=', '') for i in list(trades.columns)]
trades = trades[['成交时间', '证券代码', '证券名称', '买卖标志', '成交数量', '成交价格']]
trades['Account_Name'] = self.account_name
trades['As_Of_Date'] = datetime.date.today()
trades.rename(columns={'成交时间': 'Trade_time', '证券代码': 'Ticker', '证券名称': 'Ticker_name',
'买卖标志': 'Operate',
'成交数量': 'Number_transactions',
'成交价格': 'Average_price'}, inplace=True)
trades['Operate'] = trades['Operate'].apply(lambda x: '卖' if '证券卖出' in x else '买')
trades['Ticker'] = trades['Ticker'].apply(lambda x: tranTicker(str(x).replace('="', '').replace('"', '')))
return trades
def __trade(self, stock_no, price, amount):
'''
交易输入函数
Parameters
----------
stock_no : TYPE
DESCRIPTION.
price : TYPE
DESCRIPTION.
amount : TYPE
DESCRIPTION.
Returns
-------
None.
'''
time.sleep(0.2)
# 设置股票代码
self.main_wnd.window(control_id=0x2EE5, class_name="AfxWnd42").click()
self.main_wnd.window(control_id=0x2EE5, class_name="AfxWnd42").type_keys(str(stock_no))
time.sleep(1)
# 设置价格
self.main_wnd.window(control_id=0x2EE6, class_name="Edit").set_text(str(price))
time.sleep(1)
# 设置股数目
self.main_wnd.window(control_id=0x2EE7, class_name="Edit").set_text(str(amount))
time.sleep(1)
# 点击卖出or买入
self.main_wnd.window(control_id=0x7DA, class_name="Button").click()
time.sleep(1)
#获取弹窗返回的信息
# self._app.top_window().set_focus()
one_text = self._app.window(control_id=0x0, class_name='#32770',found_index = 0).window_text()
if '委托价格的小数部分应该为2位' in one_text:
# 点击“否”
self.log.warn(one_text.replace('/n',' '))
self._app.top_window().window(control_id=0x7, class_name='Button').click()
# 保留两位小数截断
price = str(price)[:-1]
# 重新设置价格
self.main_wnd.window(control_id=0x409, class_name="Edit").set_text(str(price))
time.sleep(0.5)
# 点击卖出or买入
self.main_wnd.window(control_id=0x3EE, class_name="Button").click()
time.sleep(1)
self._app.top_window().set_focus()
one_text = self._app.top_window().window(control_id=0x410, class_name='Static').window_text()
else:
# 摘取合同信息
self.log.info(one_text.replace('/n',' '))
time.sleep(0.2)
# 确认买入
self._app.window(control_id=0x0, class_name='#32770', found_index=0).window(control_id=0x1B67, class_name='Button').click()
# pywinauto.keyboard.send_keys("{ENTER}")
time.sleep(0.5)
# # 摘取合同信息2
# one_text2 = self._app['提示'].window(control_id=0x0, class_name='#32770').window_text()
# print(one_text2)
# self.log.info(one_text2.replace('/n', ' '))
# ImageGrab.grab(bbox=(859, 800, 1307, 836)).save(f'{root_path}/imgs/temp1231.png')
#
# image_path = f'{root_path}/imgs/temp1231.png'
# text = ocr_image(image_path)
# self.log.info(text.replace('/n', ' '))
# 确认合同
pywinauto.keyboard.send_keys("{ENTER}")
time.sleep(0.5)
def _copy_data(self):
'''
拷贝数据函数
Returns
-------
None.
'''
while True:
keyboard.send_keys('^C')
# 获取验证码
id_code = self.verify_code(0x965)
time.sleep(0.2)
self._app.top_window().Edit3.type_keys('{BACKSPACE}' * 6)
time.sleep(0.2)
self._app.top_window().Edit3.type_keys(id_code)
time.sleep(0.2)
pywinauto.keyboard.send_keys("{ENTER}")
time.sleep(0.5)
try:
self._app.top_window().set_focus()
result_idcode = self._app.top_window().window(control_id=0x966, class_name='Static').window_text()
self.log.softInfo(result_idcode)
if '验证码错误' in result_idcode:
self.log.warn('Verification code error')
self._app.top_window().window(control_id=0x2, class_name="Button").click()
time.sleep(0.5)
else:
break
except Exception:
break
def _data_to_df(self):
'''
剪切板转dataframe
Returns
-------
TYPE
DESCRIPTION.
'''
data_table = clipboard.GetData()
df_table = pd.read_csv(mio.StringIO(data_table), delimiter='\t', na_filter=False)
table_dict = df_table.to_dict('records')
if len(table_dict) == 0:
return pd.DataFrame()
else:
data = []
for i in table_dict:
data.append(pd.DataFrame.from_dict(i, orient='index').T)
return pd.concat(data, ignore_index=True)
def subscribe_stock(self):
# self.left_wnd.select(['新股/债申购','沪深新股/债申购', '新股一键申购']).click()
#新股
self.left_wnd.get_item([6,0,1]).click_input()
time.sleep(0.5)
self.main_wnd.window(class_name='#32770', control_id=0x0,found_index = 1).window(control_id=0x3E5,
class_name='Button').click()
time.sleep(0.5)
self.main_wnd.window(class_name='#32770', control_id=0x0, found_index=1).window(control_id=0x48C,
class_name='Button').click()
time.sleep(1)
self._app.window(control_id=0x0, class_name='#32770').window(control_id=0x1B67,class_name='Button').click()
time.sleep(1)
self._app.window(control_id=0x0, class_name='#32770').window(control_id=0x1B67, class_name='Button').click()
time.sleep(0.5)
# one_text = self._app.window(control_id=0x0, class_name='#32770').child_window(class_name = 'Static',found_index = 1).window_text()
# print(one_text)
time.sleep(0.5)
self._app.window(control_id=0x0, class_name='#32770').window(control_id=0x1B67, class_name='Button').click()
#新债
self.left_wnd.get_item([6, 0, 3]).click_input()
time.sleep(0.5)
self.main_wnd.window(class_name='#32770', control_id=0x0, found_index=1).window(control_id=0x3E5,
class_name='Button').click()
time.sleep(0.5)
self.main_wnd.window(class_name='#32770', control_id=0x0, found_index=1).window(control_id=0x48C,
class_name='Button').click()
time.sleep(0.5)
self._app.window(control_id=0x0, class_name='#32770').window(control_id=0x1B67, class_name='Button').click()
time.sleep(1)
self._app.window(control_id=0x0, class_name='#32770').window(control_id=0x1B67, class_name='Button').click()
time.sleep(0.5)
self._app.window(control_id=0x0, class_name='#32770').window(control_id=0x1B67, class_name='Button').click()
if __name__ == '__main__':
account_name = str('13593501968')
a = HLClientTrader(account_name)
a.login()
# a.subscribe_stock()
# a.exit()
# 买入
# a.buy('600000', '7','100')
# a.sell('600000', '7.29', '900')
# a.sell('600000', '7.29','900')
# a.sell('600000', '7.29','900')
# a.sell('600000', '7.29','900')
# a.sell('600000', '1.29','900')
# 卖出
# a.sell('600835', '12.45','900')
# a.etf_buy('600000','0')
# 获取当日实际持仓
# print(a.get_position())
# 获取资金情况
# print(a.get_balance())
# 获取当日成交
# print(a.get_today_trades())
Reference in New Issue Copy Permalink