如何将流量 CSV 数据统一为 GPM，并按 15 分钟采样？

可以用 pandas 按仪表处理数据：先确认读数的含义，再换算为 GPM，最后对齐到统一的 15 分钟时间网格。15 分钟平均流量与每隔 15 分钟的瞬时流量是不同的结果。目标时刻没有实际测量值时，仅靠重采样无法恢复真实的瞬时读数。

先确认数据是流量还是水量

原始描述中的 Cubic feet 和 gallons 都是体积单位，仅凭单位无法确定如何转换成 GPM。需要查看仪表说明或字段定义。

如果读数是区间水量：

平均 GPM = 区间水量（gallons）/ 区间时长（minutes）

如果读数是累计水量：

平均 GPM = (V[i] - V[i-1]) / (t[i] - t[i-1])

其中，V 需要先换算为 gallons，时间差使用分钟。累计表发生清零、回绕或更换时，应单独处理这些情况，不能直接把差值当作正常用量。

明确 gallon 的定义

gallon 有美制和英制之分，确认之前不能混用。以下示例假设目标是美制 GPM：

• 1 ft³ = 7.48051948 US gallons
• ft³/min × 7.48051948 = GPM
• 如果原单位是 ft³/s，还需要乘以 60。

对于单独标记为 gallons 或 Cubic feet 的数据，应先确认它是区间水量还是累计读数，不能直接套用流量换算系数。

瞬时读数可以按 15 分钟时间网格对齐

如果原始数据已经是瞬时流量，可以保留 00:00、00:15、00:30、00:45 等目标时刻的实际读数。目标时刻没有测量值时，先保留缺失值。

下面的示例假设每个 CSV 对应一个仪表，列名为 timestamp、value、unit；时间戳包含时区信息；读数均为瞬时流量；unit 已整理为 GPM 或 ft3/min。实际使用时，需要按文件结构调整这些约定。

from pathlib import Path
import pandas as pd

# 目标单位为美制 GPM，仅适用于已经确认的流量单位。
TO_GPM = {
    "GPM": 1.0,
    "ft3/min": 7.48051948,
}


def normalize_instantaneous_flow(path):
    df = pd.read_csv(
        path,
        usecols=["timestamp", "value", "unit"],
    )
    if df.empty:
        raise ValueError(f"Empty input: {path}")

    df["timestamp"] = pd.to_datetime(
        df["timestamp"], utc=True, errors="raise"
    )
    df["value"] = pd.to_numeric(df["value"], errors="raise")

    if df[["timestamp", "value", "unit"]].isna().any().any():
        raise ValueError(f"Missing required values: {path}")

    factors = df["unit"].map(TO_GPM)
    if factors.isna().any():
        unknown = df.loc[factors.isna(), "unit"].unique().tolist()
        raise ValueError(f"Unsupported units in {path}: {unknown}")

    df["gpm"] = df["value"] * factors
    df = df.set_index("timestamp").sort_index()

    if df.index.has_duplicates:
        raise ValueError(f"Duplicate timestamps: {path}")

    grid = pd.date_range(
        start=df.index.min().ceil("15min"),
        end=df.index.max().floor("15min"),
        freq="15min",
    )

    # 仅保留目标时刻的实际测量值，不推算缺失读数。
    result = df[["gpm"]].reindex(grid)
    result.index.name = "timestamp"
    return result


output_dir = Path("normalized")
output_dir.mkdir(exist_ok=True)

for path in Path("input").glob("*.csv"):
    result = normalize_instantaneous_flow(path)
    result.to_csv(output_dir / path.name)

目标时刻没有原始记录时，代码会留下缺失值。如果时间戳没有时区信息，应先按仪表实际所在时区解释时间，再转换到统一时区，不能直接把本地时间当作 UTC。

如果同一仪表的数据分布在多个文件中，应先合并该仪表的记录，再排序、检查重复时间并对齐，避免遗漏文件边界上的数据。

缺少目标时刻读数时如何处理

可以根据数据用途选择处理方式，并记录结果是否经过估算：

• 保留缺失值：适合要求使用实际测量值的场景。
• 选择最近的观测值：需要规定允许的最大时间偏差，并保留实际测量时间。附近的读数不能被描述为目标时刻的实际测量值。
• 按时间插值：适合允许估算且流量变化相对平缓的场景。应限制可插值的缺口长度，避免跨长时间停机或缺测区间插值。

如果读数实际是区间水量或累计水量，上述方法也不能把它们变成真实的瞬时测量值。

如果需要的是 15 分钟平均流量

对于等间隔、完整采样且可以使用算术平均近似区间平均的瞬时流量序列，可以使用：

average_gpm = df["gpm"].resample(
    "15min", closed="left", label="left"
).mean()

这里的时间标签表示区间起点，例如 10:00 对应 [10:00, 10:15) 的采样均值，它不是 10:00 的瞬时值。

对于不等间隔数据，直接求均值会让采样密集的时段占更大权重。若目标是时间平均流量，需要明确读数在相邻时刻之间如何变化，再据此做时间加权或积分。

对于区间水量，应先按时间重叠关系分配到目标区间，再用该区间总水量除以 15 分钟。跨区间分配通常需要假设原区间内流量恒定；缺失覆盖的部分不能默认为零。

批量处理时的注意事项

75 个文件平均各有 300,000 行，总量约为 2,250 万行。可以先逐文件处理，只读取必要的列，避免一次性合并所有数据。如果使用 chunksize 分块读取，计算累计读数差分时需要保留上一块的末条记录，区间聚合时也需要保留尚未处理完的时间区间。

处理后至少应检查单位映射、时间顺序、重复时间戳、缺失区间，以及累计表是否发生重置。最终采用实际采样、插值还是区间平均，取决于原始读数的含义和后续分析用途。

备注：内容仅供参考。