时间序列进阶：特征工程/多步预测/异常检测
1. 时间序列特征工程
importpandasaspdimportnumpyasnpdefcreate_time_features(df,date_col):df=df.copy()df[date_col]=pd.to_datetime(df[date_col])df['hour']=df[date_col].dt.hour df['day']=df[date_col].dt.day df['day_of_week']=df[date_col].dt.dayofweek df['month']=df[date_col].dt.month df['is_weekend']=df['day_of_week'].isin([5,6]).astype(int)returndfdefcreate_lag_features(df,target_col,lags=[1,7,14,30]):forlaginlags:df[f'{target_col}_lag_{lag}']=df[target_col].shift(lag)returndfdefcreate_rolling_features(df,target_col,windows=[7,14,30]):forwindowinwindows:df[f'{target_col}_rolling_mean_{window}']=df[target_col].rolling(window).mean()df[f'{target_col}_rolling_std_{window}']=df[target_col].rolling(window).std()returndf
2. 多步预测
defdirect_multi_step(model,X_last,steps=7):predictions=[]forstepinrange(steps):pred=model.predict(X_last.reshape(1,-1))[0]predictions.append(pred)X_last=np.roll(X_last,-1)X_last[-1]=predreturnpredictionsdefrecursive_multi_step(model,X_last,steps=7):predictions=[]current_input=X_last.copy()for_inrange(steps):pred=model.predict(current_input.reshape(1,-1))[0]predictions.append(pred)current_input=np.append(current_input[1:],pred)returnpredictions
3. 时间序列异常检测
fromsklearn.ensembleimportIsolationForestdefdetect_anomalies(data,column,contamination=0.05):model=IsolationForest(contamination=contamination,random_state=42)data['anomaly']=model.fit_predict(data[[column]])data['is_anomaly']=data['anomaly']==-1returndatadefstatistical_anomaly(data,column,window=30,threshold=3):rolling_mean=data[column].rolling(window).mean()rolling_std=data[column].rolling(window).std()z_scores=(data[column]-rolling_mean)/rolling_std data['is_anomaly']=abs(z_scores)>thresholdreturndata
总结
| 技术 | 方法 | 适用场景 |
|---|
| 特征工程 | 滞后/滚动/时间 | 提升精度 |
| 多步预测 | 直接/递归 | 未来预测 |
| 异常检测 | IF/Z-Score | 故障检测 |
