自动化盯盘神器:Python + ADB + OpenCV 实现股票信号识别与钉钉通知
需求:由于没时间盯盘,想写一个脚本能隔一段时间监控是否出了gs然后钉钉通知消息要实现这个功能,我们需要编写一个Python脚本,结合使用来控制手机、来进行图像识别,以及库来发送钉钉消息。这个任务比较复杂,需要一些前置准备工作。我会一步步详细地指导你完成。
·
需求:由于没时间盯盘,想写一个脚本能隔一段时间监控是否出了gs
然后钉钉通知消息
要实现这个功能,我们需要编写一个Python脚本,结合使用ADB (Android Debug Bridge) 来控制手机、OpenCV 来进行图像识别,以及requests库来发送钉钉消息。
这个任务比较复杂,需要一些前置准备工作。我会一步步详细地指导你完成。
核心思路
- 连接手机: 使用ADB工具通过USB连接并控制你的Android手机。
- 自动化操作:
- 每隔5分钟,脚本会自动执行。
- 使用ADB模拟点击,刷新APP页面以获取最新数据。
- 使用ADB截取手机当前屏幕。
- 图像识别:
- 使用OpenCV分析截图。
- 首先,定位到K线图上最后一个柱子的大致区域 (Region of Interest - ROI)。
- 然后,在这个区域内寻找是否存在 ‘S’ 或 ‘G’ 的图像特征。
- 发送通知:
- 如果识别到了 ‘S’ 或 ‘G’ 信号,就调用钉钉机器人的Webhook API,发送一条预设好的消息。
步骤一:环境和工具准备
1. 在电脑上安装 Python
如果还没有安装,请从 Python官网 下载并安装最新版的Python 3。
2. 安装必要的 Python 库
打开你的终端或命令行工具,运行以下命令:
pip install pure-python-adb opencv-python numpy requests Pillow
3. 安装 ADB 工具
- 前往 Google Android SDK Platform-Tools 页面 下载对应你操作系统的工具包。
- 解压后,你会得到一个名为
platform-tools的文件夹。为了方便,请将这个文件夹的路径添加到系统的环境变量PATH中。这样你就可以在任何地方直接使用adb命令。 - 验证安装: 连接手机后,在终端输入
adb devices,如果看到你的设备序列号,说明ADB配置成功。
4. 手机端设置
- 开启开发者模式: 通常在 “设置” -> “关于手机” -> 连续点击 “版本号” 7次。
- 开启USB调试: 在 “设置” -> “系统” -> “开发者选项” 中,找到并打开 “USB调试”。
- 用USB线将手机连接到电脑,手机上会弹出“是否允许USB调试”的提示,选择“一律允许”并确定。
5. 设置钉钉机器人
- 在你的钉钉群聊中,点击 “群设置” -> “智能群助手” -> “添加机器人”。
- 选择 “自定义” 机器人。
- 设置一个机器人名字(例如“股票盯盘助手”),在“安全设置”中,选择“自定义关键词”并添加一个关键词,比如“信号”。(非常重要,之后发送的消息里必须包含这个关键词)
- 完成添加后,复制生成的 Webhook 地址,它看起来像这样:
https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxx。我们稍后会用到它。
步骤二:准备“S”和“G”的模板图片
这是图像识别最关键的一步。我们需要从你的APP截图中,把 ‘S’ 和 ‘G’ 这两个信号图标精确地裁剪出来,作为识别的“模板”。
- 获取高清截图: 在你的手机上,截取一张同时包含 ‘S’ 和 ‘G’ 信号的清晰图片,并将其传到电脑上。
- 裁剪模板: 使用任何图片编辑工具(Windows自带的画图、Photoshop等)。
- 放大图片,精确地只裁剪 ‘S’ 字母(包括其背景色),保存为
template_s.png。 - 同样地,精确裁剪 ‘G’ 字母,保存为
template_g.png。
- 放大图片,精确地只裁剪 ‘S’ 字母(包括其背景色),保存为
裁剪要求:
- 裁剪框要尽量紧凑,只包含目标字母和一点点背景。
- 背景尽量单一。
- 将这两个
.png文件和你之后要创建的Python脚本放在同一个文件夹下。
步骤三:编写 Python 脚本
在存放模板图片的文件夹中,创建一个新的Python文件,例如 stock_monitor.py,然后将下面的代码复制进去。
import cv2
import numpy as np
import requests
import time
import os
from adb.client import Client as AdbClient
# ==============================================================================
# 配置区域 - 你需要根据你的情况修改这里
# ==============================================================================
CONFIG = {
# 你的手机设备序列号。在终端运行 `adb devices` 查看。
# 如果只有一台设备,可以留空或设置为 None。
"ADB_DEVICE_SERIAL": None,
# 从钉钉机器人设置中获取的 Webhook URL
"DINGTALK_WEBHOOK": "https://oapi.dingtalk.com/robot/send?access_token=YOUR_ACCESS_TOKEN_HERE",
# 钉钉机器人的关键词
"DINGTALK_KEYWORD": "信号",
# 检查间隔(秒)
"CHECK_INTERVAL_SECONDS": 300, # 5分钟
# 截图在手机上的临时保存路径
"SCREENSHOT_REMOTE_PATH": "/sdcard/screen.png",
# 截图下载到电脑上的路径
"SCREENSHOT_LOCAL_PATH": "screen.png",
# S 和 G 信号的模板图片文件名
"TEMPLATE_S_PATH": "template_s.png",
"TEMPLATE_G_PATH": "template_g.png",
# 【重要】定义要搜索的区域 (Region of Interest - ROI)
# 这是最后一个K线柱可能出现的位置,以避免在历史数据中找到信号
# 格式: (x_start, y_start, width, height)
# 你需要用画图工具打开你的手机截图,找到这个区域的像素坐标
# 根据你的截图,一个大概的区域可能是:从右侧向左200像素,高度覆盖K线图主体
# 示例坐标(你需要自己精确调整): (880, 280, 200, 350) 假设你的手机分辨率是 1080x2340
"ROI_COORDS": (850, 250, 230, 450),
# 图像匹配的置信度阈值 (0.0 ~ 1.0),越高越精确
"TEMPLATE_THRESHOLD": 0.85,
# 模拟点击刷新页面的坐标 (x, y)。点击"日K"标签是一个不错的选择,可以刷新视图
# 根据你的截图,"日K"的大概位置在 (200, 450) 附近
# 示例坐标(你需要自己精确调整):
"REFRESH_TAP_COORDS": (200, 450),
# K线图的公司名称或代码,用于发送通知
"STOCK_NAME": "汇量科技 (HK1860)"
}
# ==============================================================================
# 全局变量,用于防止重复发送同一个信号
last_detected_signal = {"type": None, "time": 0}
def setup_adb():
"""连接到ADB服务并获取设备"""
try:
client = AdbClient(host="127.0.0.1", port=5037)
if CONFIG["ADB_DEVICE_SERIAL"]:
device = client.device(CONFIG["ADB_DEVICE_SERIAL"])
else:
devices = client.devices()
if not devices:
print("错误:未找到任何ADB设备。请确保手机已连接并开启USB调试。")
return None
device = devices[0]
print(f"成功连接到设备: {device.serial}")
return device
except Exception as e:
print(f"ADB连接失败: {e}")
return None
def send_dingtalk_message(message):
"""发送消息到钉钉机器人"""
headers = {'Content-Type': 'application/json;charset=utf-8'}
data = {
"msgtype": "text",
"text": {
"content": f"{CONFIG['DINGTALK_KEYWORD']} - {message}"
}
}
try:
r = requests.post(CONFIG["DINGTALK_WEBHOOK"], json=data, headers=headers)
if r.json().get("errcode") == 0:
print("钉钉消息发送成功!")
else:
print(f"钉钉消息发送失败: {r.text}")
except Exception as e:
print(f"发送钉钉消息时发生网络错误: {e}")
def capture_and_pull_screenshot(device):
"""使用ADB截图并将其下载到本地"""
try:
device.shell(f"screencap -p {CONFIG['SCREENSHOT_REMOTE_PATH']}")
device.pull(CONFIG['SCREENSHOT_REMOTE_PATH'], CONFIG['SCREENSHOT_LOCAL_PATH'])
# print("截图已成功获取。")
return True
except Exception as e:
print(f"截图失败: {e}")
return False
def refresh_app_data(device):
"""模拟点击以刷新APP数据"""
x, y = CONFIG['REFRESH_TAP_COORDS']
try:
device.shell(f'input tap {x} {y}')
print(f"已模拟点击坐标 ({x}, {y}) 以刷新数据。")
except Exception as e:
print(f"模拟点击失败: {e}")
def find_signal_in_roi():
"""在截图的指定区域中寻找S或G信号"""
if not os.path.exists(CONFIG['SCREENSHOT_LOCAL_PATH']):
print("错误:本地截图文件不存在。")
return None
main_image = cv2.imread(CONFIG['SCREENSHOT_LOCAL_PATH'], cv2.IMREAD_COLOR)
template_s = cv2.imread(CONFIG['TEMPLATE_S_PATH'], cv2.IMREAD_COLOR)
template_g = cv2.imread(CONFIG['TEMPLATE_G_PATH'], cv2.IMREAD_COLOR)
if main_image is None or template_s is None or template_g is None:
print("错误:无法读取图片文件,请检查截图和模板图片路径是否正确。")
return None
# 提取感兴趣的区域 (ROI)
x, y, w, h = CONFIG['ROI_COORDS']
roi = main_image[y:y+h, x:x+w]
# 寻找 'S' 信号
res_s = cv2.matchTemplate(roi, template_s, cv2.TM_CCOEFF_NORMED)
_, max_val_s, _, _ = cv2.minMaxLoc(res_s)
# 寻找 'G' 信号
res_g = cv2.matchTemplate(roi, template_g, cv2.TM_CCOEFF_NORMED)
_, max_val_g, _, _ = cv2.minMaxLoc(res_g)
# print(f"信号置信度 -> S: {max_val_s:.2f}, G: {max_val_g:.2f}")
if max_val_s > CONFIG["TEMPLATE_THRESHOLD"]:
return "S"
if max_val_g > CONFIG["TEMPLATE_THRESHOLD"]:
return "G"
return None
def main():
"""主循环函数"""
device = setup_adb()
if not device:
return
while True:
print("\n" + "="*30)
print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - 开始新一轮检测...")
# 1. 刷新数据
refresh_app_data(device)
time.sleep(3) # 等待APP响应刷新
# 2. 截图
if not capture_and_pull_screenshot(device):
print("无法获取截图,跳过本轮检测。")
time.sleep(CONFIG["CHECK_INTERVAL_SECONDS"])
continue
# 3. 图像分析
signal = find_signal_in_roi()
current_time = time.time()
if signal:
# 检查是否是短时间内重复检测到的同一个信号
if (signal == last_detected_signal["type"] and
current_time - last_detected_signal["time"] < CONFIG["CHECK_INTERVAL_SECONDS"] * 2):
print(f"发现信号 '{signal}',但因为是近期已通知过的信号,本次不重复发送。")
else:
print(f"!!! 在指定区域内发现新信号: {signal} !!!")
message = f"【{CONFIG['STOCK_NAME']}】出现新的交易信号: **{signal}**,请关注!"
send_dingtalk_message(message)
# 更新最后检测到的信号
last_detected_signal["type"] = signal
last_detected_signal["time"] = current_time
else:
print("在指定区域内未发现S或G信号。")
# 4. 等待下一轮
print(f"本轮检测结束,等待 {CONFIG['CHECK_INTERVAL_SECONDS']} 秒...")
time.sleep(CONFIG["CHECK_INTERVAL_SECONDS"])
if __name__ == "__main__":
main()
步骤四:配置并运行脚本
-
修改配置:
- 最重要: 打开
stock_monitor.py文件,仔细修改CONFIG字典中的内容。 DINGTALK_WEBHOOK: 粘贴你自己的钉钉机器人Webhook地址。ROI_COORDS: 这是最需要调试的参数。你需要用一个截图查看软件(如 FastStone Image Viewer, IrfanView, 甚至Windows画图)打开你的手机截图,将鼠标移动到K线图最后一个柱子可能出现的区域,记录下左上角坐标(x, y)和这个区域的宽度(width)、高度(height)。多调试几次才能找到最佳区域。REFRESH_TAP_COORDS: 同理,找到“日K”这个按钮在屏幕上的像素坐标(x, y)。ADB_DEVICE_SERIAL: 如果你只连接了一台手机,可以不用管。如果有多台,请填入你要操作的手机序列号。
- 最重要: 打开
-
运行脚本:
- 确保你的手机已通过USB连接,USB调试已开启,并且屏幕已解锁,目标APP在前台显示。
- 打开终端,
cd到脚本和模板图片所在的文件夹。 - 运行命令:
python stock_monitor.py
现在,脚本会开始工作。它会每隔5分钟,自动点击刷新、截图、分析,并在发现信号时给你发送钉钉消息。
常见问题和调试技巧
- 坐标不准: 这是最常见的问题。不同手机分辨率不同,
ROI_COORDS和REFRESH_TAP_COORDS必须根据你自己的手机来设定。多试几次。 - 识别不到信号:
- 可能是
TEMPLATE_THRESHOLD太高了,可以适当调低,比如0.8。 - 也可能是你裁剪的
template_s.png或template_g.png质量不好。尝试重新裁剪。 - 检查
ROI_COORDS是否正确覆盖了最新K线柱的区域。
- 可能是
- ADB连接问题: 确保手机驱动安装正确,USB调试已授权。可以运行
adb devices查看设备状态是否为device而不是unauthorized。 - APP更新: 如果股票APP界面更新了,图标或者布局发生变化,这个脚本可能会失效。届时你需要重新裁剪模板图片和调整坐标。
火山引擎开发者社区是火山引擎打造的AI技术生态平台,聚焦Agent与大模型开发,提供豆包系列模型(图像/视频/视觉)、智能分析与会话工具,并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长,新用户可领50万Tokens权益,助力构建智能应用。
更多推荐
所有评论(0)