滾動視窗 rolling 比對計算
調用滾動視窗函數 rolling() 實作統計值的滾動計算，比如求平均值、最大值、最小值、标準差(預設有偏，無偏需要設定ddof=0)，不需要逐行周遊，提升代碼執行效率。
應用場景
針對一個期貨品種的主力合約行情，在不考慮主力合約換月帶來價格跳空影響的前提下(僅限學習使用)，分别采取逐行周遊的方式、調用 rolling() 函數實作滾動視窗的方式計算統計值(平均值、最大值、最小值、标準差)，比對輸出。
算法步驟
1、讀取期貨市場結算行情檔案，提取指定品種的結算行情;
2、将資料按照交易日升序、持倉量降序、合約升序的順序排序，按交易日分組提取首記錄(主力合約行情)，篩選出後200條記錄;
3、提取交易日、持倉量列資料，重置序号(0...199);
4、采取逐行周遊的方式滾動計算固定一段時期内持倉量的統計值;
5、調用 rolling() 實作滾動視窗，然後取統計函數;
6、将交易日字段設定為序号;
7、格式化字段值為兩位小數，儲存為csv格式的檔案.
python代碼
#coding=utf-8import numpy as np , pandas as pd,talib as tadef calRolling(quote_data, n_days): """ 調用 Pabdas 的内置函數 rolling() 實作滾動視窗 的比對計算 """ ## 提取交易日、持倉量,資料類型設定為整數型 col_fields = ['TradingDay', 'OpenInterest'] my_data = quote_data[col_fields].copy() my_data.reset_index(drop=True, inplace=True) my_data['OpenInterest'] = my_data['OpenInterest'].astype(np.int32) ## 根據定義計算 rolling 的相關函數 for i in range(len(my_data)) : if i < n_days -1 : my_data.loc[i, 'mean_P'] = my_data.loc[i, 'max_P'] = my_data.loc[i, 'min_P'] = my_data.loc[i, 'std_P'] = np.nan else: my_data.loc[i, 'mean_P'] = np.mean(my_data['OpenInterest'][i+1-n_days:i+1]) my_data.loc[i, 'max_P'] = np.max(my_data['OpenInterest'][i+1-n_days:i+1]) my_data.loc[i, 'min_P'] = np.min(my_data['OpenInterest'][i+1-n_days:i+1]) my_data.loc[i, 'std_P'] = np.std(my_data['OpenInterest'][i+1-n_days:i+1], ddof=1) ## 調用 Pandas.DataFrame.rolling() 實作滾動視窗求平均值、最大值、最小值、标準差(有偏) my_data['mean'] = my_data['OpenInterest'].rolling(window = n_days).mean() my_data['max'] = my_data['OpenInterest'].rolling(window = n_days).max() my_data['min'] = my_data['OpenInterest'].rolling(window = n_days).min() my_data['std'] = my_data['OpenInterest'].rolling(window = n_days).std() ## 将交易日置為序号 my_data.set_index(['TradingDay'], inplace=True) ## 資料格式化為兩位小數 for i in range(len(my_data.columns)): my_data[my_data.columns[i]] = pd.Series(np.around(my_data[my_data.columns[i]], decimals=2)) print(my_data.head(n_days+3)) print(my_data.tail(3)) return my_datadef main(in_file, variety, n_days, out_file): ## 提取該品種的結算行情資料 my_data = pd.read_csv(in_file, dtype='str') my_data = my_data[my_data['VarietyID'].str.lower() == variety.lower()] if len(my_data) == 0: print('--- no data about variety '+variety+' ---') return ## 篩選出主力合約行情資料,并取後200條 my_data['OpenInterest'] = my_data['OpenInterest'].astype(np.int32) my_data.sort_values(['TradingDay', 'OpenInterest', 'ContractID'], ascending=[1, 0, 1], inplace=True) my_data = my_data.groupby(['TradingDay']).head(1).tail(200) ## 對比計算 my_data = calRolling(my_data, n_days) my_data.to_csv(out_file, encoding='utf-8', index=True)in_file = './FormatData/SHFE_DailyQuote_fut.csv'out_file = './FormatData/ta_diff_cal.csv'variety = 'au'n_days = 5main(in_file, variety, n_days, out_file) 運作結果圖
内置函數
調用 Pandas.DataFrame 的 rolling() 實作滾動視窗，便于滾動統計。類似于 DataFrame.rolling(window = n).mean()
喜歡請長按下圖一鍵關注
更喜歡可以轉發哦 ^_^
