打造个人股票监控系统 实时跟踪股票价格走势
2024-04-21 02:12:19
每日股票监控器
对于非全职股民来说，很难日日盯盘，但又不想错过股市的赚钱机会，希望有个帮忙盯盘的小助手，因此开发个股票小助理
主要功能:
- 下载A股每日k线数据
- 计算每只股票的M5上穿M10时间
- 如果股票当前是M5上穿M10 并且距离M5下穿M10大于60天 则告诉我
- 如果股票当前是M5上穿M10 并且在我设置的股票列表中，则告诉我
- 如果被监控的股票价格在我设置的区间 则告诉我
股票源代码类似如下所示
'''
每日20点 运行更新当天数据 从baostock
功能列表:
如果目录不存在 下载 最近10年数据
如果是新上市股票 则下载 最近 一年 数据计算每只股票的 最后一个M5下穿M10的日期 计算每只股票的 M5上穿M10 如果在监控列表则报警
如果被监控股票的价格在设置区间，则报警 '''import baostock as bs
import pandas as pd
from datetime import datetime,timedelta
from datetime import date
import os
import numpy as np
from CorpApi import CorpApi,CORP_API_TYPE
import random
import requests
# 保存每个股票的数据
data_path = 'test-all'
# M5下穿M10的日期列表
file_path = r'm5-m10-last.csv'
today_file_path = r'today_m5-m10-last.csv'
nday = 60 #m5下穿m10 和m5上穿m10 间隔60天
monitor_code_list = ['601636']# 发送通知消息
def send_msg(message):return responsedef get_stocker_num(code):ret_data = {}try:url = 'https://hq.sinajs.cn/list=%s' % code ret = requests.get(url)if ret.status_code == 200:data = str(ret.content,encoding='gb2312')for item in data.split('\n'):if len(item) > 10:k = item.split('=')[0].replace('var hq_str_','')shuzu = item.split(',')if k.startswith('s_'):ret_data[k] = str(shuzu[1])else:ret_data[k] = float(shuzu[3])return ret_dataexcept Exception as e:print(str(e))print_except(sys._getframe().f_code.co_name, sys.exc_info(), e)return None
'''
判断股票价格是否在设置的区间'''
def start_monitor():# 判断今天是否是周末 dt = date.today() nw = dt.weekday()if nw == 5: # 周六 send_mssage_alert('monitor stock is running')returnelif nw == 6:return # 周一到周五 # 先判断指定股票是否达到特定值 try:list_keys = 's_sz399001,s_sh000001'data = {}data['sh601636'] = '14:30' # 旗滨集团data['sz002714'] = '45:100' # 牧原股份gplist = {}for k in data.keys():list_keys = list_keys + ',' + kgplist = get_stocker_num(list_keys)for k in data.keys():v = data[k]now_v = gplist[k]if now_v == 0:continuemin = float(v.split(':')[0])max = float(v.split(':')[1])if now_v < float(min) or now_v > max:message = '上证指数:' + gplist['s_sh000001'] + '\n' + '深证成指:' + gplist['s_sz399001']message = message + '\n\n股票:' + k + '\n当前价格:' + str(now_v) + '\n监控值:' + vmessage = message + '\nhttps://quote.eastmoney.com/%s.html' % str(k)send_mssage_alert(message)except Exception as e:print_except(sys._getframe().f_code.co_name, sys.exc_info(), e)
'''
下载 历史数据
'''
def download_data(date,start_data,end_date):bs.login()# 获取指定日期的指数、股票数据stock_rs = bs.query_all_stock(date)stock_df = stock_rs.get_data()data_df = pd.DataFrame()for code in stock_df["code"]:#print("Downloading :" + code)if code.startswith('bj'):continue k_rs = bs.query_history_k_data_plus(code, "date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,peTTM,pbMRQ,psTTM,pcfNcfTTM,isST",start_date=start_data, end_date=end_date,frequency="d", adjustflag="3")data_list = []while (k_rs.error_code == '0') & k_rs.next():# 获取一条记录，将记录合并在一起data_list.append(k_rs.get_row_data())result = pd.DataFrame(data_list, columns=k_rs.fields)result.to_csv("%s\\%s.csv" % (data_path,code), encoding="gbk", index=False)bs.logout()def download_one_stock_data(code,start,end):k_rs = bs.query_history_k_data_plus(code, "date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,peTTM,pbMRQ,psTTM,pcfNcfTTM,isST",start_date=start, end_date=end,frequency="d", adjustflag="3")data_list = []while (k_rs.error_code == '0') & k_rs.next():# 获取一条记录，将记录合并在一起data_list.append(k_rs.get_row_data())result = pd.DataFrame(data_list, columns=k_rs.fields)return result
# 更新数据 当前时间 到上次下载的日期 之间的数据
def update_data():# 今天日期today = datetime.now().strftime("%Y-%m-%d")#today = datetime.now().strftime("2021-12-24")print(today)# 判断是否已经存在文件bs.login()# 获取指定日期的指数、股票数据stock_rs = bs.query_all_stock(today)stock_df = stock_rs.get_data()# 没有数据则直接退出if stock_df.empty:return #print(type(stock_df))#print(stock_df)data_df = pd.DataFrame()for code in stock_df["code"]:try: save_path = "%s\\%s.csv" % (data_path,code)if not os.path.isfile(save_path):#print("it is new :" + code)start_day = (datetime.now() - timedelta(days=250)).strftime("%Y-%m-%d")new_data = download_one_stock_data(code,start_day,today)if not new_data.empty:new_data.to_csv(save_path, encoding="gbk", index=False) # 保存文件 else:old_data = pd.read_csv(save_path)# 读取文件最后一行 时间 last_date = old_data["date"].tail(1).values[0] begin_date = datetime.strptime(last_date, '%Y-%m-%d')end_date = datetime.strptime(today, '%Y-%m-%d')# 如果小于当前时间 则下载 if end_date > begin_date:begin_date += timedelta(days=1)new_data = download_one_stock_data(code,begin_date.strftime("%Y-%m-%d"),today)if not new_data.empty:# 读取原数据 old = pd.read_csv(save_path)if old.empty:return# 原数据 加上新数据new_data = pd.concat([old,new_data])new_data.to_csv(save_path, encoding="gbk", index=False) # 保存文件 except Exception as e:pass print(str(e)) bs.logout()'''
生成md5 下穿m10数据表
'''
def create_md5_md10():count = 0 result = []for filename in os.listdir(data_path):try:if filename.find('.') > -1:code = filename.split('.')[1]#过滤掉创业板 和科创板 # if code.startswith('300') or code.startswith('688'):# continuesave_path = os.path.join(data_path,filename)stock = pd.read_csv(save_path, encoding="gbk") #, index_col = 'date'if stock.empty:continue# 是否是ST# if stock.tail(1)['isST'].values[0] == 1:# continue# 市盈率判断 peTTM = stock.tail(1)['peTTM'].values[0]
# if not (peTTM >= 5 and peTTM <= 20):# continue # 计算M5 M10 M20 M30 M60 #stock_data = stockstock['m5']=np.round(stock['close'].rolling(window=5,center=False).mean(),2)stock['m10']=np.round(stock['close'].rolling(window=10,center=False).mean(),2)stock['m20']=np.round(stock['close'].rolling(window=20,center=False).mean(),2)# M5 < M10 # if stock.tail(1)['m5'].values[0] < stock.tail(1)['m10'].values[0]:# continue# M5 向上交叉 M20stock['m5-20']=stock['m5']-stock['m20']stock['diff']=np.sign(stock['m5-20'])stock['signal'] = np.sign(stock['diff'] - stock['diff'].shift(1)) # if stock.tail(1)['signal'].values[0] != 1:# continue #M5 和M10交叉点 stock['m5-10']=stock['m5']-stock['m10']stock['diff_5_10']=np.sign(stock['m5-10'])stock['signal_5_10'] = np.sign(stock['diff_5_10'] - stock['diff_5_10'].shift(1)) # 与前一天相比 如果是正数 则向上交叉 trade = pd.concat([ pd.DataFrame({"code": stock.loc[stock["signal"] == 1, "code"],"date": stock.loc[stock["signal"] == 1, "date"],"price": stock.loc[stock["signal"] == 1, "close"],"operation": "Buy"}),pd.DataFrame({"code": stock.loc[stock["signal_5_10"] == -1, "code"],"date": stock.loc[stock["signal_5_10"] == -1, "date"],"price": stock.loc[stock["signal_5_10"] == -1, "close"],"operation": "Sell"}) ])#trade.sort_index(inplace=True)trade = trade.sort_values(by = 'date') #print(trade)# 取最后10行 反转skip_index = trade.tail(1).indexlast_date = ''trade = trade.tail(100).iloc[::-1]for index, row in trade.iterrows():if row['operation'] == 'Sell':last_date = row['date']breakif last_date == '':continue else: today = datetime.now().strftime("%Y-%m-%d")end_date = datetime.strptime(today, '%Y-%m-%d')begin_date = datetime.strptime(last_date, '%Y-%m-%d')item = {}item['code'] = (stock.tail(1)['code'].values[0]).upper()item['m5m10'] = last_dateresult.append(item)count = count + 1 except Exception as e:print(str(e))outfile = pd.DataFrame(result) # 今日数据 #outfile = outfile.sort_values(by = ['day','pe'],ascending=[0,0]) # 按板块聚合 按涨幅降序排列 #print(outfile)outfile.to_csv(file_path,encoding="gbk",index=False)'''
删除股票代码前字母
'''
def delete_zimu_code(p):try:code = p['code'] if code.find('.') > -1:return code.split('.')[1]except Exception as e:return -1 return ''
def buy_celue5_monitor(df_clear):# 挑选监控的列表 if df_clear.empty:return Falsemessage = ''for index, row in df_clear.iterrows():code = row['code']if True: # 判断行业是否已经购买过 c_data = row.to_dict()message = message + '\n'for key in c_data:if key == 'list_time':message = message + '\n上市时间:' + str(c_data[key]) + '|' + str(c_data['ssn']) # 上市几年if key == 'm5m10':message = message + '\nM5下穿M10:' + str(c_data[key]) + '|' + str(c_data['day']) # 下穿时间if key == 'plate_name':message = message + '\n行业:' + c_data[key]if key == 'code':message = message + '\n股票代码:' + c_data[key] if key == 'stock_name':message = message + '\n股票:' + c_data[key] if key == 'price':message = message + '\n价格:' + str(c_data[key]) if key == 'volume_ratio':message = message + '\n量比:' + str(c_data[key] )if key == 'cur_price_to_highest52_weeks_ratio': #print('it is here')message = message + '\n52周: %s %%' % str(100 + c_data[key] )if key == 'change_rate':message = message + '\n涨跌:' + str(c_data[key] )if key == 'return_on_equity_rate':message = message + '\nROE:' + str(c_data[key] ) if key == 'pe':message = message + '\nPE:' + str(c_data[key] ) if key == 'market_val':message = message + '\n市值:' + str(int(c_data[key] /100000000) ) + '亿' if len(message) > 0:#print(message)send_msg(message) return Truedef dingshi_send():try:filename = 'sz.002439.csv'save_path = os.path.join(data_path,filename)stock = pd.read_csv(save_path, encoding="gbk") #, index_col = 'date'if stock.empty:returnlast_date = stock.tail(1)['date'].values[0] message = '股票监控器运行正常:\n最后一条数据时间:' + str(last_date)#print(message)send_msg(message)except Exception as e:pass
'''
生成md5 上穿m10数据表
'''
def today_md5_md10():count = 0 result = []for filename in os.listdir(data_path):try:if filename.find('.') > -1:code = filename.split('.')[1]# 过滤掉创业板 和科创板 if code.startswith('300') or code.startswith('688'):continuesave_path = os.path.join(data_path,filename)stock = pd.read_csv(save_path, encoding="gbk") #, index_col = 'date'if stock.empty:continue# 是否是STif stock.tail(1)['isST'].values[0] == 1:continue# 市盈率判断 peTTM = stock.tail(1)['peTTM'].values[0]# 跳过指数数据 if peTTM == 0:continue
# if not (peTTM >= 5 and peTTM <= 20):# continue # 计算M5 M10 M20 M30 M60 #stock_data = stockstock['m5']=np.round(stock['close'].rolling(window=5,center=False).mean(),2)stock['m10']=np.round(stock['close'].rolling(window=10,center=False).mean(),2)stock['m20']=np.round(stock['close'].rolling(window=20,center=False).mean(),2)# M5 < M10 if stock.tail(1)['m5'].values[0] < stock.tail(1)['m10'].values[0]:continue# M5 向上交叉 M20
# stock['m5-20']=stock['m5']-stock['m20']
# stock['diff']=np.sign(stock['m5-20'])
# stock['signal'] = np.sign(stock['diff'] - stock['diff'].shift(1)) #M5 和M10交叉点 stock['m5-10']=stock['m5']-stock['m10']stock['diff_5_10']=np.sign(stock['m5-10'])stock['signal_5_10'] = np.sign(stock['diff_5_10'] - stock['diff_5_10'].shift(1)) # 与前一天相比 如果是正数 则向上交叉 if stock.tail(1)['signal_5_10'].values[0] != 1: # M5 向上交叉 M10continue trade = pd.concat([ pd.DataFrame({"code": stock.loc[stock["signal_5_10"] == 1, "code"],"date": stock.loc[stock["signal_5_10"] == 1, "date"],"price": stock.loc[stock["signal_5_10"] == 1, "close"],"operation": "Buy"}),pd.DataFrame({"code": stock.loc[stock["signal_5_10"] == -1, "code"],"date": stock.loc[stock["signal_5_10"] == -1, "date"],"price": stock.loc[stock["signal_5_10"] == -1, "close"],"operation": "Sell"}) ])#trade = pd.concat([ pd.DataFrame({"code": stock.loc[stock["signal"] == 1, "code"],"date": stock.loc[stock["signal"] == 1, "date"],"price": stock.loc[stock["signal"] == 1, "close"],"operation": "Buy"}),# pd.DataFrame({"code": stock.loc[stock["signal_5_10"] == -1, "code"],"date": stock.loc[stock["signal_5_10"] == -1, "date"],"price": stock.loc[stock["signal_5_10"] == -1, "close"],"operation": "Sell"}) ])#trade.sort_index(inplace=True)trade = trade.sort_values(by = 'date') #print(trade)# 取最后10行 反转skip_index = trade.tail(1).indexlast_date = ''trade = trade.tail(100).iloc[::-1]for index, row in trade.iterrows():if row['operation'] == 'Sell':last_date = row['date']breakif last_date == '':continue else: today = datetime.now().strftime("%Y-%m-%d")end_date = datetime.strptime(today, '%Y-%m-%d')begin_date = datetime.strptime(last_date, '%Y-%m-%d')item = {}item['day'] = (end_date - begin_date).daysitem['pe'] = peTTMitem['price'] = stock.tail(1)['close'].values[0]item['code'] = (stock.tail(1)['code'].values[0]).upper()item['m5m10'] = last_date#print(item)result.append(item)# if (end_date - begin_date).days > 5:# print(last_date)# print((end_date - begin_date).days)# print(trade) count = count + 1 except Exception as e:print(str(e))outfile = pd.DataFrame(result) # 今日数据 outfile['no_zimu_code'] = outfile.apply(lambda x: delete_zimu_code(x), axis = 1) # 对每一行上市时间 统计年 df_clear = outfile[outfile['no_zimu_code'].isin(monitor_code_list)] # 监控设置的股票列表 buy_celue5_monitor(df_clear) #df_clear = outfile[outfile['no_zimu_code'].isin(monitor_code_list)]outfile = outfile.sort_values(by = ['day','pe'],ascending=[0,0]) # 按板块聚合 按涨幅降序排列 buy_celue5_monitor(outfile[outfile['day'] > nday]) # 两次间隔 大于 N天的 #print(outfile[(outfile['day'] > 10)])outfile.to_csv(today_file_path,encoding="gbk",index=False)def main():print('start')
# # 判断今天是否是周末
# dt = date.today()
# nw = dt.weekday()
# if nw == 6: # 周日
# #send_msg('monitor stock is running')
# dingshi_send()
# return
# elif nw == 5: #周六
# return
# #判断路径是否存在
# if not os.path.exists(data_path):
# os.mkdir(data_path)
# today = datetime.now().strftime("%Y-%m-%d")
# start_day = (datetime.now() - timedelta(days=365*10)).strftime("%Y-%m-%d")# download_data(today,start_day,today)
# else:
# update_data() # 每日更新数据 # create_md5_md10() # 生成md5数据表today_md5_md10() # 搜索m5上穿m10 start_monitor() #监控股票价格 是否在区间print('end')
main()阅读原文
打造个人股票监控系统 实时跟踪股票价格走势相关推荐
- python拿什么做可视化界面好-用python打造可视化爬虫监控系统，酷炫的图形化界面...
原标题:用python打造可视化爬虫监控系统,酷炫的图形化界面 本文并不是讲解爬虫的相关技术实现的,而是从实用性的角度,将抓取并存入 MongoDB 的数据 用 InfluxDB 进行处理,而后又通过 ...
- 使用USB无线网卡和USB摄像头打造mini2440无线监控系统
一. 我的mini2440开发板上使用的网卡设备为水星MERCURY54M无线USB网卡 MW54U ver:7.0,其内部芯片型号为ATHEROS的ar9271.mini2440的自带linux系 ...
- 7 天打造前端性能监控系统
2019独角兽企业重金招聘Python工程师标准>>> Day1.为什么要监控性能? "If you cannot measure it, you cannot impro ...
- 移动4G打造排污视频监控系统助力咸宁环保建设
针对排污企业分布零散.监管部门无法实时掌握和发现企业污染情况和环境安全隐患等问题,咸宁市环保局联合咸宁移动,在重点排污企业安装视频监控系统,利用高速.稳定的移动4G网络,实现排污情况24小时实时监控. ...
- 获取股票数据【实时更新股票数据、创建你的股票数据】、计算交易指标【买入、卖出信号、计算持仓收益、计算累计收益率】
在上一次获取股票数据[使用JQData查询行情数据.财务指标.估值指标]学习了使用JQData来查询股票相关数据, 这次则开始一点点构建咱们的量化交易系统了. 量化交易平台功能模块了解: 对于一个量化 ...
- 光伏电站监控系统实时监控保障安全运维
随着中国大型太阳能发电设施的建设和运营,如何及时掌握电厂的工作状态,以及如何满足上级或电网调度系统的监测要求,成为电厂业主和电网企业关注的重要课题.而目前的光伏电站监控系统,则是以工厂内的逆变器为基础 ...
- python中turtle画酷炫图案-用python打造可视化爬虫监控系统，酷炫的图形化界面
本文并不是讲解爬虫的相关技术实现的,而是从实用性的角度,将抓取并存入 MongoDB 的数据 用 InfluxDB 进行处理,而后又通过 Grafana 将爬虫抓取数据情况通过酷炫的图形化界面展示出来 ...
- Grafana+Prometheus打造运维监控系统（一）-安装篇
1. Prometheus.Grafana介绍 Prometheus是一个开源的系统监控和报警系统,Grafana 是一个开源的监控数据分析和可视化套件,利用Grafana+Prometheus组合, ...
- Thanos 与 VictoriaMetrics，谁才是打造大型 Prometheus 监控系统的王者？
更多精彩内容欢迎订阅我的博客:https://fuckcloudnative.io/ Thanos[1] 和 VictoriaMetrics[2] 都是用来作为 Prometheus 长期存储的成熟方 ...
最新文章
- json数据解析_VBA 实践指南 -- Split函数解析Json数据
- FreeBSD中3D加速功能的启用
- pytorch argsort
- 含有负边的图的最短路径（Bellman_ford算法）
- mysql proxy读写分离实现_使用mysql-proxy实现读写分离
- Node.Js从零开始搭建数据管理后台 （一）
- 【推荐】Nginx基础知识之————多模块(非覆盖安装、RTMP在线人数实例安装测试)
- C++STL容器,你真的会用了吗？——插入、删除、遍历和查找操作性能对比——遍历和查找(精简易懂版,句句干货)
- python3爬虫初探（三）之正则表达式
- Python学习笔记字符串Str
- PHP页面显示中文字符出现乱码
- 单反相机内部光线反射原理
- JavaScript HTML DOM 事件（笔记）
- Silverlight C# 游戏开发：Flyer10完成寻找工作
- cacti 安装部署
- php登录与注册的心得,与discuz实现同步登陆和注册
- EMeeting会议预约系统软件，会议预约新方案
- Android用MediaRecorder实现MPEG4视频监控
- 名片小程序制作公司要如何选择？
- B001 - 基于STM32的智能生态鱼缸