股票程序化交易接口实现:如何用python监控股票

例如本周五的奥飞娱乐：

昨日未涨停

#筛选备选股
#条件：1.昨日未涨停2.近10日内有涨停
rec_zt_df=df_raw.loc[df_raw.close==df_raw.high_limit]#近10日涨停股
rec_zt_list=list(set(rec_zt_df["code"]))#近10日涨停股列表
pre_zt_list=rec_zt_df[rec_zt_df.time==pre_date]["code"].tolist()#昨日涨停股列表
cand_list=list(set(rec_zt_list).difference(set(pre_zt_list)))#备选股列表
cand_preclose_dict={}#存放被选股昨日收盘价
for code in cand_list:
    pre_close=df_raw[(df_raw.time==pre_date)&(df_raw.code==code)]["close"].values[0]#昨日收盘价
    cand_preclose_dict[code]=pre_close
print("备选股总计{}只".format(len(cand_list)))

我们也可以将异动信息通过API同步推送到企业微信上，实现盘中自动提醒买卖点：

为了方便演示，这里我们只对涨幅和成交量进行监控，实际上，所有其他的盘口信息都可以进行类似的监控。上面的监控股奥飞娱乐、南宁糖业、我爱我家、剑桥科技当天全部涨停！！！

异动监控

我们选用分钟级别的监控频率，可以自由控制需要监控的时长：

#用于盘中实时监控，频率为每分钟，9点30分之后运行
for i in range(120):
    curr_time=datetime.datetime.today().strftime("%Y-%m-%d %H:%M:%S")
if curr_time[-8:-3]>="15:00":
        print("==结束今日监控程序==")
break
else:
        print("**********"+curr_time+"**********")
        curr_df=get_price(cand_list,count=1,end_date=datetime.datetime.today(),frequency="1m", fields=["close","high_limit","volume"],skip_paused=True, fq="pre",panel=False)
for code,pre_close in cand_preclose_dict.items():
            curr_p=curr_df.loc[curr_df.code==code].close.values[0]#当前价格
            curr_v=curr_df.loc[curr_df.code==code].close.values[0]#当前成交量
            curr_r=round((curr_p-pre_close)/pre_close*100,2)#当前涨幅
if (curr_r>5)&(curr_r<7):
                print("warning-5:",code,security_code_name_dict[code],"  涨幅:",curr_r,"成交量:",curr_v)
if (curr_r>=7)&(curr_r<9):
                print("warning-7:",code,security_code_name_dict[code],"  涨幅:",curr_r,"成交量:",curr_v)
if (curr_r>=9):
                print("warning-9:",code,security_code_name_dict[code],"  涨幅:",curr_r,"成交量:",curr_v)
        time.sleep(60)

最后筛选备选股池，我们可以人为规定双响炮两个炮台的时间间隔，比如10天以内。注意，抓双响炮是在走出双响炮第二根炮台的当天买入，而不是已经走出双响炮后一天再买入，那样就没有意义了。

涨停双响炮是典型的短线牛股形态之在市场中非常常见，走出双响炮的股往往后市有不错的行情。

近10日内有涨停

备选股筛选好后，我们就可以在盘中进行实时监控了。

那么双响炮备选股条件定义为：

如何用python监控股票

本文源码：量化研究分享：如何用python抓涨停双响炮

了解了什么是涨停双响炮后，接下来进入今天的正题，如何用python实现双响炮的实时监控。

首先过滤原始股池，剔除科创板、新股、ST股等，得到过滤后的股票列表：

#过滤股池
def get_normal_stocks(date,new_days=60):
"""
    获取全部股票代码，并过滤掉st、退市股、新股、科创板、创业板等
    :param date:目标日期
    :param new_days:60天前上市
    """
    trd_days = get_trade_days(end_date=date, count=new_days)
    stock_data = get_all_securities("stock", trd_days[0])
    stock_data=stock_data[~stock_data.display_name.str.contains("ST")]
    stock_data=stock_data[~stock_data.display_name.str.contains("//*")]
    stock_data=stock_data[~stock_data.display_name.str.contains("退")]
    stock_list=stock_data.index.tolist()
    stock_list=[stock for stock in stock_list if not  stock.startswith("688")]
    stock_list=[stock for stock in stock_list if not  stock.startswith("30")]
return stock_list

stock_list=get_normal_stocks(pre_date)

然后我们获取这些股票近期的行情数据，这里只用拿到收盘价和涨停价就可以了：

#获取截至昨天收盘近10日数据
df_raw = get_price(stock_list, start_date=None, end_date=pre_date,count=period,frequency="daily", fields=["close","high_limit"],skip_paused=True, fq="pre",panel=False)
print(len(df_raw))

环境准备

监控效果如下：

以聚宽为例，我们先导入相关库，并设置好需要监控的时间。

from jqdata import *#聚宽研究环境
#import jqdatasdk#本地环境，需要账号认证
#from jqdatasdk import *
import datetime
import time
import pandas as pd
import numpy as np

#设置研究时间
period=10#研究最近10天的数据
today0=datetime.datetime.today()#自动获取当天日期+时间
today1=today0.strftime("%Y-%m-%d")#转化格式：日期
trade_days=get_trade_days(end_date=today1,count=period)#获取最近period的交易日期
trade_days=list(map(lambda x:x.strftime("%Y-%m-%d"),trade_days))
today=trade_days[-1]#今天
pre_date=trade_days[-2]#上一个交易日
print("今天是{}，上一个交易日是{}".format(today,pre_date))

获取数据

南宁糖业：

文章为作者独立观点，不代表股票交易接口观点