Python海关三旬汇率 - Python练习题三

公司需要使用到海关的三旬汇率
海关提供目前汇率与历史三旬汇率
目前汇率(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)

执行结果