[Python] 股價查詢

程式語言：Python

Package：

requests

time

logging

Alpha Vantage 官網
台灣證券交易所-基本市況報導

功能：股價查詢

Alpha Vantage (美股為主)

請先申請 API Key
API 說明
股票代碼，其他國家可嘗試加上後繏，例：0050.TW，但不一定有資料
使用方法

a = AlphaVantage(api_key="your key")
a.intraday('VTI', '1min')
a.daily('VTI')
a.daily_adj('VTI')
a.weekly('VTI')
a.weekly_adj('VTI')
a.monthly('VTI')
a.monthly_adj('VTI')
a.batch_quotes('VTI, VBR')

程式碼

import requests
import time
import logging
 
 
class AlphaVantage:
    def __init__(self, api_key):
        """
        api_key: 申請網址：https://www.alphavantage.co/support/#api-key
        """    
        self.api_key = api_key
        self.logger = logging.getLogger(self.__class__.__name__)
        
    def _api(self, func,  **params):
        """Return data
        func: alphaVantage 的 API
        params: API 的相關參數
        """    
        url = "https://www.alphavantage.co/query"
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 6.1; rv:57.0) Gecko/20100101 Firefox/57.0",
                  }
        baseParams = {"function": func,
                      "apikey": self.api_key,
                      "datatype": "json"}
        params = {**baseParams, **params} 
        r = requests.get(url, params=params, headers=headers)
        data = r.json()
        if "Error Message" in data:
            self.logger.debug('{} error => url: {}'.format(func, r.url))
            return None
        elif len(data) < 2:
            self.logger.debug('{} error => url: {}'.format(func, r.url))
            # AlphaVantage 有限定 1s 一次
            time.sleep(1)
            return self._api(func, **params)
            
        return data
 
 
    def intraday(self, symbol, interval, **params):
        """Return data
        https://www.alphavantage.co/documentation/#intraday
        symbol: 股票代碼，其他國家可嘗試加上後繏，例：0050.TW，但不一定有資料
        interval: 間隔時間 (1min, 5min, 15min, 30min, 60min)
        params: 其他相關參數
        """
        return self._api("TIME_SERIES_INTRADAY", symbol=symbol, interval=interval, **params)
 
 
    def daily(self, symbol,  **params):
        """Return data
        https://www.alphavantage.co/documentation/#daily
        symbol: 股票代碼，其他國家可嘗試加上後繏，例：0050.TW，但不一定有資料
        params: 其他相關參數
        """
        return self._api("TIME_SERIES_DAILY", symbol=symbol, **params)
        
        
    def daily_adj(self, symbol,  **params):
        """Return data
        https://www.alphavantage.co/documentation/#dailyadj
        symbol: 股票代碼，其他國家可嘗試加上後繏，例：0050.TW，但不一定有資料
        params: 其他相關參數
        """
        return self._api("TIME_SERIES_DAILY_ADJUSTED", symbol=symbol, **params)
        
 
    def weekly(self, symbol,  **params):
        """Return data
        https://www.alphavantage.co/documentation/#weekly
        symbol: 股票代碼，其他國家可嘗試加上後繏，例：0050.TW，但不一定有資料
        params: 其他相關參數
        """
        return self._api("TIME_SERIES_WEEKLY", symbol=symbol, **params)
        
        
    def weekly_adj(self, symbol,  **params):
        """Return data
        https://www.alphavantage.co/documentation/#weeklyadj
        symbol: 股票代碼，其他國家可嘗試加上後繏，例：0050.TW，但不一定有資料
        params: 其他相關參數
        """
        return self._api("TIME_SERIES_WEEKLY_ADJUSTED", symbol=symbol, **params)
        
        
    def monthly(self, symbol,  **params):
        """Return data
        https://www.alphavantage.co/documentation/#monthly
        symbol: 股票代碼，其他國家可嘗試加上後繏，例：0050.TW，但不一定有資料
        params: 其他相關參數
        """
        return self._api("TIME_SERIES_MONTHLY", symbol=symbol, **params)
        
        
    def monthly_adj(self, symbol,  **params):
        """Return data
        https://www.alphavantage.co/documentation/#monthlyadj
        symbol: 股票代碼，其他國家可嘗試加上後繏，例：0050.TW，但不一定有資料
        params: 其他相關參數
        """
        return self._api("TIME_SERIES_MONTHLY_ADJUSTED", symbol=symbol, **params)
        
        
    def batch_quotes(self, symbols,  **params):
        """Return data
        https://www.alphavantage.co/documentation/#batchquotes
        symbols: 所有股票的代碼，只支援美股
        params: 其他相關參數
        """
        return self._api("BATCH_STOCK_QUOTES", symbols=symbols, **params)

台灣證券交易所 (台股)

symbol: 股票代碼，上市後繏 .TW，上櫃後繏 .TWO
使用方法

t = TWSE()
t.real_time('0050.TW')
t.real_time('1258.TWO')

程式碼

import requests
import time
import logging
 
class TWSE:
    def __init__(self):
        self.logger = logging.getLogger(self.__class__.__name__)
        self.headers = {"User-Agent": "Mozilla/5.0 (Windows NT 6.1; rv:57.0) Gecko/20100101 Firefox/57.0",
                        "Connection": "keep-alive",
                       } 
        url = 'http://mis.twse.com.tw/stock/fibest.jsp'
        r = requests.get(url, headers=self.headers)
        self.cookies = r.cookies
    
    
    def _rename(self, symbol):
        """Return quote symbol name
        symbol 命名規則 上市或上櫃_股票代號.Channel，例：tse_0050.tw
        上市/上櫃，tse/otc
        """
        symbol = symbol.upper()
        try:
            symbol, kind = symbol.split(".", 1)
            if kind == "TWO":
                return "otc_{}.tw".format(symbol)
        except ValueError:
            pass
 
        return "tse_{}.tw".format(symbol)
        
        
    def real_time(self, symbol):
        """Return real-time price
        https://github.com/Asoul/tsrtc
        symbol: 股票代碼，上市後繏 .TW，上櫃後繏 .TWO
        """
        symbol = self._rename(symbol)
        url = "http://mis.twse.com.tw/stock/api/getStockInfo.jsp"
        millis = int(round(time.time() * 1000))
        params = {"ex_ch": symbol,
                  "_": millis,
                  "json": 1,
                  "delay": 0,
                 }
        r = requests.get(url, params=params, headers=self.headers, cookies=self.cookies)
        data = r.json()
        
        return data['msgArray']

參考

台灣股票即時爬蟲。Taiwan Stock Exchange Real Time Crawler