3. 数据采集与格式规范：常见数据源、时间列解析、时区处理与统一频率重采样
时序数据预处理的第一步，往往不是算法设计，而是把散落在不同系统中的原始数据“接进来、对齐、洗干净”。实际项目中，数据可能来自业务数据库、第三方API、CSV导出文件，甚至日志文件。这些数据的时间列格式五花八门，时区不统一，采样频率也参差不齐。如果在这一阶段不建立严格的规范，后续的特征工程和模型训练都会建立在“流沙”之上。本章将围绕数据采集的三大核心任务——多源接入、时间语义解析、频率统一——展开，帮助读者构建一套可复用的时序数据接入与格式化方法论。
上图展示了从多源数据接入到标准时序表生成的完整链路。整个过程可以拆解为三个关键阶段：数据源适配、时间语义解析、频率统一。每个阶段都有其特有的技术陷阱和最佳实践，下面逐一深入展开。
3.1 常见数据源接入：CSV、数据库与API的差异化处理
核心原理：不同数据源的访问机制和数据结构差异巨大。CSV文件是静态快照，适合离线批处理；数据库支持增量查询和条件过滤，适合结构化数据的高效抽取；API则提供实时或近实时的数据访问，适合流式采集。接入策略的核心在于：最小化数据传输量、保留原始语义、确保可重放性。
操作流程与判断方法：
- CSV接入：首先检查文件编码（UTF-8无BOM最佳）、分隔符（逗号、制表符或分号）、表头行位置。使用
pd.read_csv()时，建议显式指定parse_dates参数，而不是事后转换。 - 数据库接入：优先使用增量抽取策略，例如通过自增ID或时间戳字段（
updated_at）进行分批拉取。避免一次性全表扫描，尤其是数据量超过千万行时。 - API接入：必须处理限流（Rate Limit）、分页（Pagination）和重试机制。记录每次请求的
cursor或page_，确保断点续传。
| 数据源类型 | 典型场景 | 接入方式 | 常见坑点 |
|---|---|---|---|
| CSV | 历史数据导出、离线分析 | 批量读取 | 日期格式不统一、缺失值用空字符串 |
| 数据库 | 业务系统数据、用户行为日志 | JDBC/ODBC + SQL | 时区隐式转换、索引失效导致慢查询 |
| API | 行情数据、IoT设备上报 | REST/WebSocket | 响应字段嵌套、时间戳为Unix秒/毫秒混用 |
YYYY-MM-DD HH:MM:SS格式并明确标注时区。
3.2 时间列解析：从字符串到标准时间戳的工程化处理
核心原理：时间解析的本质是将人类可读的字符串或数字编码转换为计算机可计算的datetime对象。核心难点在于格式的多样性和歧义性。例如01/02/2023可能是1月2日或2月1日，取决于地区习惯。解析策略应遵循“先规范、后解析、再校验”的原则。
判断方法与操作流程：
- 格式探测：使用
pandas的infer_datetime_format或第三方库dateutil自动识别，但生产环境建议手动指定格式，避免歧义。 - 统一转换：将所有时间列转换为
datetime64[ns]类型，并统一为UTC时区（使用tz_convert('UTC')）。 - 异常值处理：解析失败的值会被置为
NaT，需要统计比例。若超过1%，应回溯数据源检查格式是否混用。
import pandas as pd
# 推荐：显式指定格式，避免歧义
df['timestamp'] = pd.to_datetime(df['raw_time'], format='%Y-%m-%d %H:%M:%S', errors='coerce')
# 时区处理：先本地化再转换
df['timestamp_utc'] = df['timestamp'].dt.tz_localize('Asia/Shanghai').dt.tz_convert('UTC')
# 校验：检查NaT比例
na_ratio = df['timestamp_utc'].isna().mean()
print(f"解析失败比例: {na_ratio:.2%}")3.3 时区处理：消除时间歧义的关键步骤
核心原理：时区处理的核心矛盾在于“同一时刻在不同地区显示不同”。存储时统一使用UTC，展示时按需转换。但实际数据中，时间列可能不带时区信息（naive），或带有UTC偏移量（如+08:00）。处理不当会导致时间序列出现跳变或重复。
操作流程：
- 识别时区来源：检查数据字典或与业务方确认。IoT设备通常使用UTC，国内业务系统常用Asia/Shanghai。
- 统一本地化：对naive时间使用
tz_localize指定时区，对带偏移的时间直接tz_convert。 - 处理夏令时（DST）：如果数据涉及北美或欧洲地区，需注意DST切换日。使用
ambiguous='infer'或nonexistent='shift_forward'参数处理。
tz_localize到各自时区，再统一tz_convert('UTC')，最后按UTC时间排序。
3.4 统一频率重采样：降采样与升采样的工程决策
核心原理：重采样是将原始时间序列映射到目标频率的过程。降采样（如从秒级到分钟级）通过聚合减少数据量，升采样（如从分钟级到秒级）通过插值增加数据密度。选择哪种策略取决于业务需求：监控告警需要高频数据，趋势分析则低频即可。
判断方法与操作流程：
- 确定目标频率：根据业务指标计算周期（如业务峰值检测需要秒级，日报统计需要天级）。
- 降采样：使用
resample()配合聚合函数（mean,sum,ohlc）。注意处理空窗口。 - 升采样：使用
asfreq()或resample().interpolate()。线性插值适用于平滑数据，前向填充适用于状态型数据（如开关状态）。
# 降采样：5分钟K线聚合
df_5min = df.set_index('timestamp').resample('5T').agg({
'price': 'ohlc',
'volume': 'sum'
})
# 升采样：线性插值
df_1s = df.set_index('timestamp').resample('1S').interpolate(method='linear')| 场景 | 推荐方法 | 原因 |
|---|---|---|
| 传感器温度监控 | 降采样取均值 | 减少噪声影响，保留趋势 |
| 交易量统计 | 降采样求和 | 总量语义明确 |
| 设备状态切换 | 升采样前向填充 | 状态保持直到变化 |
| 缺失值较少且平滑 | 升采样线性插值 | 计算简单，效果可接受 |
resample会抛出异常或产生错误聚合。务必先执行sort_index()和groupby(level=0).last()去重。
3.5 案例拆解：多源日志统一接入与清洗
背景与约束：某电商平台需要分析用户行为路径，数据来源包括：前端埋点API（JSON格式，时间戳为Unix毫秒）、后端数据库（MySQL，时间列为本地时间Asia/Shanghai）、离线CSV导出（包含历史订单，时间格式为2023-01-01 12:30:45）。目标：构建一个统一频率（1分钟粒度）的用户活跃度时序表。
初始现象与目标：原始数据分散，时间语义不统一，无法直接进行跨源关联分析。需要清洗后生成标准表，支持后续的漏斗分析和趋势预测。
关键输入与参数：
- API数据：
{"user_id": 123, "event": "click", "ts": 1672531200000} - MySQL数据：
SELECT user_id, action, created_at FROM user_log WHERE created_at >= '2023-01-01' - CSV数据：
order_id, user_id, pay_time，其中pay_time为字符串。
分析判断过程：
- API的
ts为毫秒级Unix时间戳，需除以1000转为秒，再pd.to_datetime(unit='s')。 - MySQL的
created_at为naive时间，需先tz_localize('Asia/Shanghai')再转UTC。 - CSV的
pay_time格式统一，直接指定格式解析。
实际操作步骤：
# 1. 解析API数据
api_df['ts'] = pd.to_datetime(api_df['ts'], unit='ms', utc=True)
# 2. 解析MySQL数据
db_df['created_at'] = pd.to_datetime(db_df['created_at']).dt.tz_localize('Asia/Shanghai').dt.tz_convert('UTC')
# 3. 解析CSV数据
csv_df['pay_time'] = pd.to_datetime(csv_df['pay_time'], format='%Y-%m-%d %H:%M:%S').dt.tz_localize('Asia/Shanghai').dt.tz_convert('UTC')
# 4. 合并所有数据
all_df = pd.concat([api_df.rename(columns={'ts':'time'}), db_df.rename(columns={'created_at':'time'}), csv_df.rename(columns={'pay_time':'time'})])
# 5. 排序去重
all_df = all_df.sort_values('time').drop_duplicates(subset=['user_id', 'time'], keep='last')
# 6. 重采样为1分钟活跃度
active_df = all_df.set_index('time').resample('1T').agg({'user_id': 'nunique'})结果对比与复盘：清洗前，三个数据源的时间范围无法直接比较；清洗后，统一为UTC并重采样，成功生成分钟级活跃度曲线。复盘发现，最初遗漏了API数据中的毫秒单位处理，导致时间偏移约48年，通过数值范围检查（1.6e12）及时修正。此外，MySQL数据存在少量重复记录（同一用户同一秒多次点击），通过drop_duplicates保留最后一条，避免了活跃度虚高。