用python可以做哪些有趣的事–我：選股票
- 2019 年 10 月 7 日
- 筆記
最近炒股是買什麼就跌，一直是虧損哎，哭，作為學過python的人來講怎麼能容忍，之前也炒過股票覺得用陽包陰這樣的k線來選出來的股票還不錯。於是說做就做，我可以用python來寫一個選股的程序。
好！有了idea是第一步，要怎麼實現呢，網上找了資料，大部分都是用tushare庫來獲取股票數據的。於是動起來 寫了一個直接通過接口獲取數據的程序，從3504隻股票裏面選取出來我需要的股票，執行時間居然需要二十多分鐘， 太慢！差評！同樣不能容忍。因此，我想到了數據庫。我就想能不能將所有的A股數據添加進數據庫裏面，我每次執行的時候直接從數據庫裏面去取數據， 這樣會大大加快了我的執行速度 於是說干就干，先理清楚思路。1.需要獲取到所有股票的代碼跟名稱等。於是有了writeallstock這個文件 2.需要從所有的股票裏面找出陽包陰的股票，以及計算出它們的收益率的話，我需要所有股票的一段時間的行情 於是有了createverydatebase 3.有了這一段時間的數據，但是這些數據時死了，不會每天給我自動更新，因此我需要每天定時的將當天的數據加 進去。所以寫了writeeveryday 4.好了，所有的股票數據一段時間的行情而且會每天定時更新都存在我的數據庫裏面了，就需要去統計今天有哪些股票滿足 陽包陰的情況於是產生了findstock 5.雖然找到了當天滿足陽包陰的股票了，但是我心裏還是沒有譜，我想對比一下這個股票之前出現這種情況的時候如果 第二天買入的話到底有多少收益，所以有了winrates 6.好了整體框架和思路都出來了，那麼有兩個文件需要每個交易日都執行的，所以將它們綁在一起，而且每天的報告出來 之後也不一定都有時間打開電腦去看，所以加入了通過郵件自動發送當天報告到郵箱的功能。就有了runall 至於其它的幾個文件，打開看看下面都有解釋和注釋 目前我的選股條件是陽包陰，而且當天要漲停。
下面送上代碼。執行代碼的前提是，需要有python，需要安裝本地mysql數據庫，還需要導入mysql、tushare、pandas、lxml、bs4等需要支持的python庫。
另外：完整項目也可以訪問我的github，地址：https://github.com/cat-steel/stock_pick
rite_allstock.py
import mysql.connector import tushare as ts #將所有的股票名稱和股票代碼、行業、地區寫入到名為allstock的表中，這個文件只需要執行一次 #通過tushare庫獲取所有的A股列表 stock_info = ts.get_stock_basics() #連接數據庫 conn = mysql.connector.connect(user='root',password='password',database='test') cursor = conn.cursor() codes = stock_info.index names = stock_info.name industrys = stock_info.industry areas = stock_info.area #通過for循環遍歷所有股票，然後拆分獲取到需要的列，將數據寫入到數據庫中 a=0 for i in range(0,len(stock_info)): cursor.execute('insert into allstock (code,name,industry,area) values (%s,%s,%s,%s)',(codes[i],names[i],industrys[i],areas[i])) a += 1 #統計所有A股數量 print('共獲取到%d支股票'%a) conn.commit() cursor.close() conn.close()
creat_everydatebase.py
import tushare as ts import mysql.connector import re,time #創建所有股票的表格以及插入每支股票的近段時間的行情，這個文件只需要執行一次！！！ #想要寫入哪一段時間的數據只需要修改starttime,endtime的時間就可以了 def everdate(starttime,endtime): #獲取所有有股票 stock_info = ts.get_stock_basics() #連接數據庫 conn = mysql.connector.connect(user='root',password='password',database='test') cursor = conn.cursor() codes = stock_info.index a = 0 #通過for循環以及獲取A股只數來遍歷每一隻股票 for x in range(0,len(stock_info)): #匹配深圳股票（因為整個A股太多，所以我選擇深圳股票做個篩選） if re.match('000',codes[x]) or re.match('002',codes[x]): #以stock_加股票代碼為表名稱創建表格 cursor.execute('create table stock_' + codes[x] + ' (date varchar(32),open varchar(32),close varchar(32),high varchar(32),low varchar(32),volume varchar(32),p_change varchar(32),unique(date))') #利用tushare包獲取單只股票的階段性行情 df = ts.get_hist_data(codes[x],starttime,endtime) print('%s的表格創建完成'%codes[x]) a += 1 #這裡使用try，except的目的是為了防止一些停牌的股票，獲取數據為空，插入數據庫的時候失敗而報錯 #再使用for循環遍歷單只股票每一天的行情 try: for i in range(0,len(df)): #獲取股票日期，並轉格式（這裡為什麼要轉格式，是因為之前我2018-03-15這樣的格式寫入數據庫的時候，通過通配符%之後他居然給我把-符號當做減號給算出來了查看數據庫日期就是2000百思不得其解想了很久最後決定轉換格式） times = time.strptime(df.index[i],'%Y-%m-%d') time_new = time.strftime('%Y%m%d',times) #插入每一天的行情 cursor.execute('insert into stock_'+codes[x]+ ' (date,open,close,high,low,volume,p_change) values (%s,%s,%s,%s,%s,%s,%s)' % (time_new,df.open[i],df.close[i],df.high[i],df.low[i],df.volume[i],df.p_change[i])) except: print('%s這股票目前停牌'%codes[x]) conn.close() cursor.close() #統計總共插入了多少張表的數據 print('所有股票總共插入數據庫%d張表格'%a) everdate('2018-01-01','2018-03-14')
write_everyday.py
#每天下午三點之後進行股票數據添加到數據庫，這個文件一般只需要每天執行一次，也可以用來補行情，如果數據庫缺少那天的數據的話，只需修改new_time就行，如下示例 import tushare as ts import mysql.connector import re,time #每天行情出來了之後，插入當天的行情到每支股票的每個表格中 def everystock(): #獲取所有股票列表 stock_info = ts.get_stock_basics() #獲取股票代碼列 codes = stock_info.index #連接數據庫 conn = mysql.connector.connect(user='root',password='password',database='test') cursor = conn.cursor() #獲取當前時間 new_time = time.strftime('%Y-%m-%d') #new_time = '2018-03-13' a = 0 ##使用for循環遍歷所有的股票 for x in range(0,len(stock_info)): try: if re.match('000',codes[x]) or re.match('002',codes[x]): #獲取單只股票當天的行情 df = ts.get_hist_data(codes[x],new_time,new_time) #將時間轉換格式 times = time.strptime(new_time,'%Y-%m-%d') time_new = time.strftime('%Y%m%d',times) # #將當天的行情插入數據庫 cursor.execute('insert into stock_'+codes[x]+ ' (date,open,close,high,low,volume,p_change) values (%s,%s,%s,%s,%s,%s,%s)' % (time_new,df.open[0],df.close[0],df.high[0],df.low[0],df.volume[0],df.p_change[0])) print('%s的數據插入完成'%codes[x]) a += 1 except: print('%s無行情或者數據庫已經存在當天的數據'%codes[x]) #統計當天插入數據庫的股票數量 dir_log = 'D:\python\work\stock\WD\runlog\' filename = dir_log + new_time +'.log' flog = open(filename,'w') flog.write('今天的行情插入完成%s條'%a) # print('今天的行情插入完成%s條'%a) flog.close() conn.commit() conn.close() cursor.close() #everystock()
find_stock.py
import mysql.connector import re,time import datetime,os #從數據庫獲取股票數據，統計想要查找日期的滿足陽包陰並且當天漲停的股票 def valid_stock(dates): #載入日誌，好查錯（因為之前統計出來的股票我去實時查了一下完全不滿足條件，所以想到了加入日誌好定位是哪個地方出錯了） dir_log = 'D:\python\work\stock\WD\runlog\' filename = dir_log + dates +'.log' flog = open(filename,'w') # 先將字符串格式的時間轉換為時間格式才能計算昨天的日期 now = datetime.date(*map(int,dates.split('-'))) oneday = datetime.timedelta(days=1) yestody = str(now - oneday) #將昨天日期轉換為規定的字符串格式 times = time.strptime(yestody,'%Y-%m-%d') str_yestoday = time.strftime('%Y%m%d',times) flog.write('執行的時間前一天是%sn'%str_yestoday) #將想要查找的日期轉換為規定的字符串格式 str_today = time.strptime(dates,'%Y-%m-%d') today = time.strftime('%Y%m%d',str_today) flog.write('執行的時間是%sn'%today) #連接數據庫 conn = mysql.connector.connect(user='root',password='password',database='test') cursor = conn.cursor() #查找allstock表獲取所有股票代碼 cursor.execute('select code from allstock') value_code = cursor.fetchall() a = 0 count = [] #遍歷所有股票 for i in range(0,len(value_code)): if re.match('000',value_code[i][0]) or re.match('002',value_code[i][0]): #查詢所有匹配到的股票，將今天與昨天的數據對比 try: cursor.execute('select * from stock_'+ value_code[i][0]+ ' where date=%s or date =%s order by date desc'%(today,str_yestoday)) #當天 #cursor.execute('select * from stock_'+ value_code[i][0]+ ' where date=%s or date =%s'%('20180315','20180314')) value = cursor.fetchall() #1是昨天，2是今天 #今天的開盤價 opens1 = float(value[0][1]) #今天的收盤價 close1 = float(value[0][2]) #今天的漲幅 p_change1 = float(value[0][6]) #昨天的。。。。。 opens2 = float(value[1][1]) close2 = float(value[1][2]) p_change2 = float(value[1][6]) #加入這兩天的數據滿足昨天下跌超過2%，而且今天的開盤價低於昨天的收盤價，且今天的收盤價高於昨天的收盤價，就滿足陽包陰的條件 if opens2<close1 and close2>opens1 and p_change2<-2 and p_change1>9.8: flog.write('%s票%s的開盤價是%sn'%(value_code[i][0],today,opens1)) flog.write('%s票%s的收盤價是%sn'%(value_code[i][0],today,close1)) flog.write('%s票%s的漲幅是%sn'%(value_code[i][0],today,p_change1)) flog.write('%s票%s的開盤價是%sn'%(value_code[i][0],str_yestoday,opens2)) flog.write('%s票%s的收盤價價是%sn'%(value_code[i][0],str_yestoday,close2)) flog.write('%s票%s的漲幅是%sn'%(value_code[i][0],str_yestoday,p_change2)) #將滿足條件的股票代碼放進列表中，統計當天滿足條件的股票 count.append(value_code[i][0]) a += 1 except: #之前有次sql語句出錯了，order by後面沒加date，每次尋找都是0支，找了半個多小時才找出來是sql語句的問題 flog.write('%s停牌無數據,或者請查看sql語句是否正確n'%value_code[i][0])#一般不用管，除非執行好多天的數據都為0時那可能輸sql語句有問題了 print('總共找到%d支滿足條件的股票'%a) flog.close() conn.close() cursor.close() return count,a #valid_stock('2018-3-1')
win_rates.py
#這個文件可以聯合find_stock單獨運行，輸入todays的日期可以直接查找當天出現過的股票 import mysql.connector import re,time import datetime import find_stock #統計當天滿足陽包陰所有股票，在設置的這段時間裏面有沒有出現過類似的行情，並且計算如果出現過，那麼那天之後的5天收益率是多少 def rate(todays): print(todays) #將滿足陽包陰的這些股票，以及它們之前滿足的時候收益率都寫到報告裏面方便查看整體情況 count,a = find_stock.valid_stock(todays) dir_repor = 'D:\python\work\stock\WD\run\report\' filename = dir_repor + todays +'.txt' fp = open(filename,'w') fp.write('總共找到%d支滿足條件的股票分別是n%sn'%(a,count)) #連接數據庫 conn = mysql.connector.connect(user='root',password='password',database='test') cursor = conn.cursor() #遍歷滿足條件的這些股票 for x in count: #從數據庫里挑出它們的行情 cursor.execute('select * from stock_'+x+' order by date desc') value = cursor.fetchall() # print(value) for i in range(0,len(value)): #遍歷這支股票的所有天數 try: dates = value[i][0] opens2 = float(value[i][1]) #第i行的第一列 opens1 = float(value[i+1][1]) close2 = float(value[i][2]) #第i行的第二列 close1 = float(value[i+1][2]) p_change1 = float(value[i+1][6]) #第i行的第六列 p_change2 = float(value[i][6]) if opens2<close1 and close2>opens1 and p_change1<-2 and p_change2>9.8: #這一句很重要，就是在出現陽包陰之後得有5天的數據區統計，否則就會變成-5就會從開始統計的那天去取數據，結果就導致當天的這些股票統計收益的時候也有不過都是錯的 if i-6>0: #收益率 wins = (float(value[i-6][2])-float(value[i-1][1]))/float(value[i-1][1])*100 print('%s的%s之後5天收率為百分之%d'%(x,dates,wins)) fp.write('%s在%s之後5天收率為百分之%dn'%(x,dates,wins)) else: fp.write('%s在%s之前沒有滿足條件的行情n'%(x,dates)) except: pass # print('%s前3個月無滿足條件的情況'%x) fp.close() conn.close() cursor.close() #rate('2018-03-16')
run_all.py
#總的運行文件，實現將統計報告發送郵件到自己的郵箱，將這個文件放到Jenkin上每個交易日下午3點之後運行就可以收到當天滿足行情的股票了 import win_rates import write_everyday import time import os import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.header import Header #獲取最新的文件 def new_file(test_report_dir): lists = os.listdir(test_report_dir) lists.sort(key = lambda fn:os.path.getmtime(test_report_dir + fn)) file_path = os.path.join(test_report_dir,lists[-1]) return file_path #發送郵件 def send_email(): f = open(new_file(test_report_dir),'rb') mail_body = f.read() # print(mail_body) f.close() #設置郵件服務器和賬號密碼 smtpserver = 'smtp.163.com' user = '[email protected]' password = '*****' #設置接收郵箱和主題 sender = user receiver = '[email protected]' subject = '今天的股票行情來啦' msg = MIMEMultipart('mixed') att = MIMEText(mail_body,'txt','utf-8') att['Content-Type'] = 'application/octet-stream' att['Content-Disposition'] = 'attachment; filename = "%s.txt"'%todays msg.attach(att) msg['From'] = user msg['To'] = receiver msg['Subject'] =Header(subject,'utf-8') smtp = smtplib.SMTP() smtp.connect(smtpserver,25) smtp.login(user,password) smtp.sendmail(sender,receiver,msg.as_string()) smtp.quit() if __name__ == '__main__': test_report_dir = 'D:\python\work\stock\WD\run\report\' #如果執行的不是當天的日期的話請將第一個todays注釋掉,最好也將write_everyday.everystock()注釋掉,用第二個todays todays = time.strftime('%Y-%m-%d') # todays = '2018-03-14' #如果不是交易日執行的話write_everyday會報錯，會報tushare獲取不到行情，所以請手動輸入日期並將下面一行注釋掉 write_everyday.everystock() time.sleep(3) win_rates.rate(todays) send_email()
delete_stock.py
import mysql.connector import re,time import datetime,os def delete(dates): conn = mysql.connector.connect(user='root',password='password',database='test') cursor = conn.cursor() cursor.execute('select code from allstock') value_code = cursor.fetchall() a = 0 for i in range(0,len(value_code)): if re.match('000',value_code[i][0]) or re.match('002',value_code[i][0]): cursor.execute('delete from stock_'+ value_code[i][0]+ ' where date=%s'%(dates)) #刪除重複的數據 a +=1 print('%s已刪除'%value_code[i][0]) print('共刪除%d支股票的數據'%a) conn.commit() conn.close() cursor.close() delete('20180313')
addsqlunique.py
import mysql.connector import re,time import datetime,os #因為之前創建表格的時候沒加唯一性約束，容易插入重複的數據，導致最後計算記過不準確，所以穿件一個函數給之前沒加約束的加上去 def addunique(): conn = mysql.connector.connect(user='root',password='password',database='test') cursor = conn.cursor() cursor.execute('select code from allstock') value_code = cursor.fetchall() a = 0 for i in range(0,len(value_code)): if re.match('000',value_code[i][0]) or re.match('002',value_code[i][0]): cursor.execute('alter table stock_'+ value_code[i][0]+ ' add unique (date)') #刪除重複的數據 print('%s已添加唯一性約束完成'%value_code[i][0]) conn.close() cursor.close() addunique()
可以選股了，拿去賺錢吧，虧錢了可不要賴我哦。
