Python抓取股價歷史資料
用Python 抓取股價歷史資料|股票量化交易從零開始
在本章節中,我們將會持續教你如何使用元富 API 抓取歷史的成交資料,以及劃出 K 線圖。
如果還不懂如何運用元富證券「下單」跟「行情」API 的認證,可以參考「Python與API串接的第一步」的相關文章喔!
Python 抓取股價歷史資料步驟
首先請到元富 API 專區的網頁,並且在「下載專區」的「技術指標」欄位中,下載「Python」 的技術指標套件。
請注意,因為目前技術指標的套件,只支援 32 bit 的 Python,所以我們需要先額外設定環境。
套件下載並解壓縮後,將檔案放到我們的工作目錄底下,並用 VS Code 開啟,記得在左上角先打開 Terminal,因為稍後要輸入指令。
上面步驟完成後,接著先切換到目錄底下: cd .\Python_tech_analysis\
再來需要輸入以下指令:
◆ 查看可安裝 Python 版本: pyenv install -l
◆ 安裝特定 Python 版本: pyenv install3.8.10-win32
◆ 使用3.8.10-win32做為這個專案資料夾的python 版本:pyenv local 3.8.10-win32
◆ 確認一下現在的版本:
◆ 接下來創建 venv 環境: python -m venv venv
◆ 啟用 venv 虛擬環境: venv\\Scripts\\activate
指令輸入完成後,這時我們左邊的目錄應該會跟下圖所示一樣:
接著找到「venv」的資料夾,會看到同一資料夾底下有「example.py」跟一個「.whl」檔案,接著請依照下列指示安裝套件:
◆ 安裝套件: pip install .\\tech_analysis_api_v2-0.0.3-py3-none-win32.whl
套件安裝完成後,這樣我們就完成環境的建置。
完成環境建置後,接著我們要使用官方的範例,來測試資料的抓取。
第一步開啟「example.py」檔案,這一部分也可以參考元富的官網說明。
接著在「main function」內,修改自己的帳號密碼(黃色塗鴉處)。
並且將「股票代號 2330」後面的日期,改成昨天儲存之後,執行此程式。如果這時候你在「Terminal」看到如下圖的畫面,這樣就表示執行成功。
Python 抓取多筆歷史資料&劃K線
接下來的步驟,我們會使用筆者所撰寫的程式,透過元富 API 撈取台積電的股價資料,並繪製成 K 線圖。
如果這部分看不懂的話,也可以直接使用文章最後的完整程式碼,只要複製貼上,並且儲存後執行即可。
首先我們需要額外安裝一些 Python 套件:
◆ 升級pip: python3 -m pip install –upgrade pip
◆ 可能會有相容性問題所以需要手動安裝Pillow這個套件:pip install –only-binary Pillow Pillow
◆ 安裝需要的套件: pip install numpy mplfinance
註:「numpy」是 python 做資料處理時,很常用且重要的套件;「mplfinance」則是方便我們繪圖用的套件。
套件安裝完成後,接下來我們在同一個資料夾裡,新增一個新的檔案,至於檔案名稱則隨意,像我是命名為「historyDataAndPlot.py」並貼入下列程式碼:
from tech_analysis_api_v2.api import TechAnalysis
from tech_analysis_api_v2.model import *
import threading
import pandas as pd
from datetime import datetime, timedelta
import mplfinance as mpf
def OnDigitalSSOEvent(aIsOK, aMsg):
print(f'OnDigitalSSOEvent:{aIsOK} {aMsg}')
def OnTAConnStuEvent(aIsOK):
print(f'OnTAConnStuEvent:{aIsOK}')
if aIsOK:
event.set()
def OnUpdate(ta_Type: eTA_Type, aResultPre, aResultLast):
if aResultPre != None:
if ta_Type ==eTA_Type.SMA:
print(f'前K {str(aResultPre)}')
if ta_Type ==eTA_Type.EMA:
print(f'前K {str(aResultPre)}')
if ta_Type ==eTA_Type.WMA:
print(f'前K {str(aResultPre)}')
if ta_Type ==eTA_Type.SAR:
print(f'前K {str(aResultPre)}')
if ta_Type ==eTA_Type.RSI:
print(f'前K {str(aResultPre)}')
if ta_Type ==eTA_Type.MACD:
print(f'前K {str(aResultPre)}')
if ta_Type ==eTA_Type.KD:
print(f'前K {str(aResultPre)}')
if ta_Type ==eTA_Type.CDP:
print(f'前K {str(aResultPre)}')
if aResultLast != None:
if ta_Type == eTA_Type.SMA:
print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply},SMA:{aResultLast.Value}')
if ta_Type ==eTA_Type.EMA:
print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply},EMA:{aResultLast.Value}')
if ta_Type ==eTA_Type.WMA:
print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply},EMA:{aResultLast.Value}')
if ta_Type ==eTA_Type.SAR:
print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply},SAR:{aResultLast.SAR}, EPh:{aResultLast.EPh}, EPl:{aResultLast.EPl},AF:{aResultLast.AF}, RaiseFall:{aResultLast.RaiseFall}')
if ta_Type ==eTA_Type.RSI:
print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply},RSI:{aResultLast.RSI}, UpDn:{aResultLast.UpDn}, UpAvg:{aResultLast.UpAvg},DnAvg:{aResultLast.DnAvg}')
if ta_Type ==eTA_Type.MACD:
print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply},DIF:{aResultLast.DIF}, OSC:{aResultLast.OSC}')
if ta_Type ==eTA_Type.KD:
print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply}, K:{aResultLast.K},D:{aResultLast.D}')
if ta_Type ==eTA_Type.CDP:
print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply},CDP:{aResultLast.CDP}, AH:{aResultLast.AH}, NH:{aResultLast.NH},AL:{aResultLast.AL}, NL:{aResultLast.NL}')
def OnRcvDone(ta_Type: eTA_Type, aResult):
if ta_Type ==eTA_Type.SMA:
for x in aResult:
print(f'回補 {x}')
if ta_Type ==eTA_Type.EMA:
for x in aResult:
print(f'回補 {x}')
if ta_Type ==eTA_Type.WMA:
for x in aResult:
print(f'回補 {x}')
if ta_Type ==eTA_Type.SAR:
for x in aResult:
print(f'回補 {x}')
if ta_Type ==eTA_Type.RSI:
for x in aResult:
print(f'回補 {x}')
if ta_Type == eTA_Type.MACD:
for x in aResult:
print(f'回補 {x}')
if ta_Type == eTA_Type.KD:
for x in aResult:
print(f'回補 {x}')
if ta_Type ==eTA_Type.CDP:
for x in aResult:
print(f'回補 {x}')
def option():
ProdID = input("商品代號: ")
SNK = input("分K(1/3/5): ")
STA_Type = input("指標(SMA/EMA/WMA/SAR/RSI/MACD/KD/CDP): ")
DateBegin = input("日期(ex: 20230619): ")
NK = eNK_Kind.K_1m
if SNK == '1':
NK = eNK_Kind.K_1m
elif SNK == '3':
NK = eNK_Kind.K_3m
elif SNK == '5':
NK = eNK_Kind.K_5m
TA_Type = eTA_Type.SMA
if STA_Type == 'SMA':
TA_Type = eTA_Type.SMA
elif STA_Type == 'EMA':
TA_Type = eTA_Type.EMA
elif STA_Type == 'WMA':
TA_Type = eTA_Type.WMA
elif STA_Type == 'SAR':
TA_Type = eTA_Type.SAR
elif STA_Type == 'RSI':
TA_Type = eTA_Type.RSI
elif STA_Type == 'MACD':
TA_Type =eTA_Type.MACD
elif STA_Type == 'KD':
TA_Type = eTA_Type.KD
elif STA_Type == 'CDP':
TA_Type = eTA_Type.CDP
returnTechAnalysis.get_k_setting(ProdID, TA_Type, NK, DateBegin)
event = threading.Event()
def fetch_historical_data(prod_id='2330', date='20231115'):
"""
將帳號密碼取代為你的帳號密碼,
填入商品代號與日期,抓取指定日期的歷史成交資料。
"""
ta =TechAnalysis(OnDigitalSSOEvent, OnTAConnStuEvent, OnUpdate, OnRcvDone)
ta.Login('{{你的身分證字號}}', '{{你的密碼}}') # Replace with your credentials
event.wait()
lsBS, sErrMsg =ta.GetHisBS_Stock(prod_id, date)
if sErrMsg:
print(f"Error:{sErrMsg}")
return None
# Converting data toDataFrame
data = [{
'ProdID': x.Prod,
'Match_Time':x.Match_Time,
'Match_Price':x.Match_Price,
'Match_Quantity':x.Match_Quantity,
'Is_TryMatch':x.Is_TryMatch,
'BS': x.BS
} for x in lsBS]
df = pd.DataFrame(data)
# 應用轉換
df['Formatted_Time'] =df['Match_Time'].apply(convert_time)
return df
def convert_time(time_val):
# 先轉成 String比較好處理 因為資料是回復一串數字
time_
其實前面大部分的程式碼,都跟之前的範例檔案相同,但在這裡我們重寫抓取歷史資料的部分(fetch_historical_data),並且將其資料整理(aggregate_to_ohlc)成可以畫成 K 線圖的格式(須要有OHLC資料),然後在輸出成圖表(mpf.plot)。
所以在「terminal」那邊輸入以下執行程式時,你應該會看到如下圖的輸出:
python historyDataAndPlot.py
這樣就代表我們成功抓取資料,並且畫出 K 線圖囉,那麼本章就介紹到這邊!