#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Windows 系统托盘实时股价显示程序（国内数据源版）
支持 A股、港股、美股
安装依赖:
pip install pystray pillow requests
使用方法:
python stock_tray_icon.py
"""
import sys
import threading
import time
import json
import os
import re
from datetime import datetime
# ==================== 依赖检查 ====================
MISSING_DEPS = []
try:
import requests
except ImportError:
MISSING_DEPS.append("requests")
try:
import pystray
from pystray import MenuItem, Menu
except ImportError:
MISSING_DEPS.append("pystray")
try:
from PIL import Image, ImageDraw, ImageFont
except ImportError:
MISSING_DEPS.append("Pillow")
if MISSING_DEPS:
print("[错误] 缺少以下依赖库，请先安装：")
print(f" pip install {' '.join(MISSING_DEPS)}")
print("\n安装完成后重新运行本程序。")
input("按 Enter 键退出...")
sys.exit(1)
# ==================== 配置 ====================
UPDATE_INTERVAL = 30
ICON_SIZE = 64
FONT_SIZE = 14
CONFIG_FILE = "stock_config.json"
REQUEST_TIMEOUT = 10
# ==================== 股票代码转换 ====================
def to_tencent_code(symbol: str) -> str:
"""将用户输入转换为腾讯财经接口格式"""
symbol = symbol.strip().upper()
if symbol.startswith(("SH", "SZ", "HK", "US")):
return symbol.lower()
if symbol.isdigit():
if len(symbol) <= 5 and not symbol.startswith(("6", "0", "3")):
symbol = symbol.zfill(5)
return f"hk{symbol}"
if symbol.startswith("6"):
return f"sh{symbol}"
elif symbol.startswith(("0", "3")):
return f"sz{symbol}"
else:
return f"sh{symbol}"
return f"us{symbol}"
def get_display_symbol(tencent_code: str) -> str:
if tencent_code.startswith("sh"):
return tencent_code[2:] + ".SH"
elif tencent_code.startswith("sz"):
return tencent_code[2:] + ".SZ"
elif tencent_code.startswith("hk"):
return tencent_code[2:] + ".HK"
elif tencent_code.startswith("us"):
return tencent_code[2:]
return tencent_code
# ==================== 数据获取 ====================
class StockDataFetcher:
"""股票数据获取器（腾讯财经 + 东方财富备用）"""
def __init__(self):
self._cache = {}
self._cache_time = 0
self._cache_ttl = 10
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
})
def _parse_tencent(self, text: str, code: str) -> dict:
"""解析腾讯财经返回数据"""
try:
match = re.search(r'"(.*?)"', text)
if not match:
return {"error": "返回格式异常"}
parts = match.group(1).split("~")
if len(parts) < 33:
return {"error": "数据字段不足"}
name = parts[1]
price_str = parts[3]
prev_str = parts[4]
open_str = parts[5]
# [32] = 涨跌幅百分比 (%)
change_pct_str = parts[32] if len(parts) > 32 else "0"
# [33] = 最高价, [34] = 最低价 —— 不要误用！
if not price_str or price_str in ("0.00", "0.000", ""):
return {"error": "暂无交易数据（可能休市或未开盘）"}
price = float(price_str)
prev_close = float(prev_str) if prev_str and prev_str not in ("0.00", "0.000", "") else price
# 优先使用接口返回的涨跌幅，避免自己计算有精度问题
try:
change_pct = float(change_pct_str)
except (ValueError, TypeError):
change_pct = 0.0
change = round(price - prev_close, 2)
return {
"price": round(price, 2),
"prev_close": round(prev_close, 2),
"change": change,
"change_pct": round(change_pct, 2),
"name": name,
"error": None
}
except Exception as e:
return {"error": f"解析失败: {e}"}
def _fetch_tencent(self, code: str) -> dict:
url = f"https://qt.gtimg.cn/q={code}"
try:
resp = self.session.get(url, timeout=REQUEST_TIMEOUT)
resp.encoding = "gbk"
return self._parse_tencent(resp.text, code)
except requests.exceptions.Timeout:
return {"error": "请求超时，请检查网络"}
except Exception as e:
return {"error": f"腾讯接口异常: {e}"}
def _get_eastmoney_market(self, code: str) -> tuple:
if code.startswith("sh"):
return "1", code[2:]
elif code.startswith("sz"):
return "0", code[2:]
elif code.startswith("hk"):
return "116", code[2:]
elif code.startswith("us"):
return "105", code[2:]
return "1", code
def _fetch_eastmoney(self, code: str) -> dict:
market, num = self._get_eastmoney_market(code)
url = (
f"https://push2.eastmoney.com/api/qt/stock/get"
f"?secid={market}.{num}"
f"&fields=f43,f44,f45,f57,f58,f60,f169,f170"
)
try:
resp = self.session.get(url, timeout=REQUEST_TIMEOUT)
data = resp.json()
d = data.get("data", {})
if not d:
return {"error": "东方财富无数据"}
divisor = 1 if market == "105" else 100
price = d.get("f43", 0) / divisor
prev = d.get("f60", 0) / divisor
# f45 和 f170 都是涨跌幅(已*100)，除以100得百分比
change_pct_raw = d.get("f45") or d.get("f170") or 0
change_pct = change_pct_raw / 100
# f44 和 f169 是涨跌额
change_raw = d.get("f44") or d.get("f169") or 0
change = change_raw / divisor
name = d.get("f58", "")
if price == 0:
return {"error": "暂无交易数据"}
return {
"price": round(price, 2),
"prev_close": round(prev, 2),
"change": round(change, 2),
"change_pct": round(change_pct, 2),
"name": name,
"error": None
}
except Exception as e:
return {"error": f"东方财富异常: {e}"}
def fetch(self, tencent_code: str) -> dict:
now = time.time()
cache_key = tencent_code.upper()
if cache_key in self._cache and (now - self._cache_time) < self._cache_ttl:
return self._cache[cache_key]
result = self._fetch_tencent(tencent_code)
if result.get("error"):
print(f"[腾讯失败] {result['error']}，尝试东方财富...")
result = self._fetch_eastmoney(tencent_code)
result["symbol"] = get_display_symbol(tencent_code)
result["time"] = datetime.now().strftime("%H:%M:%S")
self._cache[cache_key] = result
self._cache_time = now
return result
# ==================== 图标生成 ====================
class IconGenerator:
def __init__(self, size=ICON_SIZE):
self.size = size
self.font = None
self._init_font()
def _init_font(self):
font_paths = [
"C:/Windows/Fonts/msyhbd.ttc",
"C:/Windows/Fonts/simhei.ttf",
"C:/Windows/Fonts/simsun.ttc",
"C:/Windows/Fonts/calibrib.ttf",
"C:/Windows/Fonts/arialbd.ttf",
]
for fp in font_paths:
if os.path.exists(fp):
try:
self.font = ImageFont.truetype(fp, FONT_SIZE)
return
except:
continue
self.font = ImageFont.load_default()
def create(self, data: dict) -> Image.Image:
price = data.get("price")
change_pct = data.get("change_pct", 0)
error = data.get("error")
if error or price is None:
bg_color = (80, 80, 80)
elif change_pct > 0:
bg_color = (200, 50, 50) # 红 = 涨
elif change_pct < 0:
bg_color = (50, 150, 50) # 绿 = 跌
else:
bg_color = (100, 100, 100) # 灰 = 平
img = Image.new("RGB", (self.size, self.size), bg_color)
draw = ImageDraw.Draw(img)
if error or price is None:
text = "N/A"
text_color = (255, 255, 255)
else:
price_str = f"{price:.1f}" if price >= 1000 else f"{price:.2f}"
if len(price_str) > 5:
price_str = f"{price:.0f}"
text = price_str
text_color = (255, 255, 255)
bbox = draw.textbbox((0, 0), text, font=self.font)
text_w = bbox[2] - bbox[0]
text_h = bbox[3] - bbox[1]
x = (self.size - text_w) // 2
y = (self.size - text_h) // 2 - 2
draw.text((x, y), text, fill=text_color, font=self.font)
return img
# ==================== 托盘应用 ====================
class StockTrayApp:
def __init__(self):
self.tencent_code = self._load_code()
self.fetcher = StockDataFetcher()
self.icon_gen = IconGenerator()
self.icon = None
self._running = True
self._update_thread = None
def _load_code(self) -> str:
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
cfg = json.load(f)
return cfg.get("code", "sh600519")
except:
pass
return "sh600519"
def _save_code(self, code: str):
try:
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump({"code": code}, f, ensure_ascii=False)
except:
pass
def _prompt_code(self):
def input_thread():
print("\n" + "="*45)
print(" 请输入股票代码（直接回车取消）")
print(" A股: 600519, 000001, 300750")
print(" 港股: 0700, 3690")
print(" 美股: AAPL, TSLA, NVDA")
print("="*45)
new_code = input("> ").strip()
if new_code:
tencent = to_tencent_code(new_code)
self.tencent_code = tencent
self._save_code(tencent)
print(f" 已切换: {get_display_symbol(tencent)}")
self._update()
else:
print(" 取消切换")
t = threading.Thread(target=input_thread, daemon=True)
t.start()
def _update(self):
if not self._running:
return
data = self.fetcher.fetch(self.tencent_code)
img = self.icon_gen.create(data)
price = data.get("price")
change_pct = data.get("change_pct", 0)
change = data.get("change", 0)
name = data.get("name", "")
error = data.get("error")
symbol = get_display_symbol(self.tencent_code)
if error or price is None:
title = f"{symbol} {name} - {error}"
else:
arrow = "▲" if change_pct > 0 else "▼" if change_pct < 0 else "━"
title = f"{symbol} {name} {price} {arrow}{change:+.2f} ({change_pct:+.2f}%)"
if self.icon:
self.icon.icon = img
self.icon.title = title
ts = datetime.now().strftime("%H:%M:%S")
if error:
print(f"[{ts}] {symbol} - 错误: {error}")
else:
print(f"[{ts}] {symbol} {name} {price} {change:+.2f} ({change_pct:+.2f}%)")
def _update_loop(self):
while self._running:
try:
self._update()
except Exception as e:
print(f"[{datetime.now().strftime('%H:%M:%S')}] 更新异常: {e}")
time.sleep(UPDATE_INTERVAL)
def _on_change(self, icon, item):
self._prompt_code()
def _on_refresh(self, icon, item):
self._update()
def _on_exit(self, icon, item):
self._running = False
icon.stop()
def _build_menu(self):
sym = get_display_symbol(self.tencent_code)
return Menu(
MenuItem(f"当前: {sym}", lambda i, item: None, enabled=False),
MenuItem("更换股票", self._on_change),
MenuItem("立即刷新", self._on_refresh),
Menu.SEPARATOR,
MenuItem("退出", self._on_exit),
)
def run(self):
sym = get_display_symbol(self.tencent_code)
print("="*50)
print(" Windows 托盘股价监控（国内数据源版）")
print(f" 当前标的: {sym}")
print(f" 更新间隔: {UPDATE_INTERVAL}秒")
print("="*50)
print("提示：右键托盘图标可更换股票或退出\n")
data = self.fetcher.fetch(self.tencent_code)
img = self.icon_gen.create(data)
self.icon = pystray.Icon(
"stock_tray",
icon=img,
title=f"{sym} - 加载中...",
menu=self._build_menu()
)
self._update_thread = threading.Thread(target=self._update_loop, daemon=True)
self._update_thread.start()
self.icon.run()
if __name__ == "__main__":
if not os.path.exists(CONFIG_FILE):
print("首次运行，请设置股票代码：")
print(" A股: 600519, 000001, 300750")
print(" 港股: 0700, 3690")
print(" 美股: AAPL, TSLA, NVDA")
user_input = input("请输入 [默认600519]: ").strip()
code = to_tencent_code(user_input) if user_input else "sh600519"
try:
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump({"code": code}, f, ensure_ascii=False)
except:
pass
app = StockTrayApp()
app.run()
阅读
40
|
点赞
3
评论已关闭