[Python] 纯文本查看 复制代码
import requests
import matplotlib.pyplot as plt
import matplotlib.ticker as mticker
import time
# 请求头
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
"Referer": "https://quote.eastmoney.com/"
}
# 股票市场前缀
def get_market(code):
if code.startswith("6"):
return "1." + code # 上证
elif code.startswith("0") or code.startswith("3"):
return "0." + code # 深证
else:
raise ValueError("股票代码格式错误，应如 600519 / 002044")
# 获取分时数据和昨日收盘价
def get_minute_data(code):
secid = get_market(code)
url = "https://push2his.eastmoney.com/api/qt/stock/trends2/get"
params = {
"secid": secid,
"fields1": "f1,f2,f3,f4,f5",
"fields2": "f51,f52,f53,f54,f55",
"ut": "fa5fd1943c7b386f172d6893dbfba10b",
"iscr": "0",
"ndays": "1"
}
r = requests.get(url, params=params, headers=HEADERS, timeout=10, proxies={"http": None, "https": None})
r.raise_for_status()
j = r.json()
data = j.get("data", {})
trends = data.get("trends", [])
if not trends:
raise ValueError("接口未返回有效分时数据")
prices = []
for item in trends:
parts = item.split(",")
try:
price = float(parts[1])
prices.append(price)
except:
continue
if len(prices) < 2:
raise ValueError("分时数据不足")
yesterday_close = float(data.get("preclose", prices[0]))
return prices, yesterday_close
# 绘制初始图
def draw_chart(prices, yesterday_close, code):
percents = [(p - yesterday_close) / yesterday_close * 100 for p in prices]
plt.ion() # 开启交互模式
fig, ax = plt.subplots(figsize=(10, 5))
line, = ax.plot(percents, linestyle="--", color="black", linewidth=1.6)
ax.axhline(0, color="black") # 0轴对应昨日收盘
# 自定义Y轴刻度带符号
def y_format(x, pos):
if x > 0:
return f"+{x:.2f}"
elif x < 0:
return f"{x:.2f}"
else:
return "0"
ax.yaxis.set_major_formatter(mticker.FuncFormatter(y_format))
plt.title(f"{code} 实时分时走势（涨跌幅 %）")
plt.xlabel("时间（分钟）")
plt.ylabel("涨跌幅 %")
plt.grid(False)
plt.show(block=False)
return fig, ax, line
# 更新曲线
def update_chart(fig, ax, line, prices, yesterday_close):
percents = [(p - yesterday_close) / yesterday_close * 100 for p in prices]
line.set_ydata(percents)
line.set_xdata(range(len(percents)))
ax.relim()
ax.autoscale_view()
fig.canvas.draw()
fig.canvas.flush_events()
# 主程序
def main():
code = input("请输入股票代码（如 600519 / 002044）：").strip()
try:
prices, yesterday_close = get_minute_data(code)
print(f"获取到 {len(prices)} 个分时点，开始绘图…")
fig, ax, line = draw_chart(prices, yesterday_close, code)
while True:
try:
prices, _ = get_minute_data(code)
update_chart(fig, ax, line, prices, yesterday_close)
time.sleep(5) # 每 5 秒刷新一次
except Exception as e:
print("更新失败:", e)
time.sleep(5)
except Exception as e:
print("获取数据失败:", e)
if __name__ == "__main__":
main()
[复制链接]
发表于 2026-1-16 16:07
|
发表于 2026-1-16 16:14
