""" author：漢江S 微博：漢江S """ from urllib.request import urlopen # python自帶爬蟲庫 import pandas as pd from datetime import datetime import time import re # 正則表達式庫 import os # 系統庫 import json # python自帶的json數據庫 pd.set_option('expand_frame_repr', False) # 當列太多時不換行 pd.set_option('display.max_rows', 5000) # 最多顯示數據的行數 # =====函數：從網頁上抓取數據 def get_content_from_internet(url, max_try_num=10, sleep_time=5): """ 使用python自帶的urlopen函數，從網頁上抓取數據 :param url: 要抓取數據的網址 :param max_try_num: 最多嘗試抓取次數 :param sleep_time: 抓取失敗后停頓的時間 :return: 返回抓取到的網頁內容 """ get_success = False # 是否成功抓取到內容 # 抓取內容 for i in range(max_try_num): try: content = urlopen(url=url, timeout=10).read() # 使用python自帶的庫，從網絡上獲取信息 get_success = True # 成功抓取到內容 break except Exception as e: print('抓取數據報錯，次數：', i+1, '報錯內容：', e) time.sleep(sleep_time) # 判斷是否成功抓取內容 if get_success: return content else: raise ValueError('使用urlopen抓取網頁數據不斷報錯，達到嘗試上限，停止程序，請盡快檢查問題所在') # =====函數：從新浪獲取指定股票的數據 def get_today_data_from_sinajs(code_list): """ 返回一串股票最近一個交易日的相關數據 從這個網址獲取股票數據：http://hq.sinajs.cn/list=sh600000,sz000002,sz300001 正常網址：https://finance.sina.com.cn/realstock/company/sh600000/nc.shtml, :param code_list: 一串股票代碼的list，可以多個，例如[sh600000, sz000002, sz300001], :return: 返回一個存儲股票數據的DataFrame """ # 構建url url = "http://hq.sinajs.cn/list=" + ",".join(code_list) # 抓取數據 content = get_content_from_internet(url) content = content.decode('gbk') # 將數據轉換成DataFrame content = content.strip() # 去掉文本前后的空格、回車等 data_line = content.split('\n') # 每行是一個股票的數據 data_line = [i.replace('var hq_str_', '').split(',') for i in data_line] df = pd.DataFrame(data_line, dtype='float') # # 對DataFrame進行整理 df[0] = df[0].str.split('="') df['stock_code'] = df[0].str[0].str.strip() df['stock_name'] = df[0].str[-1].str.strip() df['candle_end_time'] = df[30] + ' ' + df[31] # 股票市場的K線，是普遍以當跟K線結束時間來命名的 df['candle_end_time'] = pd.to_datetime(df['candle_end_time']) rename_dict = {1: 'open', 2: 'pre_close', 3: 'close', 4: 'high', 5: 'low', 6: 'buy1', 7: 'sell1', 8: 'amount', 9: 'volume', 32: 'status'} # 自己去對比數據，會有新的返現 # 其中amount單位是股，volume單位是元 df.rename(columns=rename_dict, inplace=True) df['status'] = df['status'].str.strip('";') df = df[['stock_code', 'stock_name', 'candle_end_time', 'open', 'high', 'low', 'close', 'pre_close', 'amount', 'volume', 'buy1', 'sell1', 'status']] return df # test = get_today_data_from_sinajs(code_list=['sh000001','sz399001']) # print(test) # =====函數：判斷今天是否是交易日 def is_today_trading_day(): """ 判斷今天是否是交易日 :return: 如果是返回True，否則返回False """ # 獲取上證指數今天的數據 df = get_today_data_from_sinajs(code_list=['sh000001','sz399001']) sh_date = df.iloc[0]['candle_end_time'] # 上證指數最近交易日 # 判斷今天日期和sh_date是否相同 return datetime.now().date() == sh_date.date() # test3 = is_today_trading_day() # print(test3) # =====函數：從新浪獲取所有股票的數據 def get_all_today_stock_data_from_sina_marketcenter(): """ http://vip.stock.finance.sina.com.cn/mkt/#stock_hs_up 從新浪網址的上述的網址，逐頁獲取最近一個交易日所有股票的數據 :return: 返回一個存儲股票數據的DataFrame """ # ===數據網址 headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.2704.103 Safari/537.36", } #股票數據 raw_url = 'http://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/Market_Center.getHQNodeData?page=%s' \ '&num=40&sort=symbol&asc=1&node=hs_a&symbol=&_s_r_a=sort' #可轉債數據 raw_url = 'http://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/Market_Center.getHQNodeDataSimple?page=%s&num=80&sort=symbol&asc=0&node=hskzz_z&_s_r_a=sort' page_num = 1 # ===存儲數據的DataFrame all_df = pd.DataFrame() # ===獲取上證指數最近一個交易日的日期。此段代碼在課程視頻中沒有，之后補上的 df = get_today_data_from_sinajs(code_list=['sh000001','sz399001']) sh_date = df.iloc[0]['candle_end_time'].date() # 上證指數最近交易日 # ===開始逐頁遍歷，獲取股票數據 while True: # 構建url url = raw_url % (page_num) print('開始抓取頁數：', page_num) # 抓取數據 content = get_content_from_internet(url) content = content.decode('gbk') # 判斷頁數是否為空 if '[]' in content: print('抓取到頁數的盡頭，退出循環') break # 通過正則表達式，給key加上引號 content = re.sub(r'(?<={|,)([a-zA-Z][a-zA-Z0-9]*)(?=:)', r'"\1"', content) # 將數據轉換成dict格式 content = json.loads(content) # 將數據轉換成DataFrame格式 df = pd.DataFrame(content, dtype='float') # 對數據進行整理 # 重命名 rename_dict = {'symbol': '股票代碼', 'name': '股票名稱', 'open': '開盤價', 'high': '最高價', 'low': '最低價', 'trade': '收盤價', 'settlement': '前收盤價', 'volume': '成交量', 'amount': '成交額'} df.rename(columns=rename_dict, inplace=True) # 添加交易日期 df['交易日期'] = pd.to_datetime(sh_date) # 在課程視頻中使用的是上一行代碼，現在改成本行代碼，程序更加穩健 # 取需要的列 df = df[['股票代碼', '股票名稱', '交易日期', '開盤價', '最高價', '最低價', '收盤價', '前收盤價', '成交量', '成交額']] # 合並數據 all_df = all_df.append(df, ignore_index=True) # 將頁數+1 page_num += 1 time.sleep(1) # ===將當天停盤的股票刪除，此段代碼在課程視頻中沒有，之后補上的 all_df = all_df[all_df['開盤價'] - 0 > 0.00001] all_df.reset_index(drop=True, inplace=True) # ===返回結果 return all_df # test = get_all_today_stock_data_from_sina_marketcenter() # print(test) # 判斷今天是否是交易日 if is_today_trading_day() is False: print('今天不是交易日，不需要更新股票數據，退出程序') exit() # 判斷當前時間是否超過15點 if datetime.now().hour < 15: # 保險起見可以小於16點 print('今天股票尚未收盤，不更新股票數據，退出程序') exit() # 獲取今天所有的股票數據 df = get_all_today_stock_data_from_sina_marketcenter() # 對數據進行存儲 for i in df.index: t = df.iloc[i:i+1, :] stock_code = t.iloc[0]['股票代碼'] # 構建存儲文件路徑 path = '../data/'+ stock_code + '.csv' # 文件存在，不是新股 if os.path.exists(path): t.to_csv(path, header=None, index=False, mode='a', encoding='gbk') # 文件不存在，說明是新股 else: # 先將頭文件輸出 t.to_csv(path, index=False, mode='a', encoding='gbk') print(stock_code)