公司需要使用到海關的三旬匯率
海關提供目前匯率與歷史三旬匯率
目前匯率(TXT):https://portal.sw.nat.gov.tw/APGQ/GC331!downLoad?formBean.downLoadFile=CURRENT_TXT
目前匯率(JSON):https://portal.sw.nat.gov.tw/APGQ/GC331!downLoad?formBean.downLoadFile=CURRENT_JSON
歷史匯率(TXT):https://portal.sw.nat.gov.tw/APGQ/GC331!downLoad?formBean.downLoadFile=HISTORY_TXT
對Python不熟悉, 邊google邊寫, 程式的語法與效能有些不是很恰當, 請見諒
import pandas as pd
import datetime
import calendar
def nextMonth(d):
#下一個月
if d.month == 12:
nextMonthDate = d.replace(year=d.year+1, month=1)
else:
nextMonthDate = d.replace(month=d.month+1)
return nextMonthDate
def getHistTxt():
col_names = ['code','year','month','secCode','buyValue','sellValue']
col_widths = [3, 3, 2, 1, 8, 8]
col_types = {'code':str,'year':int,'month':int,'secCode':int,'buyValue':str,'sellValue':str}
filePath = 'https://portal.sw.nat.gov.tw/APGQ/GC331!downLoad?formBean.downLoadFile=HISTORY_TXT'
df = pd.read_fwf(filePath, names=col_names, widths=col_widths, converters=col_types)
rangeDay = ['1','11','21'] #上旬起始日:01, 中旬起始日:11, 下旬起始日:21
#df['buyValue'] = df['buyValue'].apply(lambda x:x[:3]+'.'+x[3:])
#df['sellValue'] = df['sellValue'].apply(lambda x:x[:3]+'.'+x[3:])
#df[['buyValue','sellValue']] = df[['buyValue','sellValue']].astype('float')
# 格式整理
for rowIdx in df.index:
secCode = df.iloc[rowIdx,3] # 1.上旬/2.中旬/3.下旬
# 組日期字串, 民國年月 依 上中下旬轉為 西元年
startStr = str(df.iloc[rowIdx,1] + 1911) +'/'+ str(df.iloc[rowIdx,2]) +'/'+ rangeDay[secCode-1]
# 字串轉日期
sDate = pd.to_datetime(startStr, format='%Y/%m/%d')
# 三旬日期區間處理
if secCode==3:
# 3.下旬 : 21日到月底
eDate = nextMonth(sDate.replace(day=1)) - datetime.timedelta(days=1)
else:
# 1.上旬 : 1日到10日, 2.中旬 : 11日到20日
eDate = sDate + datetime.timedelta(days=9)
buyStr = df.iloc[rowIdx,4]
sellStr = df.iloc[rowIdx,5]
buyValue = float(buyStr[:3] + '.' + buyStr[3:])
sellValue = float(sellStr[:3] + '.' + sellStr[3:])
aList = [sDate, eDate] + list(df.iloc[rowIdx, [0]]) + [buyValue, sellValue]
aRow = pd.Series(aList) # 轉為Series
dfRow = pd.DataFrame([aRow]) # 轉為DataFrame
# 整併到新格式的df
if rowIdx==0:
dfCurr = dfRow
else:
dfCurr = pd.concat([dfCurr, dfRow], ignore_index=True)
# 設定column name
dfCurr.columns = ['start','end','code','buyValue','sellValue']
return dfCurr
def getCurrTxt():
# 文字檔gc331_current.txt的抬頭, 買進與賣出間有2個tab
# 幣別 年 月 旬 買進 賣出
# 使用 df.rename 修正 column name
df = pd.read_table('https://portal.sw.nat.gov.tw/APGQ/GC331!downLoad?formBean.downLoadFile=CURRENT_TXT', sep='\t', encoding='big5')
# 共6個column, 若為7, 表示抬頭真的多了一個tab, 將最後一個column刪除
if len(df.iloc[0]) == 7:
df.drop(columns=['賣出'], inplace=True)
# 重新定義抬頭, 改column的語法 : df.rename(columns={'Unnamed: 5': '賣出'}, inplace=True)
df.columns = ['code','year','month','secCode','buyValue','sellValue']
# 民國年月 依 上中下旬轉為 西元年
secCode = df.iloc[0,3] # 1.上旬/2.中旬/3.下旬
rangeDay = ['1','11','21'] #上旬起始日:01, 中旬起始日:11, 下旬起始日:21
# 組日期字串, 字串補零 str.zfill(2)
startStr = str(df.iloc[0,1] + 1911) +'/'+ str(df.iloc[0,2]) +'/'+ rangeDay[secCode-1]
# 字串轉日期
sDate = pd.to_datetime(startStr, format='%Y/%m/%d')
# 三旬日期區間處理
if secCode==3:
# 3.下旬 : 21日到月底
eDate = datetime.datetime(sDate.year, sDate.month, calendar.monthrange(sDate.year, sDate.month)[1])
else:
# 1.上旬 : 1日到10日, 2.中旬 : 11日到20日
eDate = sDate + datetime.timedelta(days=9)
# 格式整理
for rowIdx in df.index:
aList = [sDate, eDate] + list(df.iloc[rowIdx, [0,4,5]]) # 轉為list
aRow = pd.Series(aList) # 轉為Series
dfRow = pd.DataFrame([aRow]) # 轉為DataFrame
# 整併到新格式的df
if rowIdx==0:
dfCurr = dfRow
else:
dfCurr = pd.concat([dfCurr, dfRow], ignore_index=True)
# 設定column name
dfCurr.columns = ['start','end','code','buyValue','sellValue']
return dfCurr
def getCurrJson():
df = pd.read_json('https://portal.sw.nat.gov.tw/APGQ/GC331!downLoad?formBean.downLoadFile=CURRENT_JSON')
# 三旬起始日期
sDate = pd.to_datetime(str(df.iloc[0,0]), format='%Y%m%d')
# 三旬結束日期
eDate = pd.to_datetime(str(df.iloc[0,1]), format='%Y%m%d')
for rowIdx in df.index:
dic = df.iloc[rowIdx, 2] # 回傳dict
aList = [sDate, eDate, dic['code'], float(dic['buyValue']), float(dic['sellValue'])]
aRow = pd.Series(aList)
dfRow = pd.DataFrame([aRow])
if rowIdx==0:
dfCurr = dfRow
else:
dfCurr = pd.concat([dfCurr, dfRow], ignore_index=True)
# 設定column name
dfCurr.columns = ['start','end','code','buyValue','sellValue']
return dfCurr
dfCurrXR = getCurrJson()
print(dfCurrXR.head())
#print(dfCurrXR.dtypes)
print('-'*20)
dfCurrXR = getCurrTxt()
print(dfCurrXR.head())
#print(dfCurrXR.dtypes)
print('-'*20)
dfCurrXR = getHistTxt()
print(dfCurrXR.head(20))
#print(dfCurrXR.dtypes)
執行結果