基于 Python 监控股票涨停情况_efinance股票 python

作者：Gausst松鼠会 | 2024-02-12 20:55:02

踩

efinance股票 python

前言

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

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

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

开始之前

Python 版本要求

python 3.7 及其以上

如果没有安装 Python，可以参考我写的这篇安装教程

需要安装的库

efinance>=0.4.4

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

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

pip install efinance --upgrade
1

正文

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

突然发现，知乎的代码块不支持放置表情符号，本来代码里面有小火箭的，但粘贴上去后被吞了，所以如果需要原样代码的话，可以看我给的 gist

import requests
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
requests.packages.urllib3.util.ssl_.DEFAULT_CIPHERS = 'ALL:@SECLEVEL=1'

@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

    @property
    def zt_keep_seconds(self) -> int:
        """
        涨停保持秒数

        Returns
        -------
        int

        """
        return (self.latest_zt_dt - self.latest_nzt_dt).seconds


class Clock:
    def __init__(self) -> None:
        self.dt = datetime.now()

    def next(self) -> bool:
        """
        是否在 09:15:00 - 15:00:00

        Returns
        -------
        bool
        """
        dt = datetime.now()
        st = '09:15:00'
        et = '15:00:00'
        self.dt = dt
        return st <= dt.strftime('%H:%M:%S') <= et


def 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.task
    def 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: Clock

    def __post_init__(self) -> None:
        self.stock_code_info: Dict[str, StockQuoteInfo] = {}

    def next(self) -> None:
        dt = self.clock.dt

        quotes = ef.stock.get_realtime_quotes()
        quotes.index = quotes['股票代码'].values
        quotes = quotes[quotes['涨跌幅'] != '-']
        # * 初步选出即将涨停的股票
        quotes = quotes[quotes['涨跌幅'] > 7]
        if len(quotes) == 0:
            return
        sns = 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 None
            if 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_info
            buy_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_TIP
                    pre_info.latest_zt_dt = dt
                # * 保持涨停则更新最新涨停时间
                elif current_price == pre_info.price:
                    tip = ZT_KEEP_TIP
                    pre_info.latest_zt_dt = dt
                # * 炸板后更新最新的不涨停时间
                else:
                    tip = ZT_BREAK_TIP
                    pre_info.latest_nzt_dt = dt

            # * 非涨停 更新价格
            else:
                pre_info.latest_nzt_dt = dt
            # * 不管有没有涨停均更新
            pre_info.price = current_price
            pre_info.dt = dt

            # * 在这里根据涨停状况做通知
            # * 如果需要推送到微信，可查看我写的 wechat_work 这个库
            # * 地址为 https://github.com/Micro-sheep/wechat_work
            if 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 = True

ZT_TIP = '刚涨停'
ZT_KEEP_TIP = '保持涨停'
ZT_BREAK_TIP = '涨停炸板'
# * 保持涨停通知超时时间 涨停保持秒数超过它则不做通知
ZT_NOTICE_MAX_SECONDS = 60

clock = Clock()
strategy = Strategy(clock)
while clock.next() or TEST_MODE:
    dt = clock.dt
    rich.print(f'[{dt.strftime("%m-%d %H:%M:%S")}] 刷新')
    strategy.next()
print('今日监控结束')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184

在终端（对于 Windows 不建议使用 cmd，建议使用 Windows Terminal 或者 PowerShell）运行上面的代码之后，得到的输出形式如下
在这里插入图片描述
我给的程序中，通知方式是仅仅把消息输出到终端上，这样子显然是不合理的，毕竟不可能一直去盯着屏幕。而实际上，我选择的是推送到微信上，毕竟手机总是在身旁的，且微信通知总处于打开状态，收到消息了，也可以手动在手机交易软件上挂单。

写在最后

在学习python中有任何困难不懂的可以微信扫描下方CSDN官方认证二维码加入python交流学习
多多交流问题，互帮互助，这里有不错的学习教程和开发工具。
（这里每天都会不定时更新python不同题型和教程，希望大家一起学习，一起进步）

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Gausst松鼠会/article/detail/78574