尧图网站开发 
1. Pandas时间序列处理基础时间序列数据是数据分析中最常见的数据类型之一它记录了随时间变化的观测值。在金融、物联网、气象等领域时间序列数据无处不在。Pandas作为Python数据分析的核心库提供了强大的时间序列处理能力。1.1 时间序列数据的特点时间序列数据与普通表格数据最大的区别在于其索引通常是时间戳。这种数据结构带来了几个独特特性时间连续性数据点按固定或可变间隔排列季节性/周期性数据可能呈现周期性变化模式时间相关性相邻时间点的数据往往存在关联在Pandas中我们主要使用Timestamp对象表示单个时间点用DatetimeIndex表示时间索引。创建时间序列数据的基本方法是import pandas as pd # 创建时间范围 date_rng pd.date_range(start1/1/2023, end1/10/2023, freqD) # 创建时间序列数据 time_series pd.Series(range(10), indexdate_rng)1.2 时间序列的创建与转换实际工作中我们经常需要将各种格式的时间数据转换为Pandas可识别的时间序列。常见转换场景包括字符串转时间戳pd.to_datetime([2023-01-01, Jan 2, 2023, 03/01/2023])Unix时间戳转换pd.to_datetime([1672531200, 1672617600], units)处理混合格式pd.to_datetime([20230101, 2023-01-02, Jan 3, 2023], formatmixed)注意当处理大量日期字符串时指定format参数可以显著提高转换速度。例如format%Y%m%d比让Pandas自动推断要快得多。2. 时间序列索引与切片2.1 基础索引操作时间序列数据的索引操作是数据分析的基础。Pandas提供了多种灵活的方式来选择和切片时间序列数据# 选择特定日期 time_series[2023-01-05] # 选择日期范围 time_series[2023-01-03:2023-01-07] # 部分字符串匹配 time_series[2023-01] # 选择整个1月数据2.2 高级索引技巧对于更复杂的时间选择需求Pandas提供了强大的索引功能使用truncate方法截取时间范围time_series.truncate(before2023-01-04, after2023-01-08)使用between_time选择特定时间段适用于日内数据# 假设我们有日内数据 intraday pd.Series( range(24), indexpd.date_range(2023-01-01, periods24, freqH) ) intraday.between_time(9:00, 17:00) # 选择交易时段使用at_time选择特定时间点intraday.at_time(15:00) # 选择每天下午3点的数据3. 时间序列重采样与频率转换3.1 基础重采样操作重采样是时间序列分析中最常用的操作之一它可以将数据从一个频率转换到另一个频率# 创建示例数据 rng pd.date_range(2023-01-01, periods100, freqD) ts pd.Series(np.random.randn(len(rng)), indexrng) # 降采样日数据→月数据 ts.resample(M).mean() # 升采样日数据→小时数据 ts.resample(H).asfreq() # 产生NaN ts.resample(H).ffill() # 前向填充3.2 高级重采样技巧实际应用中我们经常需要更复杂的重采样逻辑自定义聚合函数def first_last(series): return series.iloc[[0, -1]] ts.resample(M).apply(first_last)分组重采样适用于多时间序列df pd.DataFrame({ group: [A]*50 [B]*50, value: np.random.randn(100) }, indexrng) df.groupby(group).resample(M, onindex)[value].mean()使用offset调整重采样锚点# 以每月第25天为锚点 ts.resample(MS).mean() # 默认月初 ts.resample(MS, offset25D).mean()提示处理财务数据时常用的重采样频率包括B工作日W-FRI每周五Q季度末A年末4. 时间序列的移动窗口操作4.1 滚动统计量滚动窗口计算是时间序列分析的核心技术之一用于计算移动平均值、标准差等统计量# 7天滚动平均 ts.rolling(window7).mean() # 带最小观测值要求的滚动计算 ts.rolling(window30, min_periods10).std() # 非对称窗口 ts.rolling(window5D).sum() # 5日历日4.2 扩展窗口与指数加权除了固定窗口Pandas还支持扩展窗口和指数加权移动平均# 扩展窗口累计 ts.expanding().mean() # 指数加权移动平均 ts.ewm(span30).mean() # 半衰期加权 ts.ewm(halflife15).std()4.3 高级窗口类型Pandas 1.3引入了更灵活的窗口类型可变偏移窗口ts.rolling(window5D, min_periods1).mean()自定义权重窗口def custom_weights(window): return np.arange(1, len(window)1) ts.rolling(window5).apply(lambda x: np.sum(custom_weights(x)*x)/sum(custom_weights(x)))事件驱动窗口基于索引距离而非观测次数ts.rolling(windowpd.Timedelta(days5)).mean()5. 时间序列缺失值处理5.1 常见缺失值场景时间序列数据中的缺失值可能由多种原因导致数据采集中断节假日/非交易日传感器故障数据合并时的日期不匹配5.2 缺失值处理方法Pandas提供了多种处理时间序列缺失值的方法插值法ts.interpolate(methodtime) # 按时间权重插值 ts.interpolate(methodspline, order3) # 三次样条插值前向/后向填充ts.ffill() # 前向填充 ts.bfill() # 后向填充复杂填充策略# 使用滚动平均值填充 ts.fillna(ts.rolling(5, min_periods1).mean())注意事项选择填充方法时应考虑数据特性。金融价格数据通常使用前向填充而气象数据可能更适合插值。6. 时区处理与时间转换6.1 时区基础操作处理跨时区数据是时间序列分析中的常见需求# 本地化时区 ts ts.tz_localize(UTC) # 时区转换 ts.tz_convert(America/New_York) # 处理夏令时 ts pd.date_range(2023-03-12, periods72, freqH, tzAmerica/New_York)6.2 高级时区技巧处理模糊时间夏令时转换期间pd.Timestamp(2023-11-05 01:30:00, tzAmerica/New_York, ambiguousinfer)不存在时间处理夏令时开始pd.Timestamp(2023-03-12 02:30:00, tzAmerica/New_York, nonexistentshift_forward)跨时区聚合df pd.DataFrame({ value: np.random.randn(100), tz: [America/New_York]*50 [Europe/London]*50 }, indexrng) df[utc_time] df.index.tz_localize(df[tz]).tz_convert(UTC)7. 时间序列特征工程7.1 基础时间特征从时间戳中提取特征是时间序列分析的常见预处理步骤df[hour] df.index.hour df[day_of_week] df.index.dayofweek df[is_weekend] df.index.dayofweek 5 df[month] df.index.month df[quarter] df.index.quarter7.2 高级时间特征节假日特征from pandas.tseries.holiday import USFederalHolidayCalendar cal USFederalHolidayCalendar() holidays cal.holidays(startdf.index.min(), enddf.index.max()) df[is_holiday] df.index.isin(holidays)季节特征df[season] (df.index.month % 12 3) // 3 # 1冬, 2春, 3夏, 4秋时间周期特征df[sin_hour] np.sin(2 * np.pi * df.index.hour / 24) df[cos_hour] np.cos(2 * np.pi * df.index.hour / 24)8. 时间序列可视化分析8.1 基础可视化Pandas内置的plot方法可以快速可视化时间序列ts.plot(titleDaily Time Series, figsize(12, 6))8.2 高级可视化技巧滚动统计可视化import matplotlib.pyplot as plt fig, ax plt.subplots(figsize(12, 6)) ts.plot(axax, labelOriginal) ts.rolling(7).mean().plot(axax, label7-day Rolling Mean) ts.rolling(30).mean().plot(axax, label30-day Rolling Mean) ax.legend()季节性子图from pandas.plotting import seasonal_plot seasonal_plot(ts.resample(D).mean(), periodM, freqD)自相关图from pandas.plotting import autocorrelation_plot autocorrelation_plot(ts)9. 时间序列性能优化9.1 数据存储优化处理大规模时间序列数据时存储格式选择很重要使用适当的数据类型ts ts.astype(float32) # 节省内存使用分类类型存储重复时间特征df[day_of_week] df[day_of_week].astype(category)9.2 计算优化技巧使用eval表达式加速计算pd.eval(ts1 ts2 * ts3)避免链式操作# 不推荐 result ts.fillna(0).rolling(5).mean().shift(2) # 推荐 filled ts.fillna(0) rolled filled.rolling(5).mean() result rolled.shift(2)使用numba加速自定义函数from numba import jit jit def custom_ema(series, alpha0.1): result np.empty_like(series) result[0] series[0] for i in range(1, len(series)): result[i] alpha * series[i] (1-alpha) * result[i-1] return result ts.rolling(window10).apply(custom_ema, rawTrue)10. 实际案例分析股票价格分析10.1 数据准备与清洗让我们通过一个股票价格分析的案例来综合应用所学知识# 假设我们已经加载了股票数据 stock pd.read_csv(stock_prices.csv, parse_dates[Date], index_colDate) # 处理缺失值 stock stock.asfreq(B) # 工作日频率 stock stock.ffill() # 前向填充 # 计算日收益率 stock[Return] stock[Close].pct_change()10.2 技术指标计算移动平均线stock[MA_10] stock[Close].rolling(10).mean() stock[MA_50] stock[Close].rolling(50).mean()布林带stock[Rolling_Std] stock[Close].rolling(20).std() stock[Upper_Band] stock[MA_20] 2 * stock[Rolling_Std] stock[Lower_Band] stock[MA_20] - 2 * stock[Rolling_Std]RSI指标delta stock[Close].diff() gain delta.where(delta 0, 0) loss -delta.where(delta 0, 0) avg_gain gain.rolling(14).mean() avg_loss loss.rolling(14).mean() rs avg_gain / avg_loss stock[RSI] 100 - (100 / (1 rs))10.3 交易信号生成基于技术指标生成简单的交易信号stock[Signal] 0 stock.loc[stock[MA_10] stock[MA_50], Signal] 1 stock.loc[stock[MA_10] stock[MA_50], Signal] -111. 时间序列预测基础11.1 特征工程为时间序列预测准备特征# 滞后特征 for lag in [1, 2, 3, 7, 14]: stock[fReturn_lag_{lag}] stock[Return].shift(lag) # 滚动统计特征 stock[Rolling_Mean_7] stock[Return].rolling(7).mean() stock[Rolling_Std_7] stock[Return].rolling(7).std() # 时间特征 stock[Day_of_week] stock.index.dayofweek stock[Month] stock.index.month11.2 简单预测模型使用滞后特征构建简单预测模型from sklearn.linear_model import LinearRegression # 准备数据 data stock.dropna() X data[[Return_lag_1, Return_lag_2, Return_lag_3]] y data[Return] # 训练测试分割 split_date 2022-01-01 X_train, X_test X[X.index split_date], X[X.index split_date] y_train, y_test y[y.index split_date], y[y.index split_date] # 训练模型 model LinearRegression() model.fit(X_train, y_train) # 预测 data[Predicted_Return] model.predict(X)12. 时间序列数据库集成12.1 与数据库交互处理大规模时间序列数据时通常需要与专业数据库交互从InfluxDB读取数据from influxdb import DataFrameClient client DataFrameClient(hostlocalhost, port8086, usernameuser, passwordpass) query SELECT * FROM stock_prices WHERE time now() - 365d result client.query(query) stock result[stock_prices]写入TimescaleDBimport psycopg2 from sqlalchemy import create_engine engine create_engine(postgresql://user:passlocalhost:5432/tsdb) stock.to_sql(stock_prices, engine, if_existsappend, methodmulti)12.2 优化数据库交互批量写入# 分块写入大数据集 for chunk in np.array_split(stock, 10): chunk.to_sql(stock_prices, engine, if_existsappend)使用COPY命令加速import io output io.StringIO() stock.to_csv(output, sep\t, headerFalse, indexFalse) output.seek(0) connection engine.raw_connection() cursor connection.cursor() cursor.copy_from(output, stock_prices, null) connection.commit()13. 时间序列异常检测13.1 统计方法检测标准差法mean, std stock[Return].mean(), stock[Return].std() stock[Is_Outlier] (stock[Return].abs() mean 3*std)滚动Z-Scorerolling_mean stock[Return].rolling(30).mean() rolling_std stock[Return].rolling(30).std() stock[Z_Score] (stock[Return] - rolling_mean) / rolling_std stock[Is_Outlier] stock[Z_Score].abs() 313.2 机器学习方法隔离森林from sklearn.ensemble import IsolationForest clf IsolationForest(contamination0.01) stock[Anomaly_Score] clf.fit_predict(stock[[Return]])LSTM自动编码器from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, RepeatVector, TimeDistributed # 准备序列数据 sequence_length 10 X np.array([stock[Return].values[i:isequence_length] for i in range(len(stock)-sequence_length)]) # 构建模型 model Sequential([ LSTM(32, activationrelu, input_shape(sequence_length, 1)), RepeatVector(sequence_length), LSTM(32, activationrelu, return_sequencesTrue), TimeDistributed(Dense(1)) ]) model.compile(optimizeradam, lossmse) model.fit(X, X, epochs10, batch_size32) # 计算重构误差 predictions model.predict(X) mse np.mean(np.square(X - predictions), axis1) stock[Reconstruction_Error] np.concatenate([np.zeros(sequence_length), mse])14. 时间序列最佳实践14.1 代码组织建议创建时间序列处理工具类class TimeSeriesProcessor: def __init__(self, data): self.data data.copy() def add_time_features(self): self.data[hour] self.data.index.hour self.data[day_of_week] self.data.index.dayofweek return self def add_rolling_features(self, window7): self.data[frolling_mean_{window}] self.data[value].rolling(window).mean() return self def process(self): return self.data processor TimeSeriesProcessor(ts) processed_data processor.add_time_features().add_rolling_features().process()使用装饰器计时import time from functools import wraps def timeit(func): wraps(func) def wrapper(*args, **kwargs): start time.time() result func(*args, **kwargs) end time.time() print(f{func.__name__} executed in {end-start:.4f} seconds) return result return wrapper timeit def process_large_ts(ts): # 复杂处理逻辑 return ts.rolling(1000).mean()14.2 性能优化经验处理大型时间序列时使用pd.read_csv()的chunksize参数分块读取尽早过滤不需要的时间范围使用category类型存储重复的时间特征内存优化技巧# 查看内存使用 ts.memory_usage(deepTrue) # 减少内存使用 ts ts.astype(float32)并行处理from joblib import Parallel, delayed def process_chunk(chunk): return chunk.rolling(30).mean() results Parallel(n_jobs4)( delayed(process_chunk)(ts[i:i1000]) for i in range(0, len(ts), 1000) )15. 常见问题与解决方案15.1 性能问题排查重采样操作缓慢尝试先过滤数据范围再重采样使用closed和label参数减少计算量考虑使用asfreq()替代resample()如果只需要频率转换滚动计算内存不足使用enginenumba参数如果可用分块处理数据使用更简单的聚合函数15.2 数据一致性问题处理时区混淆尽早统一时区使用tz_localize和tz_convert明确时区转换记录原始时区信息处理非均匀时间序列# 计算时间间隔 intervals ts.index.to_series().diff() # 过滤异常间隔 regular_ts ts[intervals pd.Timedelta(1D)]15.3 其他实用技巧快速查看时间序列统计ts.describe(percentiles[0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99])查找时间序列中的突变点changes ts.diff().abs() 3 * ts.diff().std() change_points ts.index[changes]处理多时间序列对齐aligned pd.concat([ts1, ts2, ts3], axis1).dropna()在实际项目中处理时间序列数据时我发现最常遇到的挑战是处理不规则的采样频率和大量的缺失值。一个实用的技巧是在数据清洗阶段就建立完整的时间索引框架然后使用reindex方法将所有数据对齐到这个框架上。这样可以确保后续分析的时序一致性避免因数据缺失导致的错误结论。 
1. Pandas时间序列处理基础时间序列数据是数据分析中最常见的数据类型之一它记录了随时间变化的观测值。在金融、物联网、气象等领域时间序列数据无处不在。Pandas作为Python数据分析的核心库提供了强大的时间序列处理能力。1.1 时间序列数据的特点时间序列数据与普通表格数据最大的区别在于其索引通常是时间戳。这种数据结构带来了几个独特特性时间连续性数据点按固定或可变间隔排列季节性/周期性数据可能呈现周期性变化模式时间相关性相邻时间点的数据往往存在关联在Pandas中我们主要使用Timestamp对象表示单个时间点用DatetimeIndex表示时间索引。创建时间序列数据的基本方法是import pandas as pd # 创建时间范围 date_rng pd.date_range(start1/1/2023, end1/10/2023, freqD) # 创建时间序列数据 time_series pd.Series(range(10), indexdate_rng)1.2 时间序列的创建与转换实际工作中我们经常需要将各种格式的时间数据转换为Pandas可识别的时间序列。常见转换场景包括字符串转时间戳pd.to_datetime([2023-01-01, Jan 2, 2023, 03/01/2023])Unix时间戳转换pd.to_datetime([1672531200, 1672617600], units)处理混合格式pd.to_datetime([20230101, 2023-01-02, Jan 3, 2023], formatmixed)注意当处理大量日期字符串时指定format参数可以显著提高转换速度。例如format%Y%m%d比让Pandas自动推断要快得多。2. 时间序列索引与切片2.1 基础索引操作时间序列数据的索引操作是数据分析的基础。Pandas提供了多种灵活的方式来选择和切片时间序列数据# 选择特定日期 time_series[2023-01-05] # 选择日期范围 time_series[2023-01-03:2023-01-07] # 部分字符串匹配 time_series[2023-01] # 选择整个1月数据2.2 高级索引技巧对于更复杂的时间选择需求Pandas提供了强大的索引功能使用truncate方法截取时间范围time_series.truncate(before2023-01-04, after2023-01-08)使用between_time选择特定时间段适用于日内数据# 假设我们有日内数据 intraday pd.Series( range(24), indexpd.date_range(2023-01-01, periods24, freqH) ) intraday.between_time(9:00, 17:00) # 选择交易时段使用at_time选择特定时间点intraday.at_time(15:00) # 选择每天下午3点的数据3. 时间序列重采样与频率转换3.1 基础重采样操作重采样是时间序列分析中最常用的操作之一它可以将数据从一个频率转换到另一个频率# 创建示例数据 rng pd.date_range(2023-01-01, periods100, freqD) ts pd.Series(np.random.randn(len(rng)), indexrng) # 降采样日数据→月数据 ts.resample(M).mean() # 升采样日数据→小时数据 ts.resample(H).asfreq() # 产生NaN ts.resample(H).ffill() # 前向填充3.2 高级重采样技巧实际应用中我们经常需要更复杂的重采样逻辑自定义聚合函数def first_last(series): return series.iloc[[0, -1]] ts.resample(M).apply(first_last)分组重采样适用于多时间序列df pd.DataFrame({ group: [A]*50 [B]*50, value: np.random.randn(100) }, indexrng) df.groupby(group).resample(M, onindex)[value].mean()使用offset调整重采样锚点# 以每月第25天为锚点 ts.resample(MS).mean() # 默认月初 ts.resample(MS, offset25D).mean()提示处理财务数据时常用的重采样频率包括B工作日W-FRI每周五Q季度末A年末4. 时间序列的移动窗口操作4.1 滚动统计量滚动窗口计算是时间序列分析的核心技术之一用于计算移动平均值、标准差等统计量# 7天滚动平均 ts.rolling(window7).mean() # 带最小观测值要求的滚动计算 ts.rolling(window30, min_periods10).std() # 非对称窗口 ts.rolling(window5D).sum() # 5日历日4.2 扩展窗口与指数加权除了固定窗口Pandas还支持扩展窗口和指数加权移动平均# 扩展窗口累计 ts.expanding().mean() # 指数加权移动平均 ts.ewm(span30).mean() # 半衰期加权 ts.ewm(halflife15).std()4.3 高级窗口类型Pandas 1.3引入了更灵活的窗口类型可变偏移窗口ts.rolling(window5D, min_periods1).mean()自定义权重窗口def custom_weights(window): return np.arange(1, len(window)1) ts.rolling(window5).apply(lambda x: np.sum(custom_weights(x)*x)/sum(custom_weights(x)))事件驱动窗口基于索引距离而非观测次数ts.rolling(windowpd.Timedelta(days5)).mean()5. 时间序列缺失值处理5.1 常见缺失值场景时间序列数据中的缺失值可能由多种原因导致数据采集中断节假日/非交易日传感器故障数据合并时的日期不匹配5.2 缺失值处理方法Pandas提供了多种处理时间序列缺失值的方法插值法ts.interpolate(methodtime) # 按时间权重插值 ts.interpolate(methodspline, order3) # 三次样条插值前向/后向填充ts.ffill() # 前向填充 ts.bfill() # 后向填充复杂填充策略# 使用滚动平均值填充 ts.fillna(ts.rolling(5, min_periods1).mean())注意事项选择填充方法时应考虑数据特性。金融价格数据通常使用前向填充而气象数据可能更适合插值。6. 时区处理与时间转换6.1 时区基础操作处理跨时区数据是时间序列分析中的常见需求# 本地化时区 ts ts.tz_localize(UTC) # 时区转换 ts.tz_convert(America/New_York) # 处理夏令时 ts pd.date_range(2023-03-12, periods72, freqH, tzAmerica/New_York)6.2 高级时区技巧处理模糊时间夏令时转换期间pd.Timestamp(2023-11-05 01:30:00, tzAmerica/New_York, ambiguousinfer)不存在时间处理夏令时开始pd.Timestamp(2023-03-12 02:30:00, tzAmerica/New_York, nonexistentshift_forward)跨时区聚合df pd.DataFrame({ value: np.random.randn(100), tz: [America/New_York]*50 [Europe/London]*50 }, indexrng) df[utc_time] df.index.tz_localize(df[tz]).tz_convert(UTC)7. 时间序列特征工程7.1 基础时间特征从时间戳中提取特征是时间序列分析的常见预处理步骤df[hour] df.index.hour df[day_of_week] df.index.dayofweek df[is_weekend] df.index.dayofweek 5 df[month] df.index.month df[quarter] df.index.quarter7.2 高级时间特征节假日特征from pandas.tseries.holiday import USFederalHolidayCalendar cal USFederalHolidayCalendar() holidays cal.holidays(startdf.index.min(), enddf.index.max()) df[is_holiday] df.index.isin(holidays)季节特征df[season] (df.index.month % 12 3) // 3 # 1冬, 2春, 3夏, 4秋时间周期特征df[sin_hour] np.sin(2 * np.pi * df.index.hour / 24) df[cos_hour] np.cos(2 * np.pi * df.index.hour / 24)8. 时间序列可视化分析8.1 基础可视化Pandas内置的plot方法可以快速可视化时间序列ts.plot(titleDaily Time Series, figsize(12, 6))8.2 高级可视化技巧滚动统计可视化import matplotlib.pyplot as plt fig, ax plt.subplots(figsize(12, 6)) ts.plot(axax, labelOriginal) ts.rolling(7).mean().plot(axax, label7-day Rolling Mean) ts.rolling(30).mean().plot(axax, label30-day Rolling Mean) ax.legend()季节性子图from pandas.plotting import seasonal_plot seasonal_plot(ts.resample(D).mean(), periodM, freqD)自相关图from pandas.plotting import autocorrelation_plot autocorrelation_plot(ts)9. 时间序列性能优化9.1 数据存储优化处理大规模时间序列数据时存储格式选择很重要使用适当的数据类型ts ts.astype(float32) # 节省内存使用分类类型存储重复时间特征df[day_of_week] df[day_of_week].astype(category)9.2 计算优化技巧使用eval表达式加速计算pd.eval(ts1 ts2 * ts3)避免链式操作# 不推荐 result ts.fillna(0).rolling(5).mean().shift(2) # 推荐 filled ts.fillna(0) rolled filled.rolling(5).mean() result rolled.shift(2)使用numba加速自定义函数from numba import jit jit def custom_ema(series, alpha0.1): result np.empty_like(series) result[0] series[0] for i in range(1, len(series)): result[i] alpha * series[i] (1-alpha) * result[i-1] return result ts.rolling(window10).apply(custom_ema, rawTrue)10. 实际案例分析股票价格分析10.1 数据准备与清洗让我们通过一个股票价格分析的案例来综合应用所学知识# 假设我们已经加载了股票数据 stock pd.read_csv(stock_prices.csv, parse_dates[Date], index_colDate) # 处理缺失值 stock stock.asfreq(B) # 工作日频率 stock stock.ffill() # 前向填充 # 计算日收益率 stock[Return] stock[Close].pct_change()10.2 技术指标计算移动平均线stock[MA_10] stock[Close].rolling(10).mean() stock[MA_50] stock[Close].rolling(50).mean()布林带stock[Rolling_Std] stock[Close].rolling(20).std() stock[Upper_Band] stock[MA_20] 2 * stock[Rolling_Std] stock[Lower_Band] stock[MA_20] - 2 * stock[Rolling_Std]RSI指标delta stock[Close].diff() gain delta.where(delta 0, 0) loss -delta.where(delta 0, 0) avg_gain gain.rolling(14).mean() avg_loss loss.rolling(14).mean() rs avg_gain / avg_loss stock[RSI] 100 - (100 / (1 rs))10.3 交易信号生成基于技术指标生成简单的交易信号stock[Signal] 0 stock.loc[stock[MA_10] stock[MA_50], Signal] 1 stock.loc[stock[MA_10] stock[MA_50], Signal] -111. 时间序列预测基础11.1 特征工程为时间序列预测准备特征# 滞后特征 for lag in [1, 2, 3, 7, 14]: stock[fReturn_lag_{lag}] stock[Return].shift(lag) # 滚动统计特征 stock[Rolling_Mean_7] stock[Return].rolling(7).mean() stock[Rolling_Std_7] stock[Return].rolling(7).std() # 时间特征 stock[Day_of_week] stock.index.dayofweek stock[Month] stock.index.month11.2 简单预测模型使用滞后特征构建简单预测模型from sklearn.linear_model import LinearRegression # 准备数据 data stock.dropna() X data[[Return_lag_1, Return_lag_2, Return_lag_3]] y data[Return] # 训练测试分割 split_date 2022-01-01 X_train, X_test X[X.index split_date], X[X.index split_date] y_train, y_test y[y.index split_date], y[y.index split_date] # 训练模型 model LinearRegression() model.fit(X_train, y_train) # 预测 data[Predicted_Return] model.predict(X)12. 时间序列数据库集成12.1 与数据库交互处理大规模时间序列数据时通常需要与专业数据库交互从InfluxDB读取数据from influxdb import DataFrameClient client DataFrameClient(hostlocalhost, port8086, usernameuser, passwordpass) query SELECT * FROM stock_prices WHERE time now() - 365d result client.query(query) stock result[stock_prices]写入TimescaleDBimport psycopg2 from sqlalchemy import create_engine engine create_engine(postgresql://user:passlocalhost:5432/tsdb) stock.to_sql(stock_prices, engine, if_existsappend, methodmulti)12.2 优化数据库交互批量写入# 分块写入大数据集 for chunk in np.array_split(stock, 10): chunk.to_sql(stock_prices, engine, if_existsappend)使用COPY命令加速import io output io.StringIO() stock.to_csv(output, sep\t, headerFalse, indexFalse) output.seek(0) connection engine.raw_connection() cursor connection.cursor() cursor.copy_from(output, stock_prices, null) connection.commit()13. 时间序列异常检测13.1 统计方法检测标准差法mean, std stock[Return].mean(), stock[Return].std() stock[Is_Outlier] (stock[Return].abs() mean 3*std)滚动Z-Scorerolling_mean stock[Return].rolling(30).mean() rolling_std stock[Return].rolling(30).std() stock[Z_Score] (stock[Return] - rolling_mean) / rolling_std stock[Is_Outlier] stock[Z_Score].abs() 313.2 机器学习方法隔离森林from sklearn.ensemble import IsolationForest clf IsolationForest(contamination0.01) stock[Anomaly_Score] clf.fit_predict(stock[[Return]])LSTM自动编码器from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, RepeatVector, TimeDistributed # 准备序列数据 sequence_length 10 X np.array([stock[Return].values[i:isequence_length] for i in range(len(stock)-sequence_length)]) # 构建模型 model Sequential([ LSTM(32, activationrelu, input_shape(sequence_length, 1)), RepeatVector(sequence_length), LSTM(32, activationrelu, return_sequencesTrue), TimeDistributed(Dense(1)) ]) model.compile(optimizeradam, lossmse) model.fit(X, X, epochs10, batch_size32) # 计算重构误差 predictions model.predict(X) mse np.mean(np.square(X - predictions), axis1) stock[Reconstruction_Error] np.concatenate([np.zeros(sequence_length), mse])14. 时间序列最佳实践14.1 代码组织建议创建时间序列处理工具类class TimeSeriesProcessor: def __init__(self, data): self.data data.copy() def add_time_features(self): self.data[hour] self.data.index.hour self.data[day_of_week] self.data.index.dayofweek return self def add_rolling_features(self, window7): self.data[frolling_mean_{window}] self.data[value].rolling(window).mean() return self def process(self): return self.data processor TimeSeriesProcessor(ts) processed_data processor.add_time_features().add_rolling_features().process()使用装饰器计时import time from functools import wraps def timeit(func): wraps(func) def wrapper(*args, **kwargs): start time.time() result func(*args, **kwargs) end time.time() print(f{func.__name__} executed in {end-start:.4f} seconds) return result return wrapper timeit def process_large_ts(ts): # 复杂处理逻辑 return ts.rolling(1000).mean()14.2 性能优化经验处理大型时间序列时使用pd.read_csv()的chunksize参数分块读取尽早过滤不需要的时间范围使用category类型存储重复的时间特征内存优化技巧# 查看内存使用 ts.memory_usage(deepTrue) # 减少内存使用 ts ts.astype(float32)并行处理from joblib import Parallel, delayed def process_chunk(chunk): return chunk.rolling(30).mean() results Parallel(n_jobs4)( delayed(process_chunk)(ts[i:i1000]) for i in range(0, len(ts), 1000) )15. 常见问题与解决方案15.1 性能问题排查重采样操作缓慢尝试先过滤数据范围再重采样使用closed和label参数减少计算量考虑使用asfreq()替代resample()如果只需要频率转换滚动计算内存不足使用enginenumba参数如果可用分块处理数据使用更简单的聚合函数15.2 数据一致性问题处理时区混淆尽早统一时区使用tz_localize和tz_convert明确时区转换记录原始时区信息处理非均匀时间序列# 计算时间间隔 intervals ts.index.to_series().diff() # 过滤异常间隔 regular_ts ts[intervals pd.Timedelta(1D)]15.3 其他实用技巧快速查看时间序列统计ts.describe(percentiles[0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99])查找时间序列中的突变点changes ts.diff().abs() 3 * ts.diff().std() change_points ts.index[changes]处理多时间序列对齐aligned pd.concat([ts1, ts2, ts3], axis1).dropna()在实际项目中处理时间序列数据时我发现最常遇到的挑战是处理不规则的采样频率和大量的缺失值。一个实用的技巧是在数据清洗阶段就建立完整的时间索引框架然后使用reindex方法将所有数据对齐到这个框架上。这样可以确保后续分析的时序一致性避免因数据缺失导致的错误结论。