ring520/dx_sl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

提交新建内容

Browse Source
This commit is contained in:
lintaogood
2025-09-01 12:08:41 +08:00
commit 198ec2dc8f
14 changed files with 1911 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

290
order_executor.py Normal file
View File

@@ -0,0 +1,290 @@
import subprocess
import time
import pygetwindow as gw
import win32gui
import win32con
import cv2
import os
import pyautogui
# from logger_utils import log_info, log_warning, log_error, log_trigger
from logger_utils_new import log_info, log_warning, log_error, log_trigger, log_debug, setup_logger, LOG_STYLES
# 移除 pyautogui 导入
class OrderExecutor:
def __init__(self, target_window_titles=None):
if target_window_titles is None:
# 默认支持的行情软件列表
self.target_window_titles = ["东方财富终端", "通达信金融终端", "同花顺"]
else:
self.target_window_titles = target_window_titles
self.current_software = None
self.order_window_title = 'FlashTradeDlgDlg'
self.order_count_window_title = '提示'
self.ahk_script_path = "D:/gp_data/order_executor.ahk"
def detect_trading_software(self):
"""检测当前运行的行情软件"""
for title in self.target_window_titles:
window_list = gw.getWindowsWithTitle(title)
if window_list:
self.current_software = title
log_info(f"检测到行情软件: {title}")
return title
log_warning("未检测到支持的行情软件")
return None
def run_ahk_script(self, command):
"""执行AHK脚本命令"""
try:
# 尝试查找AHK安装路径
ahk_path = None
# 常见安装路径
possible_paths = [
"C:\\Program Files\\AutoHotkey\\v2\\AutoHotkey.exe", # v2版本
"C:\\Program Files\\AutoHotkey\\AutoHotkey.exe",
"AutoHotkey.exe" # 如果在PATH环境变量中
]
for path in possible_paths:
if os.path.exists(path):
ahk_path = path
break
if not ahk_path:
log_error("未找到AutoHotkey安装路径")
return False
subprocess.Popen(
[ahk_path, '/restart', self.ahk_script_path, command],
creationflags=subprocess.CREATE_NO_WINDOW
)
return True
except Exception as e:
log_error(f"执行AHK脚本失败: {e}")
return False
def click_confirm_button(self, window_title, confirm_ratio=(0.8, 0.8), delay=0.3):
"""
修改为使用AHK点击确认按钮
"""
try:
time.sleep(delay)
window_list = gw.getWindowsWithTitle(window_title)
if not window_list:
log_warning(f"未找到行情软件 '{window_title}' 的窗口")
return False
window = window_list[0]
left, top, width, height = window.left, window.top, window.width, window.height
x = left + int(width * confirm_ratio[0])
y = top + int(height * confirm_ratio[1])
# 使用AHK执行点击
command = f"click {x} {y}"
self.run_ahk_script(command)
log_trigger(f"AHK点击 '{window_title}' ，位置: ({x}, {y})")
return True
except Exception as e:
log_error(f"点击确认按钮失败: {e}")
return False
def click_button_by_image(self, image_path, window_title=None,
timeout=10, retry_interval=0.5,
confidence=0.8, move_duration=0.2):
"""
修改为使用AHK进行图像识别点击
"""
log_info(f"开始AHK图像识别: {image_path}")
# 构建AHK命令
command = f"image_click {image_path}"
if window_title:
command += f" {window_title}"
command += f" {confidence}"
# 执行AHK命令
success = self.run_ahk_script(command)
if success:
log_trigger(f"AHK成功识别并点击图像 '{image_path}'")
return True, (0, 0) # AHK不返回坐标，用(0,0)占位
else:
log_warning(f"AHK未能识别图像 '{image_path}'")
return False, None
def place_order_tongdaxin(self, pure_code, auto_push):
"""通达信下单方法"""
try:
# 获取所有匹配标题的窗口
window_list = gw.getWindowsWithTitle(self.current_software)
if not window_list:
log_warning(f"未找到标题为 '{self.current_software}' 的窗口")
return False
# 筛选逻辑：优先选择可见且活动的窗口
target_window = None
def is_window_visible(hwnd):
return win32gui.IsWindowVisible(hwnd) and win32gui.IsWindowEnabled(hwnd)
for window in window_list:
hwnd = window._hWnd
if is_window_visible(hwnd) and window.isActive:
target_window = window
break
# 如果没有活动窗口，选择第一个可见窗口
if not target_window:
for window in window_list:
hwnd = window._hWnd
if is_window_visible(hwnd):
target_window = window
break
# 如果都没有，选择第一个匹配的窗口
if not target_window:
target_window = window_list[0]
# 使用筛选后的目标窗口进行操作
hwnd = target_window._hWnd
log_info(f"找到窗口句柄: {hwnd}")
target_window.restore()
target_window.maximize()
target_window.activate()
time.sleep(0.2)
# 通达信操作方式：先按F1进入买入界面
self.run_ahk_script("press f1")
time.sleep(0.1)
# 输入代码
self.run_ahk_script(f"type {pure_code}")
self.run_ahk_script("press enter")
time.sleep(0.1)
# 判断是否自动下单
if auto_push == 1:
# 在通达信中直接按F2进行全仓买入
self.run_ahk_script("press f2")
time.sleep(0.1)
self.run_ahk_script("press enter")
time.sleep(0.5)
# 判断是否弹出仓位的框
if self.is_window_exists(self.order_count_window_title, 0.5):
log_error(f"剩余金额不满足购买{pure_code}最低需求。")
except Exception as e:
log_error(f"通达信下单失败: {str(e)}")
return False
def place_order_eastmoney(self, pure_code, auto_push):
"""东方财富下单方法"""
try:
# 获取所有匹配标题的窗口
window_list = gw.getWindowsWithTitle(self.current_software)
if not window_list:
log_warning(f"未找到标题为 '{self.current_software}' 的窗口")
return False
# 筛选逻辑：优先选择可见且活动的窗口
target_window = None
def is_window_visible(hwnd):
return win32gui.IsWindowVisible(hwnd) and win32gui.IsWindowEnabled(hwnd)
for window in window_list:
hwnd = window._hWnd
if is_window_visible(hwnd) and window.isActive:
target_window = window
break
# 如果没有活动窗口，选择第一个可见窗口
if not target_window:
for window in window_list:
hwnd = window._hWnd
if is_window_visible(hwnd):
target_window = window
break
# 如果都没有，选择第一个匹配的窗口
if not target_window:
target_window = window_list[0]
# 使用筛选后的目标窗口进行操作
hwnd = target_window._hWnd
log_info(f"找到窗口句柄: {hwnd}")
target_window.restore()
target_window.maximize()
target_window.activate()
time.sleep(0.2)
# 使用AHK点击中心位置
self.click_confirm_button(self.current_software, (0.5, 0.5), 0.1)
# 使用AHK输入代码
self.run_ahk_script(f"type {pure_code}")
self.run_ahk_script("press enter")
time.sleep(0.1)
# 判断是否自动下单
if auto_push == 1:
self.run_ahk_script("type 21")
self.run_ahk_script("press enter")
time.sleep(0.1)
# 点击全仓按钮
success, pos = self.click_button_by_image(
image_path="../images/full_position.png",
window_title=self.order_window_title,
timeout=10,
retry_interval=0.3,
confidence=0.9
)
if success:
log_info("已点击全仓按钮")
# 判断是否弹出仓位的框
if self.is_window_exists(self.order_count_window_title, 0.5):
log_error(f"剩余金额不满足购买{pure_code}最低需求。")
else:
log_warning("未找到全仓按钮图像")
except Exception as e:
log_error(f"东方财富下单失败: {str(e)}")
return False
def place_order(self, pure_code, auto_push):
"""
根据检测到的行情软件选择相应的下单方法
"""
# 检测当前运行的行情软件
if not self.current_software:
self.detect_trading_software()
if not self.current_software:
log_error("未找到支持的行情软件窗口")
return False
# 根据不同的行情软件调用不同的下单方法
if "通达信" in self.current_software:
return self.place_order_tongdaxin(pure_code, auto_push)
elif "东方财富" in self.current_software:
return self.place_order_eastmoney(pure_code, auto_push)
else:
# 默认使用东方财富的方法
log_warning(f"未针对 {self.current_software} 实现行情软件的特殊操作，使用默认方法")
return self.place_order_eastmoney(pure_code, auto_push)
def is_window_exists(self, window_title, timeout=1.0):
"""
检查窗口是否存在
"""
start_time = time.time()
while time.time() - start_time < timeout:
window_list = gw.getWindowsWithTitle(window_title)
if window_list:
return True
time.sleep(0.1)
return False