用电量数据分享：从清洗到接口的完整实践指南

简介面向制造业电力数据分析与时间序列预测场景的LSTM建模资源适合准备学习循环神经网络、电力需求预测或能源管理的初学者与进阶开发者。压缩包为RAR格式共106个文件整体约4.57MB文件类型以59个CSV数据文件为主含多组步长与损失函数的预测结果另搭配7个Python源码、24张结果可视化JPG、2个PTH模型权重、1个XLSX训练损失工作表以及少量XML、PYC、LOG等工程辅助文件。目前已有1522人浏览学习。内容覆盖从原始用电量数据清洗、LSTM模型构建到训练损失跟踪与多组预测结果对比的完整流程便于读者结合源码和权重文件复现实验借助训练损失记录可观察损失下降趋势、判断过拟合点并理解LSTM输入门、遗忘门和输出门在捕捉电力消耗季节性、周期性与趋势性时的作用适合在此基础上调整超参数或迁移到其他时序预测任务。1. 用电量数据分享不止是把报表发给对方那么点事做过能源数据对接的人都有体会“用电量数据分享”听起来就是把读数导出来发过去的事但真正落地时面对的却是不同电表厂商导出的格式各异的 CSV、时间戳时区错乱、同一时刻重复上报、以及对方拿到数据后无法按业务口径查询的尴尬。这个需求要解决的是用电量数据的标准化、隔离与可控交付先把来源不同的数据清洗成统一格式再按接收方权限把数据安全地交出去。适合正在做能源管理平台、数据服务接口或者被“本周对不上账”折磨过的开发者和数据运维人员。2. 用电量数据接入与清洗先把不同来源的 CSV 收拾明白2.1 为什么说清洗是分享的地基用电量数据分享链路里清洗是成本最高的一环也是最容易被低估的一环。电表厂商给的导出文件五花八门有的列名是“正向有功总电能”有的是“Total kWh”有的按 15 分钟一个点导出有的按整点冻结时间字段有带时区的 ISO 格式也有“2024/1/5 08:00”这种字符串。如果不统一后续建表、做接口全是白搭。我一般会先把清洗拆成三个关键动作列名映射、时间标准化、单位统一。这三个动作做完数据才有资格进入存储层。列名映射负责把各厂商的“方言”翻译成内部统一字段时间标准化负责把不同格式的时间戳统一成 UTC避免时区差单位统一负责把电量数值全部折算成 kWh。这三个动作单独看都不难难在它们要能应对持续新增的数据源。每接入一个新厂区就可能多一种导出格式清洗脚本如果写得太死后面每次都要改代码。2.2 一个直接能用的清洗脚本列名映射、时间标准化与去重下面这个脚本用 pandas 实现覆盖最常见的“一表一个 CSV”场景。重点看列名映射和时间处理的方式拿到自己的环境里按实际表头改配置即可。import pandas as pd from pathlib import Path def clean_usage_csv(src_path: Path) - pd.DataFrame: df pd.read_csv(src_path) # 列名映射不同厂商导出格式差异大建议维护在配置表里 # 而不是写死在这段函数里方便新增数据源时只加配置不改逻辑 col_map { 表号: meter_code, Device No: meter_code, 正向有功总电能: kwh_raw, Total kWh: kwh_raw, 冻结时间: reading_time, data_time: reading_time, ts: reading_time, } df.rename(columnscol_map, inplaceTrue) # 单位统一如果原始表里已有 MWh 单位的列单独折算成 kWh # 需要注意表头叫电量(kWh)但数值实际是 MWh 的情况见第 5 章 if 电量(MWh) in df.columns: df[kwh] df[电量(MWh)] * 1000 else: df[kwh] df[kwh_raw] # 时间统一存 UTCutcTrue 会把带时区的时间正确平移 # formatmixed 会自动识别多种格式但速度比指定格式慢 df[reading_time_utc] pd.to_datetime( df[reading_time], utcTrue, formatmixed ) # 剔除关键字段为空的记录时间缺失会影响排序电量缺失无法计算 df.dropna(subset[meter_code, reading_time_utc, kwh], inplaceTrue) # 同一时刻重复上报只保留最后一条 # 常见场景是集中器拉取数据和电表本地冻结数据重复推送 df.sort_values(reading_time_utc, inplaceTrue) df.drop_duplicates( subset[meter_code, reading_time_utc], keeplast ).reset_index(dropTrue) return df[[meter_code, reading_time_utc, kwh]]这段脚本的逻辑说明先把原始列名映射成内部统一的 meter_code、kwh_raw、reading_time接着处理单位差异再用 pandas 的 to_datetime 统一时间格式最后按表计和时间去重。这里有几个参数值得注意。utcTrue 是关键入库前统一到 UTC后续所有接口查询都按 UTC 处理可以避开时区偏移带来的对账问题。formatmixed 会降低解析性能数据量大时可以改成明确的 format 参数比如 %Y-%m-%d %H:%M:%S。drop_duplicates 的 keeplast 表示重复记录保留最后一条因为同一时刻多次上报时通常越晚上报的值越接近最终冻结值。还有一个细节去重后如果发现同一 meter_code 和 reading_time 出现了不同的 kwh 值说明表计或集抄逻辑本身有问题去重只是掩盖问题。这种情况我习惯在日志里记一条 warning方便后续排查采集链路。2.3 清洗完先自检用对账逻辑拦截脏数据清洗脚本跑通不等于数据能分享。我习惯在入库前加两道自检。第一道是连续性检查按 meter_code 和 reading_time 排序后统计每个表计在预期时间间隔内的缺失点数量。比如 15 分钟粒度一天应该有 96 个点如果缺了 10 个以上说明采集链路某一段不稳定。第二道是对账检查把清洗后的数据按日求和再和电表本身的月度冻结读数对比正常误差应在读取时间差对应的电量范围内。如果对不上优先怀疑清洗阶段的单位换算、时区偏移和去重策略而不是怀疑采集系统。这两道自检可以做成脚本放进 CI也可以做成入库前的一个 Python 函数跑不过就阻断入库。另外我还会加一道量级检查住宅表计日电量超过 1000 kWh 的直接标红人工确认这类数据十有八九是单位折算出了问题。3. 用电量数据分享的底层模型多租户隔离与建表细节3.1 多租户隔离分享接口安全的第一道闸门用电量数据的分享场景几乎不会是单一用户。企业内部的多个事业部、不同园区的能源管理平台、供应链上下游的能耗核查方都在同一套系统里订阅数据。每个订阅方只应看到自己拥有权限的数据。多租户有三种常见实现方式独立数据库、独立 Schema、共享表加租户 ID。从成本和运维角度看中小规模项目用共享表加租户 ID 最务实迁移成本低备份恢复简单按 tenant_id 过滤即可。需要注意租户隔离不只要在接口层做过滤存储层的主键、索引和查询语句都必须带上租户维度否则一旦接口某个分支漏了过滤条件就是一次完整的数据泄漏。3.2 最小可用表结构与建表 SQL以模拟项目X为例最简单的分享系统只需要两张表一张保存电表基础信息另一张保存清洗后的冻结读数。-- 电表主数据表 CREATE TABLE meter ( id INTEGER PRIMARY KEY, tenant_id INTEGER NOT NULL, -- 数据归属方所有查询必须带上 meter_code TEXT NOT NULL, -- 物理电表编号如 HN_001 location TEXT, granularity INTEGER NOT NULL DEFAULT 15, -- 冻结周期单位分钟 created_at TEXT NOT NULL DEFAULT (datetime(now)) ); -- 冻结读数表 CREATE TABLE usage_data ( id INTEGER PRIMARY KEY, meter_id INTEGER NOT NULL REFERENCES meter(id), tenant_id INTEGER NOT NULL, -- 冗余 tenant_id避免每次 join reading_time TEXT NOT NULL, -- 统一存 UTCISO8601 字符串 kwh REAL NOT NULL, -- 统一为 kWh source_file TEXT, -- 记录来自哪个文件方便追溯 created_at TEXT NOT NULL DEFAULT (datetime(now)), updated_at TEXT NOT NULL DEFAULT (datetime(now)), UNIQUE (meter_id, reading_time) ); CREATE INDEX idx_usage_tenant_time ON usage_data(tenant_id, reading_time); CREATE INDEX idx_usage_meter_time ON usage_data(meter_id, reading_time);建表有几个细节值得说明。usage_data 里的 tenant_id 看起来冗余实际上是为了让所有查询都能在单表上完成租户过滤避免查 usage_data 还要 join meter 才能拿到租户字段。分享接口面对高频查询减少 join 是值得的。reading_time 统一存 UTC 的 ISO8601 字符串而不是带时区的本地时间这个选择是为了规避跨时区的夏令时问题和不同厂区本地时间制差异。UNIQUE (meter_id, reading_time) 是防止重复导入的最后防线配合 INSERT ON CONFLICT 可以做成幂等写入这一点在第 5 章展开。还要注意 meter_code 与 meter.id 的区别。物理电表编号是业务字符串同一个位置换表后 meter_code 可能改变但 meter.id 不受影响历史数据仍挂在原 meter 上。所以在清洗层我们只认 meter_code入库时再映射到 meter.id。3.3 索引策略分享接口慢八成是这里没做好索引这块值得单独说。常见做法是只给 reading_time 建索引然后查询写成 WHERE tenant_id ? AND reading_time BETWEEN ? AND ?这样只能用到 reading_time 索引再把结果在内存里按租户过滤。数据量到千万行时会明显变慢。正确做法是建立 (tenant_id, reading_time) 复合索引让租户过滤和时段过滤同时命中索引。如果还经常按表计维度查某一天的数据再补 (meter_id, reading_time) 复合索引。索引不是越多越好写入频繁的分享系统要注意索引维护成本二级索引过多会让入库速度肉眼可见地下降。4. 用电量数据分享接口带着鉴权把数据交出去4.1 先定契约两个面向场景的接口设计分享接口不要设计成“能查表就行”要面向接收方的实际使用场景。最常见的两个场景一是按期拉取某块表的原始冻结数据用于绘制曲线二是按月获取某个区域的总用电量用于对账和报表。我一般会定义两个 REST 接口。场景接口关键参数返回粒度明细拉取GET /api/v1/usage/detailtenant_id, meter_id, start, end原始冻结周期15 分钟或 1 小时聚合对账GET /api/v1/usage/summarytenant_id, meter_ids[], start, end, interval按 day / month 聚合细节约定start/end 统一要求传 UTC 时间避免双方对“今天”的定义不同meter_ids 限制单次最大数量比如 100 个防止一次拉爆服务interval 只允许 day 和 month不开放任意小时聚合。返回结构统一成 JSON明细接口的返回长这样{ tenant_id: 1001, meter_id: 1, start: 2024-01-01T00:00:00Z, end: 2024-01-02T00:00:00Z, points: [ {time: 2024-01-01T00:00:00Z, kwh: 12.5} ], total_kwh: 300.5 }4.2 用 FastAPI 实现带租户过滤的分享服务服务端用 FastAPI 比较顺手自带参数校验和 OpenAPI 文档写分享接口比 Flask 少很多样板代码。先看一个最简实现突出租户隔离怎么落地。from datetime import datetime from fastapi import FastAPI, Depends, HTTPException, Query from fastapi.security import APIKeyHeader from sqlite3 import connect app FastAPI() api_key_header APIKeyHeader(nameX-API-Key) def get_db(): conn connect(usage.db) try: yield conn finally: conn.close() def verify_tenant(api_key: str Depends(api_key_header)) - int: # 演示用真实环境应从 token 服务校验并做缓存 token_map { tk_sub_a_1234: 1001, tk_sub_b_5678: 1002, } if api_key not in token_map: raise HTTPException(status_code401, detailinvalid api key) return token_map[api_key] app.get(/api/v1/usage/detail) def usage_detail( tenant_id: int Depends(verify_tenant), meter_id: int Query(...), start: datetime Query(...), end: datetime Query(...), connDepends(get_db) ): # 关键点WHERE 条件里强制带 tenant_id而不是查完再过滤 rows conn.execute( SELECT reading_time, kwh FROM usage_data WHERE tenant_id ? AND meter_id ? AND reading_time BETWEEN ? AND ? ORDER BY reading_time , (tenant_id, meter_id, start.isoformat(), end.isoformat())).fetchall() if not rows: raise HTTPException(status_code404, detailno data) points [{time: r[0], kwh: r[1]} for r in rows] return { tenant_id: tenant_id, meter_id: meter_id, start: start.isoformat(), end: end.isoformat(), points: points, total_kwh: round(sum(r[1] for r in rows), 3) }逻辑说明verify_tenant 依赖注入把 API Key 映射成 tenant_id后续所有查询使用这个变量token_map 只是演示实际生产中应把 Key 存数据库并带过期时间。查询 SQL 里写死 WHERE tenant_id ? 是刻意为之即使后台将来加了缓存或换了 ORM只要 SQL 里不丢这个条件租户隔离就还在。start/end 用 FastAPI 的 datetime 类型自动解析 ISO8601 格式前端传带时区偏移的时间也会被规范化比手写字符串解析省心。这里还有一个隐患代码没有对 meter_id 做归属校验。如果调用方传了一个属于其他租户的 meter_id接口会查出别人的数据。稳妥做法是先把 meter_id 和 tenant_id 一起到 meter 表里确认归属查不到就返回 404。这个坑我在第 5 章单独讲。4.3 Token 签发与撤销分享权限的生命周期管理分享凭证建议用 JWT 而不是自造随机串。JWT 载荷里可以直接携带 tenant_id 和订阅套餐的过期时间接收方不用再调一次查询接口确认权限。签发侧用一个服务端密钥签发import jwt from datetime import datetime, timedelta, timezone SECRET replace-with-a-long-random-secret def issue_share_token(tenant_id: int, days: int 30) - str: payload { sub: str(tenant_id), # 租户 ID 放在标准字段里 scope: usage:read, # 只授权读取用电数据 exp: datetime.now(timezone.utc) timedelta(daysdays) } return jwt.encode(payload, SECRET, algorithmHS256)参数说明sub 字段存 tenant_id 而不是用户名解析 token 时直接 int(payload[sub]) 就能拿到隔离维度。exp 是 JWT 标准过期字段对内系统给 30 天对外部系统按合同周期设置避免一次签发永久有效。JWT 一旦签发在过期前无法单独作废所以发现某个 key 泄漏时最快的手段是更换服务端 SECRET但这会让所有订阅方同时失效。更可控的做法是在 Redis 里维护一个撤销列表解析 token 后先查一下 jti 是否在列表里。对于中小规模项目我更喜欢直接在数据库里存租户状态解析 token 后 join 一张 tenant_status 表判断该租户是否被禁用这条路最简单也最不容易出错。5. 用电量数据分享避坑指南五个最容易翻车的地方5.1 时区混用导致对账差一小时现象不同厂区上报的时间戳有的带 08:00有的是本地裸时间对账时部分时段偏差整整一小时。 原因清洗脚本对时间做 to_datetime 时没有统一 utc 参数入库和接口查询也各用各的时区。 解决从清洗到存储到接口全部使用 UTC展示时再转本地时区。清洗阶段用 pd.to_datetime(..., utcTrue)数据库 reading_time 字段存 UTC 字符串接口 start/end 参数要求 ISO8601 并带时区服务端统一转 UTC 后查询。规则定死比任何技巧都管用。5.2 重复导入让日电量悄悄翻倍现象前一天跑批导入成功第二天运维重跑一次当日电量变成两倍接收方月度对账才发现。 原因导入脚本没有做幂等控制同一 CSV 再次入库时新增了重复行。 解决表结构里加 UNIQUE(meter_id, reading_time)写导入用 INSERT ON CONFLICT DO UPDATE SET kwhexcluded.kwh, updated_atdatetime(now)。第一次实现时容易只看入库成功与否忽略重复行校验加上唯一约束后至少数据不会翻倍重跑只更新时间戳。5.3 单位不统一使结果差一千倍现象有的电表导出文件里电量单位是 MWh有的表头写“电量(kWh)”但数值实际仍是 MWh合表汇总后数值小得离谱或大得离谱。 原因清洗脚本只按表头判断单位偏偏遇上了表头与数值单位不一致的厂商。 解决清洗阶段把单位统一成 kWh除了读表头还要按数值量级做合理性检查。住宅表计日电量超过 1000 kWh、工厂表计日电量低于 1 kWh都标记出来人工确认。单位换算逻辑集中放在清洗函数里不散落在入库脚本各个角落。5.4 只靠应用层过滤租户漏了一个字段就泄漏数据现象某次分享查询返回了别的租户的电量曲线接收方截图反馈才发现接口漏了 tenant 条件。 原因代码在拼接查询条件时漏掉 tenant_id 过滤或者先查 meter 表拿到 meter_id 列表、再查 usage_data 时没有重新校验租户归属。 解决使用 4.2 的强制 WHERE tenant_id ? 写法同时把“查询 meter 归属”变成接口逻辑里的必经步骤先执行 SELECT tenant_id FROM meter WHERE id? AND tenant_id?查不到就返回 404。租户隔离不能依赖某个开发者的临场发挥要在接口逻辑结构上保证它不丢。5.5 接口裸奔没有限流和审计日志现象分享接口上线后订阅方脚本写了死循环每分钟请求上千次数据库连接被打满其他人全部超时。 原因对外分享接口没有限流也没有请求日志异常流量没有任何拦截渠道。 解决在反向代理层对每个 API Key 或租户加 QPS 限制比如每分钟最多 120 次每次请求写一条审计日志至少记录 tenant_id、路径、时间、返回行数。审计日志除了定位问题也是将来和订阅方产生数据争议时对账的依据。6. 把分享做成服务聚合视图、异常检测与可视化技巧6.1 服务端聚合按日汇总在 SQL 里完成分享接口除了返回明细最好提供一个聚合视图。按日汇总的 SQL 可以这样写SELECT tenant_id, meter_id, date(reading_time) AS day, SUM(kwh) AS total_kwh FROM usage_data WHERE tenant_id ? AND meter_id ? AND reading_time BETWEEN ? AND ? GROUP BY tenant_id, meter_id, date(reading_time) ORDER BY day;聚合放在服务端而不是把明细传给客户端去算是为了减少流量和客户端开发成本。注意 date(reading_time) 基于 UTC如果接收方业务上使用东八区日界会差 8 小时。常见做法是存储时把 UTC 时间按接收方时区偏移后再取日期或者提供 asia_shanghai_day 这样的别名列避免两边各算各的。6.2 一个不引入 ML 的异常用电检测分享场景里接收方最常用到的增值功能是异常用电提醒。不需要上机器学习先用一个简单的同比环比双阈值检测THRESHOLD_YOY 1.8 # 同比超过 80% 涨幅一般由生产活动变化导致 THRESHOLD_MOM 1.5 # 环比超过 50% 涨幅常见于表计故障 def detect_anomaly(row, history): # history 是前 4 周同时段的日用电量列表 if not history: return False baseline_yoy sum(history[-7:]) / 7.0 prev_day history[-1] if baseline_yoy 0 and row.total_kwh / baseline_yoy THRESHOLD_YOY: return True if prev_day 0 and row.total_kwh / prev_day THRESHOLD_MOM: return True return FalseTHRESHOLD_YOY 和 THRESHOLD_MOM 是两个需要按场景调整的阈值。数据分享给外部订阅方时建议保守一点阈值调高减少误报内部节能分析时可以调低。历史窗口取 4 周目的是对齐周内生产节律避免连续几天低负荷后再波动被误判。检测逻辑放在聚合查询之后不改数据库结构就能跑。6.3 可视化最后再做的报表层数据分享稳定运行之后接收方大概率会提出“能看个图吗”。强烈建议把可视化放在服务稳定之后再做而不是一开始就搭大屏。简单时序图用 15 分钟粒度的阶梯图比折线图更符合电表冻结特性日电量对比用分组柱状图最直观。我积攒过一次教训可视化做太早指标口径还没定后面接口改了字段报表全部要跟着返工。先把明细、聚合、鉴权这三件事做扎实报表只是套壳。希望帮到你。本文还有配套的精品资源点击获取