滚动窗口 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()
喜欢请长按下图一键关注
更喜欢可以转发哦 ^_^
