基于 Python 监控股票涨停情况

前言

偶然一次打开炒股软件，发现了这么一个下面这么一个区域

我的第一反应是：天呐，竟然还能这么玩？简单粗暴！所以我立马想到写一个涨停板监控程序，一旦监控到股票涨停并且有大量买单则立马通知我，当然还可以自动执行挂涨停买入的操作，接着在第二天挂跌停卖出（挂跌停可以保证优先卖出，但不一定以跌停价成交，这样子可以避免遇到多个核按钮，并且总体盈利更加平稳）。

我的程序在实操过程中，打板进入的成功率一般有 80% 及其以上，说明在市场上，手动打板的人还是比较多，否则即使让程序自动挂单也很难买进去。

考虑到很久没有更新文章了，今天就以监控涨停股为主题，分享我基于 Python 写的监控涨停股，以及时打板的程序。程序化挂单代码则省略，有能力的自行接入自动下单接口。

开始之前

Python 版本要求

python 3.7 及其以上

需要安装的库

efinance>=0.4.2

注意：一定要保证 efinance版本至少为0.4.2

库的安装方法是：打开 cmd（命令提示符或者其他终端工具），输入以下代码

pip install efinance --upgrade

正文

多余的话不说了，直接上代码，代码中有详细注释

import multitasking
import pandas as pd
from typing import List
from typing import Dict
import efinance as ef
from dataclasses import dataclass
from datetime import datetime
import rich@dataclass()
class StockQuoteInfo:# * 股票代码stock_code: str# * 股票名称stock_name: str# * 行情时间dt: datetime# * 最新价price: float# * 涨停价top_price: float# * 跌停价bottom_price: float# * 最新涨停时间latest_zt_dt: datetime# * 最新非涨停时间latest_nzt_dt: datetime@propertydef zt_keep_seconds(self) -> int:"""涨停保持秒数Returns-------int"""return (self.latest_zt_dt - self.latest_nzt_dt).secondsclass Clock:def __init__(self) -> None:self.dt = datetime.now()def next(self) -> bool:"""是否在 09:15:00 - 15:00:00Returns-------bool"""dt = datetime.now()st = '09:15:00'et = '15:00:00'self.dt = dtreturn st <= dt.strftime('%H:%M:%S') <= etdef get_snapshot_fast(stock_codes: List[str]) -> Dict[str, pd.DataFrame]:"""获取多只股票的最新行情快照Parameters----------stock_codes : List[str]股票代码列表Returns-------Dict[str, DataFrame]股票代码为键，行情快照为值的字典"""sns: Dict[str, pd.DataFrame] = {}@multitasking.taskdef start(stock_code: str) -> None:sns[stock_code] = ef.stock.get_quote_snapshot(stock_code)for stock_code in stock_codes:start(stock_code)multitasking.wait_for_tasks()return sns@dataclass()
class Strategy:clock: Clockdef __post_init__(self) -> None:self.stock_code_info: Dict[str, StockQuoteInfo] = {}def next(self) -> None:dt = self.clock.dtquotes = ef.stock.get_realtime_quotes()quotes.index = quotes['股票代码'].valuesquotes = quotes[quotes['涨跌幅'] != '-']# * 初步选出即将涨停的股票quotes = quotes[quotes['涨跌幅'] > 7]if len(quotes) == 0:returnsns = get_snapshot_fast(quotes.index.values)for row in quotes.iloc:stock_code = row['股票代码']stock_name = row['股票名称']# * 最新行情快照sn = sns[stock_code]# * 涨停价top_price = sn['涨停价']# * 跌停价bottom_price = sn['跌停价']# * 最新价格current_price = sn['最新价']# * 上一次刷新时的行情pre_info = self.stock_code_info.get(stock_code)# * 该股是不是第一次被检测first = pre_info is Noneif first:pre_info = StockQuoteInfo(stock_code=stock_code,stock_name=stock_name,dt=dt,price=current_price,top_price=top_price,bottom_price=bottom_price,latest_nzt_dt=dt,latest_zt_dt=None)self.stock_code_info[stock_code] = pre_infobuy_list = []for i in range(1, 6):buy_list.append(f'买 {i}: {sn[f"买{i}数量"]}')# * 买单情况buy_str = '\n'.join(buy_list)tip: str = None# * 检测是否刚涨停或者打开涨停if abs(top_price-current_price) <= 1e-2:# * 刚涨停则更新最新涨停时间if first or current_price > pre_info.price:tip = ZT_TIPpre_info.latest_zt_dt = dt# * 保持涨停则更新最新涨停时间elif current_price == pre_info.price:tip = ZT_KEEP_TIPpre_info.latest_zt_dt = dt# * 炸板后更新最新的不涨停时间else:tip = ZT_BREAK_TIPpre_info.latest_nzt_dt = dt# * 非涨停 更新价格else:pre_info.latest_nzt_dt = dt# * 不管有没有涨停均更新pre_info.price = current_pricepre_info.dt = dt# * 在这里根据涨停状况做通知# * 如果需要推送到微信，可查看我写的 wechat_work 这个库# * 地址为 https://github.com/Micro-sheep/wechat_workif tip == ZT_TIP or (tip == ZT_KEEP_TIP and pre_info.zt_keep_seconds <= ZT_NOTICE_MAX_SECONDS):msg = f'股票代码: {stock_code}\n股票名称: {stock_name}\n  封单情况  \n{buy_str}\n  {tip}  \n  涨停保持秒数: {pre_info.zt_keep_seconds}  'rich.print(msg)# * 是否为测试模式 如果是 True 则不管是否在 09:15:00 - 15:00:00 都会执行
# * 如果是 False 则只有在 09:15:00 - 15:00:00 才会执行
TEST_MODE = TrueZT_TIP = '刚涨停'
ZT_KEEP_TIP = '保持涨停'
ZT_BREAK_TIP = '涨停炸板'
# * 保持涨停通知超时时间 涨停保持秒数超过它则不做通知
ZT_NOTICE_MAX_SECONDS = 60clock = Clock()
strategy = Strategy(clock)
while clock.next() or TEST_MODE:dt = clock.dtrich.print(f'[{dt.strftime("%m-%d %H:%M:%S")}] 刷新')strategy.next()
print('今日监控结束')

在终端（对于 Windows 不建议使用 cmd，建议使用 Windows Terminal 或者 PowerShell）运行上面的代码之后，得到的输出形式如下

我给的程序中，通知方式是仅仅把消息输出到终端上，这样子显然是不合理的，毕竟不可能一直去盯着屏幕。而实际上，我选择的是推送到微信上，毕竟手机总是在身旁的，且微信通知总处于打开状态，收到消息了，也可以手动在手机交易软件上挂单。

为此，我特地基于企业微信 API，封装了一个发通知的库，链接如下

https://github.com/Micro-sheep/wechat_work

里面有详细使用说明，大家自行考虑是否接入吧！

写在最后

欢迎大家踊跃尝试，在学编程,学Python的小伙伴们，一个人摸黑学会很难,博主也是过来人, 这里新建了一个扣群：1020465983，给大家准备了学习资源、好玩的项目，欢迎大家加入一起交流。