AI 编写简单的 Python 脚本来实现,例如调用新浪财经等免费的股票数据接口
AI 自己编写一个简单的 Python 脚本来实现,例如调用新浪财经等免费的股票数据接口。
这是一个非常实用的实战场景。既然你已经配置好了 OpenClaw/ClawX,我们可以直接利用 AI 的代码执行能力来完成这个任务。
针对你的需求(监控几百只股票、触发止盈、微信提醒),最稳健的方案是让 AI 编写一个 Python 脚本,配合系统的定时任务(Crontab)运行。
以下是具体的实现步骤和代码,你可以直接复制给 ClawX 中的 AI,或者在 ClawX 的“代码解释器/技能”中运行。
🛠️ 核心方案:Python 脚本 + 新浪财经接口 + 微信推送
第一步:准备“微信通知”通道
在写代码前,先解决“怎么发微信”的问题。最简单且免费的方法是使用 Server酱 或 PushPlus。
- 去 Server酱官网 微信扫码登录。
- 获取你的
SEND_KEY(一串字符)。 - 以后只要访问特定的 URL,你的微信就会收到消息。
第二步:让 AI 编写监控脚本
你可以直接把下面的提示词发给 ClawX 的 AI,让它生成脚本:
提示词:
“请帮我写一个 Python 脚本stock_monitor.py。
- 读取我本地一个名为
watchlist.txt的文件,里面每行是一只股票的代码(如sh600519)。- 调用新浪财经接口获取这些股票的实时价格。
- 对比我预设的‘止盈价’(可以在 txt 文件里备注,或者在脚本里用字典定义)。
- 如果当前价格 >= 止盈价,调用 Server酱的 API 发送微信通知,内容包含:股票名称、当前价格、涨幅。
- 做好异常处理,防止因为网络波动报错。”
第三步:参考代码(AI 生成的版本)
如果 AI 生成的代码有问题,你可以直接使用下面这个经过优化的版本。它使用了 requests 库,并模拟了浏览器头,防止被新浪屏蔽。
python
1import requests
2import time
3import logging
4
5# --- 配置区域 ---
6# 你的 Server酱 SEND_KEY
7SEND_KEY = "你的_SEND_KEY_在这里"
8# 监控列表:代码: 股票名: 止盈价
9WATCH_LIST = {
10 "sh600519": ["贵州茅台", 1800],
11 "sz00700": ["腾讯控股", 350], # 示例
12 # 你可以在这里加几百只,或者改成读取 txt 文件
13}
14
15# 设置日志
16logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
17
18def get_stock_data(code):
19 """调用新浪财经接口获取实时行情"""
20 url = f"http://hq.sinajs.cn/list={code}"
21 headers = {
22 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
23 }
24 try:
25 resp = requests.get(url, headers=headers, timeout=2)
26 if resp.status_code == 200:
27 # 新浪返回格式: var hq_str_sh600519="贵州茅台,1750.00,..."
28 text = resp.text
29 if "=" in text:
30 data = text.split('=')[1].strip().strip('"').split(',')
31 return {
32 "name": data[0],
33 "price": float(data[3]), # 当前价通常是第4个字段
34 "open": float(data[1]), # 今开
35 "change": float(data[3]) - float(data[2]) # 简略计算涨跌
36 }
37 return None
38 except Exception as e:
39 logging.error(f"获取 {code} 数据失败: {e}")
40 return None
41
42def send_wechat_msg(title, content):
43 """发送微信通知"""
44 if not SEND_KEY or SEND_KEY == "你的_SEND_KEY_在这里":
45 logging.warning("未配置 SEND_KEY,跳过发送微信")
46 return
47
48 url = f"https://sctapi.ftqq.com/{SEND_KEY}.send"
49 data = {"title": title, "desp": content}
50 try:
51 requests.post(url, data=data)
52 logging.info(f"微信发送成功: {title}")
53 except Exception as e:
54 logging.error(f"微信发送失败: {e}")
55
56def main():
57 logging.info("开始本轮股票监控...")
58 for code, info in WATCH_LIST.items():
59 target_price = info[1] # 止盈价
60 data = get_stock_data(code)
61
62 if data:
63 current_price = data['price']
64 # 触发止盈逻辑
65 if current_price >= target_price:
66 msg_title = f"⚠️ {data['name']} 触发止盈提醒!"
67 msg_content = f"股票:{data['name']} ({code})\n当前价:**{current_price}**\n目标价:{target_price}\n\n建议关注后续走势,考虑分批止盈。"
68 send_wechat_msg(msg_title, msg_content)
69 else:
70 logging.info(f"{data['name']} 当前 {current_price},未达止盈 {target_price}")
71
72 # 稍微休眠一下,防止请求过快被封 IP
73 time.sleep(0.5)
74
75if __name__ == "__main__":
76 main()
第四步:设置定时任务
脚本写好后,你不能一直手动运行。你需要让 ClawX 或者你的服务器/电脑自动运行它。
-
如果你在用 ClawX (Linux/Docker 环境):
你可以直接告诉 AI:“帮我设置一个 Crontab 任务,每 5 分钟运行一次这个stock_monitor.py脚本。”
AI 会生成类似这样的命令让你执行:*/5 9-15 * * 1-5 python3 /path/to/stock_monitor.py
(意思是:周一到周五,9点到15点,每5分钟执行一次) -
如果你是在本地 Windows 电脑:
可以使用“任务计划程序”来定时运行这个 Python 脚本。
💡 进阶技巧
- 批量监控几百只:上面的代码是轮询,几百只可能会慢。如果需要极速监控,可以让 AI 把代码改成使用
concurrent.futures进行多线程并发请求,速度能提升几十倍。 - 数据源:新浪财经接口是免费的,但对于高频交易可能不够快。如果是做超短线,建议换成腾讯财经接口或专业的量化接口(如 Tushare,需要 Token)。
- 止盈策略:你可以在脚本里写更复杂的逻辑,比如“连续 3 次检测都高于止盈价再报警”,防止瞬间波动误报。
现在,你只需要把上面的代码丢给 ClawX,让它帮你保存并配置环境,你的私人股票盯盘员就上岗了。
import requests import time import logging # --- 配置区域 --- # 你的 Server酱 SEND_KEY SEND_KEY = "你的_SEND_KEY_在这里" # 监控列表:代码: 股票名: 止盈价 WATCH_LIST = { "sh600519": ["贵州茅台", 1800], "sz00700": ["腾讯控股", 350], # 示例 # 你可以在这里加几百只,或者改成读取 txt 文件 } # 设置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def get_stock_data(code): """调用新浪财经接口获取实时行情""" url = f"http://hq.sinajs.cn/list={code}" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" } try: resp = requests.get(url, headers=headers, timeout=2) if resp.status_code == 200: # 新浪返回格式: var hq_str_sh600519="贵州茅台,1750.00,..." text = resp.text if "=" in text: data = text.split('=')[1].strip().strip('"').split(',') return { "name": data[0], "price": float(data[3]), # 当前价通常是第4个字段 "open": float(data[1]), # 今开 "change": float(data[3]) - float(data[2]) # 简略计算涨跌 } return None except Exception as e: logging.error(f"获取 {code} 数据失败: {e}") return None def send_wechat_msg(title, content): """发送微信通知""" if not SEND_KEY or SEND_KEY == "你的_SEND_KEY_在这里": logging.warning("未配置 SEND_KEY,跳过发送微信") return url = f"https://sctapi.ftqq.com/{SEND_KEY}.send" data = {"title": title, "desp": content} try: requests.post(url, data=data) logging.info(f"微信发送成功: {title}") except Exception as e: logging.error(f"微信发送失败: {e}") def main(): logging.info("开始本轮股票监控...") for code, info in WATCH_LIST.items(): target_price = info[1] # 止盈价 data = get_stock_data(code) if data: current_price = data['price'] # 触发止盈逻辑 if current_price >= target_price: msg_title = f"⚠️ {data['name']} 触发止盈提醒!" msg_content = f"股票:{data['name']} ({code})\n当前价:**{current_price}**\n目标价:{target_price}\n\n建议关注后续走势,考虑分批止盈。" send_wechat_msg(msg_title, msg_content) else: logging.info(f"{data['name']} 当前 {current_price},未达止盈 {target_price}") # 稍微休眠一下,防止请求过快被封 IP time.sleep(0.5) if __name__ == "__main__": main() 逐行解释
这是一个非常实用的量化监控脚本。为了让你完全理解它的运行逻辑,我将逐行、逐模块地为你解释这段代码的作用。
🛠️ 第一部分:导入依赖库与配置
python
编辑
1import requests
2import time
3import logging
- 解释:导入脚本运行所需的外部库。
requests:用于发送 HTTP 请求(比如访问新浪股票接口、发送微信消息)。time:用于处理时间,比如让程序暂停几秒,防止请求太快。logging:用于在终端输出日志信息(运行状态、错误提示)。
python
编辑
1# --- 配置区域 ---
2# 你的 Server酱 SEND_KEY
3SEND_KEY = "你的_SEND_KEY_在这里"
- 解释:这是你接收微信推送的密钥。你需要去 Server酱官网获取一个 KEY,填在这里才能发微信。如果留空或没改,脚本会跳过发送微信的步骤。
python
编辑
1# 监控列表:代码: 股票名: 止盈价
2WATCH_LIST = {
3 "sh600519": ["贵州茅台", 1800],
4 "sz00700": ["腾讯控股", 350], # 示例
5 # 你可以在这里加几百只,或者改成读取 txt 文件
6}
- 解释:这是你的“监控清单”。它是一个字典(Dictionary)。
- Key (键):股票代码,如
"sh600519"(sh代表上海,sz代表深圳)。 - Value (值):一个列表,包含
[股票名称, 止盈价格]。 - 作用:脚本会遍历这个列表里的每一项,去查价格并判断是否需要报警。
- Key (键):股票代码,如
python
编辑
1# 设置日志
2logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
- 解释:配置日志的显示格式。
level=logging.INFO:表示只显示 INFO 级别及以上的信息(INFO 是普通信息,ERROR 是错误信息)。format:定义了日志的前缀,会显示时间、日志级别和具体内容。
📡 第二部分:定义函数 - 获取股票数据
python
编辑
1def get_stock_data(code):
2 """调用新浪财经接口获取实时行情"""
- 解释:定义一个名为
get_stock_data的函数,它接收一个参数code(股票代码),用来专门负责获取单只股票的数据。
python
编辑
1 url = f"http://hq.sinajs.cn/list={code}"
- 解释:构造请求的 URL。
- 新浪财经的接口格式是
http://hq.sinajs.cn/list=股票代码。 f-string会把变量code的值插入到 URL 中。
- 新浪财经的接口格式是
python
编辑
1 headers = {
2 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
3 }
- 解释:定义请求头(Headers)。
- 服务器会检查是谁在访问。如果不加这个头,服务器可能会认为是机器人(爬虫)并拒绝服务。
- 这里伪装成 Chrome 浏览器进行访问。
python
编辑
1 try:
2 resp = requests.get(url, headers=headers, timeout=2)
- 解释:发送 GET 请求。
requests.get():发起网络请求。timeout=2:设置超时时间为 2 秒。如果 2 秒内没收到回复,就放弃,防止程序卡死。
python
编辑
1 if resp.status_code == 200:
- 解释:检查响应状态码。
200代表请求成功。如果不是 200,说明网络有问题或接口失效。
python
编辑
1 # 新浪返回格式: var hq_str_sh600519="贵州茅台,1750.00,..."
2 text = resp.text
3 if "=" in text:
4 data = text.split('=')[1].strip().strip('"').split(',')
- 解释:解析返回的文本数据(这是最核心的数据处理部分)。
- 新浪返回的不是标准的 JSON,而是一段 JavaScript 代码字符串,例如:
var hq_str_sh600519="贵州茅台,1650.00,1640.00,1660.00..."。 text.split('='):以等号=切割字符串,取后面的部分(即引号内的内容)。[1]:取切割后的第二部分(索引为1)。.strip().strip('"'):去除首尾的空白字符和引号"。.split(','):以逗号,切割字符串,得到一个包含所有字段的列表(如:[名称, 今日开盘价, 昨日收盘价, 当前价, ...])。
- 新浪返回的不是标准的 JSON,而是一段 JavaScript 代码字符串,例如:
python
编辑
1 return {
2 "name": data[0],
3 "price": float(data[3]), # 当前价通常是第4个字段
4 "open": float(data[1]), # 今开
5 "change": float(data[3]) - float(data[2]) # 简略计算涨跌
6 }
- 解释:整理数据并返回。
data[0]是股票名称。data[3]是当前价格(字符串转为浮点数float)。data[1]是今日开盘价。change是涨跌额,用当前价减去昨日收盘价(data[2])计算得出。
python
编辑
1 return None
2 except Exception as e:
3 logging.error(f"获取 {code} 数据失败: {e}")
4 return None
- 解释:错误处理。
- 如果网络请求失败(比如超时、断网)或者数据解析出错,
try块会报错,程序会跳到except块。 logging.error:在终端打印红色的错误日志。- 最终返回
None,表示这次查询失败了。
- 如果网络请求失败(比如超时、断网)或者数据解析出错,
📢 第三部分:定义函数 - 发送微信消息
python
编辑
1def send_wechat_msg(title, content):
2 """发送微信通知"""
- 解释:定义一个发送微信消息的函数,接收标题和内容两个参数。
python
编辑
1 if not SEND_KEY or SEND_KEY == "你的_SEND_KEY_在这里":
2 logging.warning("未配置 SEND_KEY,跳过发送微信")
3 return
- 解释:防御性检查。
- 如果你忘记改上面的
SEND_KEY,或者留空了,程序会直接退出这个函数,不再执行发送操作,防止报错。
- 如果你忘记改上面的
python
编辑
1 url = f"https://sctapi.ftqq.com/{SEND_KEY}.send"
2 data = {"title": title, "desp": content}
- 解释:构造 Server酱的请求参数。
url:Server酱的 API 地址,需要把你的 KEY 插入其中。data:发送的数据体。title是消息标题,desp是消息内容(支持 Markdown 格式)。
python
编辑
1 try:
2 requests.post(url, data=data)
3 logging.info(f"微信发送成功: {title}")
4 except Exception as e:
5 logging.error(f"微信发送失败: {e}")
- 解释:发送 POST 请求。
- 如果发送成功,打印成功日志;如果失败(如网络问题),打印错误日志。
⚙️ 第四部分:主程序逻辑
python
编辑
1def main():
2 logging.info("开始本轮股票监控...")
- 解释:定义主函数。程序开始运行时,先打印一条日志表示开始。
python
编辑
1 for code, info in WATCH_LIST.items():
- 解释:遍历监控列表。
WATCH_LIST.items()会把字典里的每一项(键值对)取出来。code是股票代码(如 sh600519)。info是对应的列表(如 ["贵州茅台", 1800])。
python
编辑
1 target_price = info[1] # 止盈价
2 data = get_stock_data(code)
- 解释:
info[1]取出列表里的第二个元素,即你设定的止盈价格。get_stock_data(code)调用上面的函数,获取这只股票的实时数据。
python
编辑
1 if data:
2 current_price = data['price']
3 # 触发止盈逻辑
4 if current_price >= target_price:
5 msg_title = f"⚠️ {data['name']} 触发止盈提醒!"
6 msg_content = f"股票:{data['name']} ({code})\n当前价:**{current_price}**\n目标价:{target_price}\n\n建议关注后续走势,考虑分批止盈。"
7 send_wechat_msg(msg_title, msg_content)
- 解释:核心判断逻辑。
if data::先判断刚才获取数据是否成功(如果不为空)。if current_price >= target_price::判断当前价格是否大于等于你设定的止盈价。- 如果条件成立,就拼接消息的标题和内容(
msg_content里用了 Markdown 的加粗语法**),然后调用send_wechat_msg发送。
python
编辑
1 else:
2 logging.info(f"{data['name']} 当前 {current_price},未达止盈 {target_price}")
- 解释:如果价格没达到止盈点,就在终端打印一条普通日志,告诉你目前的价格状态。
python
编辑
1 # 稍微休眠一下,防止请求过快被封 IP
2 time.sleep(0.5)
- 解释:程序暂停 0.5 秒。
- 如果你监控几百只股票,如果不暂停,瞬间发出几百个请求,新浪服务器会认为你在攻击,从而封禁你的 IP 地址。这里做一个简单的节流控制。
▶️ 第五部分:程序入口
python
编辑
1if __name__ == "__main__":
2 main()
- 解释:这是 Python 的标准写法。
- 意思是:“如果这个脚本是直接运行的(而不是被别人 import 导入的),就执行
main()函数”。 - 这是程序的起点。
- 意思是:“如果这个脚本是直接运行的(而不是被别人 import 导入的),就执行
📌 总结
这段代码的运行流程是:
- 读取你设定的股票列表。
- 依次去新浪接口查询每只股票的实时价格。
- 拿实时价格和你设定的止盈价对比。
- 如果涨到了,就通过 Server酱发微信给你;没涨到就记录一下日志。
- 程序结束(配合定时任务,比如每5分钟跑一次,就能实现全天监控)。
AI 自己编写一个简单的 Python 脚本来实现,例如调用新浪财经等免费的股票数据接口。
针对你的需求(监控几百只股票、触发止盈、微信提醒),最稳健的方案是让 AI 编写一个 Python 脚本,配合系统的定时任务(Crontab)运行。
以下是具体的实现步骤和代码,你可以直接复制给 ClawX 中的 AI,或者在 ClawX 的“代码解释器/技能”中运行。
🛠️ 核心方案:Python 脚本 + 新浪财经接口 + 微信推送
第一步:准备“微信通知”通道
在写代码前,先解决“怎么发微信”的问题。最简单且免费的方法是使用 Server酱 或 PushPlus。
去 Server酱官网 微信扫码登录。
获取你的 SEND_KEY(一串字符)。
以后只要访问特定的 URL,你的微信就会收到消息。
第二步:让 AI 编写监控脚本
你可以直接把下面的提示词发给 ClawX 的 AI,让它生成脚本:
提示词:
“请帮我写一个 Python 脚本 stock_monitor.py。
读取我本地一个名为 watchlist.txt 的文件,里面每行是一只股票的代码(如 sh600519)。
调用新浪财经接口获取这些股票的实时价格。
对比我预设的‘止盈价’(可以在 txt 文件里备注,或者在脚本里用字典定义)。
如果当前价格 >= 止盈价,调用 Server酱的 API 发送微信通知,内容包含:股票名称、当前价格、涨幅。
做好异常处理,防止因为网络波动报错。”
第三步:参考代码(AI 生成的版本)
如果 AI 生成的代码有问题,你可以直接使用下面这个经过优化的版本。它使用了 requests 库,并模拟了浏览器头,防止被新浪屏蔽。
python
import requests
import time
import logging
# --- 配置区域 ---
# 你的 Server酱 SEND_KEY
SEND_KEY = "你的_SEND_KEY_在这里"
# 监控列表:代码: 股票名: 止盈价
WATCH_LIST = {
"sh600519": ["贵州茅台", 1800],
"sz00700": ["腾讯控股", 350], # 示例
# 你可以在这里加几百只,或者改成读取 txt 文件
}
# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def get_stock_data(code):
"""调用新浪财经接口获取实时行情"""
url = f"http://hq.sinajs.cn/list={code}"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
}
try:
resp = requests.get(url, headers=headers, timeout=2)
if resp.status_code == 200:
# 新浪返回格式: var hq_str_sh600519="贵州茅台,1750.00,..."
text = resp.text
if "=" in text:
data = text.split('=')[1].strip().strip('"').split(',')
return {
"name": data[0],
"price": float(data[3]), # 当前价通常是第4个字段
"open": float(data[1]), # 今开
"change": float(data[3]) - float(data[2]) # 简略计算涨跌
}
return None
except Exception as e:
logging.error(f"获取 {code} 数据失败: {e}")
return None
def send_wechat_msg(title, content):
"""发送微信通知"""
if not SEND_KEY or SEND_KEY == "你的_SEND_KEY_在这里":
logging.warning("未配置 SEND_KEY,跳过发送微信")
return
url = f"https://sctapi.ftqq.com/{SEND_KEY}.send"
data = {"title": title, "desp": content}
try:
requests.post(url, data=data)
logging.info(f"微信发送成功: {title}")
except Exception as e:
logging.error(f"微信发送失败: {e}")
def main():
logging.info("开始本轮股票监控...")
for code, info in WATCH_LIST.items():
target_price = info[1] # 止盈价
data = get_stock_data(code)
if data:
current_price = data['price']
# 触发止盈逻辑
if current_price >= target_price:
msg_title = f"⚠️ {data['name']} 触发止盈提醒!"
msg_content = f"股票:{data['name']} ({code})\n当前价:**{current_price}**\n目标价:{target_price}\n\n建议关注后续走势,考虑分批止盈。"
send_wechat_msg(msg_title, msg_content)
else:
logging.info(f"{data['name']} 当前 {current_price},未达止盈 {target_price}")
# 稍微休眠一下,防止请求过快被封 IP
time.sleep(0.5)
if __name__ == "__main__":
main()
第四步:设置定时任务
脚本写好后,你不能一直手动运行。你需要让 ClawX 或者你的服务器/电脑自动运行它。
如果你在用 ClawX (Linux/Docker 环境):
你可以直接告诉 AI:“帮我设置一个 Crontab 任务,每 5 分钟运行一次这个 stock_monitor.py 脚本。”
AI 会生成类似这样的命令让你执行:
*/5 9-15 * * 1-5 python3 /path/to/stock_monitor.py
(意思是:周一到周五,9点到15点,每5分钟执行一次)
如果你是在本地 Windows 电脑:
可以使用“任务计划程序”来定时运行这个 Python 脚本。
💡 进阶技巧
批量监控几百只:上面的代码是轮询,几百只可能会慢。如果需要极速监控,可以让 AI 把代码改成使用 concurrent.futures 进行多线程并发请求,速度能提升几十倍。
数据源:新浪财经接口是免费的,但对于高频交易可能不够快。如果是做超短线,建议换成腾讯财经接口或专业的量化接口(如 Tushare,需要 Token)。
止盈策略:你可以在脚本里写更复杂的逻辑,比如“连续 3 次检测都高于止盈价再报警”,防止瞬间波动误报。
现在,你只需要把上面的代码丢给 ClawX,让它帮你保存并配置环境,你的私人股票盯盘员就上岗了。
import requests
import time
import logging
# --- 配置区域 ---
# 你的 Server酱 SEND_KEY
SEND_KEY = "你的_SEND_KEY_在这里"
# 监控列表:代码: 股票名: 止盈价
WATCH_LIST = {
"sh600519": ["贵州茅台", 1800],
"sz00700": ["腾讯控股", 350], # 示例
# 你可以在这里加几百只,或者改成读取 txt 文件
}
# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def get_stock_data(code):
"""调用新浪财经接口获取实时行情"""
url = f"http://hq.sinajs.cn/list={code}"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
}
try:
resp = requests.get(url, headers=headers, timeout=2)
if resp.status_code == 200:
# 新浪返回格式: var hq_str_sh600519="贵州茅台,1750.00,..."
text = resp.text
if "=" in text:
data = text.split('=')[1].strip().strip('"').split(',')
return {
"name": data[0],
"price": float(data[3]), # 当前价通常是第4个字段
"open": float(data[1]), # 今开
"change": float(data[3]) - float(data[2]) # 简略计算涨跌
}
return None
except Exception as e:
logging.error(f"获取 {code} 数据失败: {e}")
return None
def send_wechat_msg(title, content):
"""发送微信通知"""
if not SEND_KEY or SEND_KEY == "你的_SEND_KEY_在这里":
logging.warning("未配置 SEND_KEY,跳过发送微信")
return
url = f"https://sctapi.ftqq.com/{SEND_KEY}.send"
data = {"title": title, "desp": content}
try:
requests.post(url, data=data)
logging.info(f"微信发送成功: {title}")
except Exception as e:
logging.error(f"微信发送失败: {e}")
def main():
logging.info("开始本轮股票监控...")
for code, info in WATCH_LIST.items():
target_price = info[1] # 止盈价
data = get_stock_data(code)
if data:
current_price = data['price']
# 触发止盈逻辑
if current_price >= target_price:
msg_title = f"⚠️ {data['name']} 触发止盈提醒!"
msg_content = f"股票:{data['name']} ({code})\n当前价:**{current_price}**\n目标价:{target_price}\n\n建议关注后续走势,考虑分批止盈。"
send_wechat_msg(msg_title, msg_content)
else:
logging.info(f"{data['name']} 当前 {current_price},未达止盈 {target_price}")
# 稍微休眠一下,防止请求过快被封 IP
time.sleep(0.5)
if __name__ == "__main__":
main() 逐行解释
这是一个非常实用的量化监控脚本。
为了让你完全理解它的运行逻辑,我将逐行、逐模块地为你解释这段代码的作用。
🛠️ 第一部分:导入依赖库与配置
python
import requests
import time
import logging
解释:导入脚本运行所需的外部库。
requests:用于发送 HTTP 请求(比如访问新浪股票接口、发送微信消息)。
time:用于处理时间,比如让程序暂停几秒,防止请求太快。
logging:用于在终端输出日志信息(运行状态、错误提示)。
python
编辑
# --- 配置区域 ---
# 你的 Server酱 SEND_KEY
SEND_KEY = "你的_SEND_KEY_在这里"
解释:这是你接收微信推送的密钥。你需要去 Server酱官网获取一个 KEY,填在这里才能发微信。如果留空或没改,脚本会跳过发送微信的步骤。
python
编辑
# 监控列表:代码: 股票名: 止盈价
WATCH_LIST = {
"sh600519": ["贵州茅台", 1800],
"sz00700": ["腾讯控股", 350], # 示例
# 你可以在这里加几百只,或者改成读取 txt 文件
}
解释:这是你的“监控清单”。它是一个字典(Dictionary)。
Key (键):股票代码,如 "sh600519"(sh代表上海,sz代表深圳)。
Value (值):一个列表,包含 [股票名称, 止盈价格]。
作用:脚本会遍历这个列表里的每一项,去查价格并判断是否需要报警。
python
编辑
# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
解释:配置日志的显示格式。
level=logging.INFO:表示只显示 INFO 级别及以上的信息(INFO 是普通信息,ERROR 是错误信息)。
format:定义了日志的前缀,会显示时间、日志级别和具体内容。
📡 第二部分:定义函数 - 获取股票数据
python
编辑
def get_stock_data(code):
"""调用新浪财经接口获取实时行情"""
解释:定义一个名为 get_stock_data 的函数,它接收一个参数 code(股票代码),用来专门负责获取单只股票的数据。
python
编辑
url = f"http://hq.sinajs.cn/list={code}"
解释:构造请求的 URL。
新浪财经的接口格式是 http://hq.sinajs.cn/list=股票代码。
f-string 会把变量 code 的值插入到 URL 中。
python
编辑
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
}
解释:定义请求头(Headers)。
服务器会检查是谁在访问。如果不加这个头,服务器可能会认为是机器人(爬虫)并拒绝服务。
这里伪装成 Chrome 浏览器进行访问。
python
编辑
try:
resp = requests.get(url, headers=headers, timeout=2)
解释:发送 GET 请求。
requests.get():发起网络请求。
timeout=2:设置超时时间为 2 秒。如果 2 秒内没收到回复,就放弃,防止程序卡死。
python
编辑
if resp.status_code == 200:
解释:检查响应状态码。
200 代表请求成功。如果不是 200,说明网络有问题或接口失效。
python
编辑
# 新浪返回格式: var hq_str_sh600519="贵州茅台,1750.00,..."
text = resp.text
if "=" in text:
data = text.split('=')[1].strip().strip('"').split(',')
解释:解析返回的文本数据(这是最核心的数据处理部分)。
新浪返回的不是标准的 JSON,而是一段 JavaScript 代码字符串,例如:var hq_str_sh600519="贵州茅台,1650.00,1640.00,1660.00..."。
text.split('='):以等号 = 切割字符串,取后面的部分(即引号内的内容)。
[1]:取切割后的第二部分(索引为1)。
.strip().strip('"'):去除首尾的空白字符和引号 "。
.split(','):以逗号 , 切割字符串,得到一个包含所有字段的列表(如:[名称, 今日开盘价, 昨日收盘价, 当前价, ...])。
python
编辑
return {
"name": data[0],
"price": float(data[3]), # 当前价通常是第4个字段
"open": float(data[1]), # 今开
"change": float(data[3]) - float(data[2]) # 简略计算涨跌
}
解释:整理数据并返回。
data[0] 是股票名称。
data[3] 是当前价格(字符串转为浮点数 float)。
data[1] 是今日开盘价。
change 是涨跌额,用当前价减去昨日收盘价(data[2])计算得出。
python
编辑
return None
except Exception as e:
logging.error(f"获取 {code} 数据失败: {e}")
return None
解释:错误处理。
如果网络请求失败(比如超时、断网)或者数据解析出错,try 块会报错,程序会跳到 except 块。
logging.error:在终端打印红色的错误日志。
最终返回 None,表示这次查询失败了。
📢 第三部分:定义函数 - 发送微信消息
python
编辑
def send_wechat_msg(title, content):
"""发送微信通知"""
解释:定义一个发送微信消息的函数,接收标题和内容两个参数。
python
编辑
if not SEND_KEY or SEND_KEY == "你的_SEND_KEY_在这里":
logging.warning("未配置 SEND_KEY,跳过发送微信")
return
解释:防御性检查。
如果你忘记改上面的 SEND_KEY,或者留空了,程序会直接退出这个函数,不再执行发送操作,防止报错。
python
编辑
url = f"https://sctapi.ftqq.com/{SEND_KEY}.send"
data = {"title": title, "desp": content}
解释:构造 Server酱的请求参数。
url:Server酱的 API 地址,需要把你的 KEY 插入其中。
data:发送的数据体。title 是消息标题,desp 是消息内容(支持 Markdown 格式)。
python
编辑
try:
requests.post(url, data=data)
logging.info(f"微信发送成功: {title}")
except Exception as e:
logging.error(f"微信发送失败: {e}")
解释:发送 POST 请求。
如果发送成功,打印成功日志;如果失败(如网络问题),打印错误日志。
⚙️ 第四部分:主程序逻辑
python
编辑
def main():
logging.info("开始本轮股票监控...")
解释:定义主函数。程序开始运行时,先打印一条日志表示开始。
python
编辑
for code, info in WATCH_LIST.items():
解释:遍历监控列表。
WATCH_LIST.items() 会把字典里的每一项(键值对)取出来。
code 是股票代码(如 sh600519)。
info 是对应的列表(如 ["贵州茅台", 1800])。
python
编辑
target_price = info[1] # 止盈价
data = get_stock_data(code)
解释:
info[1] 取出列表里的第二个元素,即你设定的止盈价格。
get_stock_data(code) 调用上面的函数,获取这只股票的实时数据。
python
编辑
if data:
current_price = data['price']
# 触发止盈逻辑
if current_price >= target_price:
msg_title = f"⚠️ {data['name']} 触发止盈提醒!"
msg_content = f"股票:{data['name']} ({code})\n当前价:**{current_price}**\n目标价:{target_price}\n\n建议关注后续走势,考虑分批止盈。"
send_wechat_msg(msg_title, msg_content)
解释:核心判断逻辑。
if data::先判断刚才获取数据是否成功(如果不为空)。
if current_price >= target_price::判断当前价格是否大于等于你设定的止盈价。
如果条件成立,就拼接消息的标题和内容(msg_content 里用了 Markdown 的加粗语法 **),然后调用 send_wechat_msg 发送。
python
编辑
else:
logging.info(f"{data['name']} 当前 {current_price},未达止盈 {target_price}")
解释:如果价格没达到止盈点,就在终端打印一条普通日志,告诉你目前的价格状态。
python
编辑
# 稍微休眠一下,防止请求过快被封 IP
time.sleep(0.5)
解释:程序暂停 0.5 秒。
如果你监控几百只股票,如果不暂停,瞬间发出几百个请求,新浪服务器会认为你在攻击,从而封禁你的 IP 地址。这里做一个简单的节流控制。
▶️ 第五部分:程序入口
python
编辑
if __name__ == "__main__":
main()
解释:这是 Python 的标准写法。
意思是:“如果这个脚本是直接运行的(而不是被别人 import 导入的),就执行 main() 函数”。
这是程序的起点。
📌 总结
这段代码的运行流程是:
读取你设定的股票列表。
依次去新浪接口查询每只股票的实时价格。
拿实时价格和你设定的止盈价对比。
如果涨到了,就通过 Server酱发微信给你;没涨到就记录一下日志。
程序结束(配合定时任务,比如每5分钟跑一次,就能实现全天监控)。
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐


所有评论(0)