您现在的位置：程式師世界 >> 編程語言 > >> 更多編程語言 >> Python

Python量化-如何獲取實時股票信息

編輯：Python

如何獲取實時股票信息

股票信息的接口有很多，之前大家常用的是新浪的，但在年初的時候，新浪的接口突然不能使用，給大家造成了很大的困擾，為此網上也有很多教程教大家如何從新浪獲取數據，跟著教程弄了半天也不行，索性換到126（也就是網易了），感覺速度都還不錯。

首先我們看下接口地址：http://api.money.126.net/data/feed/1000001,money.api

其中的1000001就是股票代碼了，跟新浪的不同，他的第一位代表交易所，後面6位是股票代碼

0：上交所
1：深交所
2：北交所

先通過浏覽器看下數據結構：

_ntes_quote_callback({
"1000001": {
"code": "1000001",
"percent": 0.002113,
"high": 14.25,
"askvol3": 1026758,
"askvol2": 810700,
"askvol5": 290493,
"askvol4": 461100,
"price": 14.23,
"open": 14.2,
"bid5": 14.18,
"bid4": 14.19,
"bid3": 14.2,
"bid2": 14.21,
"bid1": 14.22,
"low": 14.11,
"updown": 0.03,
"type": "SZ",
"bidvol1": 323600,
"status": 0,
"bidvol3": 244200,
"bidvol2": 673474,
"symbol": "000001",
"update": "2022/06/25 17:59:57",
"bidvol5": 343500,
"bidvol4": 145200,
"volume": 86604061,
"askvol1": 817268,
"ask5": 14.27,
"ask4": 14.26,
"ask1": 14.23,
"name": "平安銀行",
"ask3": 14.25,
"ask2": 14.24,
"arrow": "↑",
"time": "2022/06/24 16:00:58",
"yestclose": 14.2,
"turnover": 1227798687.09
}
});

可以看出_ntes_quote_callback()中的就是標准的json數據，我們只要通過正則表達式就可以取出。
我們先定義一個數據結構：

class NetTick:
def __init__(self, dict={}):
self.name = dict.get('name') # 股票名稱
self.yestclose = dict.get('yestclose') # 昨日收盤價
self.bidvol5 = dict.get('bidvol5') # 買5數量
self.bidvol4 = dict.get('bidvol4') # 買4數量
self.bidvol3 = dict.get('bidvol3') # 買3數量
self.bidvol2 = dict.get('bidvol2') # 買2數量
self.bidvol1 = dict.get('bidvol1') # 買1數量
self.bid5 = dict.get('bid5') # 買5價格
self.bid4 = dict.get('bid4') # 買4價格
self.bid3 = dict.get('bid3') # 買3價格
self.bid2 = dict.get('bid2') # 買2價格
self.bid1 = dict.get('bid1') # 買1價格
self.askvol5 = dict.get('askvol5') # 賣5數量
self.askvol4 = dict.get('askvol4') # 賣4數量
self.askvol3 = dict.get('askvol3') # 賣3數量
self.askvol2 = dict.get('askvol2') # 賣2數量
self.askvol1 = dict.get('askvol1') # 賣1數量
self.ask5 = dict.get('ask5') # 賣5價格
self.ask4 = dict.get('ask4') # 賣4價格
self.ask3 = dict.get('ask3') # 賣3價格
self.ask2 = dict.get('ask2') # 賣2價格
self.ask1 = dict.get('ask1') # 賣1價格
self.symbol = dict.get('symbol') # 股票代碼 第一位1：深交所 0：上交所 2北交所
self.volume = dict.get('volume') # 成交量
self.price = dict.get('price') # 當前價格
self.open = dict.get('open') # 開盤價
self.low = dict.get('low') # 最低價
self.high = dict.get('high') # 最高價
self.code = dict.get('code') # 去除標記為的股票代碼
self.turnover = dict.get('turnover') # 成交額
self.percent = dict.get('percent') # 漲跌幅
self.updown = dict.get('updown') # 漲跌金額

通過研究，我們發現126的接口支持多個股票查詢，那我們可以定義兩個方法，一個查單個，一個查多個，具體實現如下：

import requests
import re
from models.nettick import NetTick
from utils.packages import *
class NetEaseData:
@staticmethod
def get_realtime_data(symbol):
"""
網易的實時數據接口
:param symbol: 股票代碼
:return: Tick
"""
code = NetEaseData.convert_market(symbol)
try:
response = requests.get("http://api.money.126.net/data/feed/{},money.api".format(code)).text
re_find = NetEaseData.__re_find(response)
if re_find is not None:
find_stock = re_find.get(code)
if find_stock is not None:
return NetTick(find_stock)
except Exception as e:
logger.error('請求網易接口出錯，錯誤信息：{}'.format(e))
return None
@staticmethod
def convert_market(other_market_code=str):
"""
轉換通用股票代碼 sz sh bj開頭+股票代碼
"""
if other_market_code[0:2].lower() == 'sh':
return '0' + other_market_code[2:]
elif other_market_code[0:2].lower() == 'sz':
return '1' + other_market_code[2:]
else:
return '2' + other_market_code[2:]
@staticmethod
def get_realtime_datas(symbols=[]):
"""
網易的實時數據接口
:param symbols: 股票代碼列表
:return: Ticks列表
"""
codes = [NetEaseData.convert_market(code) for code in symbols]
result = []
try:
response = requests.get("http://api.money.126.net/data/feed/{},money.api".format(','.join(codes))).text
re_find = NetEaseData.__re_find(response)
if re_find is not None:
for code in re_find:
item = re_find[code]
result.append(NetTick(item))
except Exception as e:
logger.error('請求網易接口出錯，錯誤信息：{}'.format(e))
return result
@staticmethod
def __re_find(response):
find = re.findall(r"_ntes_quote_callback\((.*)\);", response)
if len(find) >= 1:
return to_obj(find[-1])
return None
if __name__ == '__main__':
ticks = NetEaseData.get_realtime_datas(['sh588000', 'sz000001', 'bj831010'])
[print(tick.symbol, tick.name, tick.price) for tick in ticks]
tick = NetEaseData.get_realtime_data('sz127045')
print(tick.symbol, tick.name, tick.price)

使用也非常簡單