python不停获取沪深300价格 到警戒值报警提示
import akshare as ak
import time
from datetime import datetime
import winsound
import threading
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
import pandas as pd
import matplotlib.font_manager as fm
import matplotlib.dates as mdates
from matplotlib.offsetbox import AnchoredText
# 沪深300指数代码
HS300_CODE = "000300"
# 预警阈值
ALERT_THRESHOLD = 3000
# 检查间隔（秒）
CHECK_INTERVAL = 10
# Y轴最高刻度值
Y_AXIS_MAX = 20000
# Y轴最低刻度值（可根据需要调整）
Y_AXIS_MIN = 0
def setup_chinese_font():
"""配置Matplotlib中文字体"""
font_candidates = [
"SimHei", "WenQuanYi Micro Hei", "Heiti TC",
"Microsoft YaHei", "Arial Unicode MS"
]
available_fonts = []
try:
font_files = fm.findSystemFonts(fontpaths=None, fontext='ttf')
for font_file in font_files:
try:
font = fm.FontProperties(fname=font_file)
available_fonts.append(font.get_name())
except:
continue
except Exception as e:
print(f"字体检测失败: {str(e)}")
for font in font_candidates:
if font in available_fonts:
plt.rcParams["font.family"] = font
print(f"已设置中文字体: {font}")
return True
plt.rcParams["font.family"] = ["sans-serif"]
plt.rcParams["axes.unicode_minus"] = False
print("未找到指定中文字体，使用默认字体")
return False
class HS300Monitor:
def __init__(self):
self.prices = [] # 存储(时间戳, 价格)元组
self.alert_triggered = False
self.running = True
self.current_price = 0 # 当前价格
self.price_text = None # 价格显示对象
def get_hs300_price(self):
"""获取沪深300实时价格"""
try:
index_df = ak.stock_zh_index_spot_em()
hs300_df = index_df[index_df['代码'] == HS300_CODE]
if not hs300_df.empty:
try:
current_price = float(hs300_df.iloc[0]['最新价'])
change_amount = float(hs300_df.iloc[0]['涨跌额'])
change_percent = float(hs300_df.iloc[0]['涨跌幅'])
except ValueError as e:
print(f"价格数据转换失败: {str(e)}")
return None
price_info = {
'code': hs300_df.iloc[0]['代码'],
'name': hs300_df.iloc[0]['名称'],
'current_price': current_price,
'change_amount': change_amount,
'change_percent': change_percent,
'datetime': datetime.now()
}
self.current_price = current_price
return price_info
else:
print(f"未找到沪深300指数({HS300_CODE})数据")
return None
except Exception as e:
print(f"获取价格失败: {str(e)}")
return None
def voice_alert(self):
"""电脑语音提醒"""
try:
winsound.Beep(440, 1000)
time.sleep(0.5)
winsound.Beep(660, 1000)
import win32com.client
speaker = win32com.client.Dispatch("SAPI.SpVoice")
speaker.Speak(f"警告，沪深300指数已经跌破{ALERT_THRESHOLD}点")
except ImportError:
for _ in range(3):
winsound.Beep(800, 500)
time.sleep(0.3)
except Exception as e:
print(f"语音提醒失败: {str(e)}")
def monitor_price(self, interval=5):
"""价格监控线程"""
self.alert_triggered = False
while self.running:
price_info = self.get_hs300_price()
if price_info:
current_price = price_info['current_price']
current_time = price_info['datetime']
current_time_str = current_time.strftime('%H:%M:%S')
self.prices.append((current_time, current_price))
if len(self.prices) > 100:
self.prices.pop(0)
print(f"[{current_time_str}] 沪深300价格: {current_price:.2f}，涨跌幅: {price_info['change_percent']:.2f}%")
if current_price < ALERT_THRESHOLD and not self.alert_triggered:
print(f"\n 沪深300指数跌破{ALERT_THRESHOLD}点！当前价格: {current_price:.2f}\n")
threading.Thread(target=self.voice_alert, daemon=True).start()
self.alert_triggered = True
if current_price >= ALERT_THRESHOLD and self.alert_triggered:
self.alert_triggered = False
print(f"沪深300指数回升至{ALERT_THRESHOLD}点以上")
time.sleep(interval)
def start_monitor(self, interval=5):
"""启动监控（Y轴最高刻度固定为20000）"""
print(f"开始监控沪深300指数，检查间隔: {interval}秒，预警阈值: {ALERT_THRESHOLD}点")
print(f"界面Y轴最高刻度值设置为: {Y_AXIS_MAX}\n")
monitor_thread = threading.Thread(target=self.monitor_price, args=(interval,), daemon=True)
monitor_thread.start()
setup_chinese_font()
fig, ax = plt.subplots(figsize=(10, 6))
line, = ax.plot([], [], 'b-', lw=2)
# 设置Y轴范围（最高20000）
ax.set_ylim(Y_AXIS_MIN, Y_AXIS_MAX)
# 设置Y轴刻度间隔（每2000点一个刻度）
ax.set_yticks(range(Y_AXIS_MIN, Y_AXIS_MAX + 1, 2000))
# 添加价格显示框
self.price_text = AnchoredText(
f"当前价格: ---\n警戒值: {ALERT_THRESHOLD}",
loc='upper left',
prop={'fontsize': 10},
bbox_to_anchor=(0.02, 0.98),
bbox_transform=ax.transAxes,
borderpad=0.5
)
self.price_text.patch.set_boxstyle("round,pad=0.3,rounding_size=0.2")
ax.add_artist(self.price_text)
ax.set_title(f'沪深300指数实时走势 (Y轴最高: {Y_AXIS_MAX})')
ax.set_xlabel('时间')
ax.set_ylabel('指数价格')
ax.grid(True, which='both') # 显示所有网格线
# 警戒线
ax.axhline(
y=ALERT_THRESHOLD,
color='r',
linestyle='--',
alpha=0.7,
label=f'警戒值: {ALERT_THRESHOLD}'
)
ax.legend()
# 时间格式设置
ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S'))
fig.autofmt_xdate()
def init():
line.set_data([], [])
return line,
def update(frame):
# 更新价格显示
if self.current_price > 0:
color = 'red' if self.current_price < ALERT_THRESHOLD else 'green'
price_text = f"当前价格: {self.current_price:.2f}\n警戒值: {ALERT_THRESHOLD}"
# 更新文本框
try:
ax.patches.remove(self.price_text.patch)
ax.texts.remove(self.price_text.txt)
except:
pass
self.price_text = AnchoredText(
price_text,
loc='upper left',
prop={'fontsize': 10, 'color': color},
bbox_to_anchor=(0.02, 0.98),
bbox_transform=ax.transAxes,
borderpad=0.5
)
self.price_text.patch.set_boxstyle("round,pad=0.3,rounding_size=0.2")
ax.add_artist(self.price_text)
# 更新曲线
if self.prices:
times = [t for t, _ in self.prices]
prices = [p for _, p in self.prices]
line.set_data(times, prices)
ax.relim()
ax.autoscale_view(scaley=False) # 仅自动调整X轴，Y轴保持固定
ax.set_title(f'沪深300指数实时走势 (Y轴最高: {Y_AXIS_MAX}) - 最后更新: {datetime.now().strftime("%H:%M:%S")}')
return line, self.price_text
ani = FuncAnimation(
fig,
update,
init_func=init,
interval=1000,
blit=True,
cache_frame_data=False
)
try:
plt.tight_layout()
plt.show()
except KeyboardInterrupt:
print("\n用户终止监控")
finally:
self.running = False
monitor_thread.join(timeout=1)
print("监控已停止")
if __name__ == "__main__":
monitor = HS300Monitor()
try:
monitor.start_monitor(interval=5)
except Exception as e:
monitor.running = False
print(f"程序异常终止: {str(e)}")