简介一套面向制造业用电量时间序列分析与LSTM预测的完整工程资源适合有一定Python基础、希望将深度学习应用于电力需求预测或能源管理场景的学习者。压缩包共106个文件大小约4.57MB主要包含59个csv用电数据与预测结果7个py源码及对应编译缓存还有模型权重pth、训练日志log、训练损失xlsx等文件能覆盖从数据预处理、RNN/LSTM模型构建到训练监控、结果导出的全流程。内容预览显示数据基于Tetuan City用电记录并划分了不同算例与步长2/4/6的MSE、SmoothL1Loss预测对比便于系统观察模型在不同时间步下的表现。目前已有1521人浏览学习此资源不仅提供可直接运行的代码和训练好的模型文件还保留了.idea工程配置与环境依赖方便快速复现迭代。对于研究LSTM在时序预测中的调参、损失曲线分析和误差评估的读者具有明确的参考价值。1. 用电量数据分享到底在解决什么问题它不是导出一张 Excel很多工程师第一次听说“用电量数据分享”以为责任就是把分表的 Excel 汇总好发给财务。真实需求比这大得多园区能源平台要把几百块电表的用能数据送到云端做负荷预测储能和分布式光伏厂商需要连续负荷曲线去做容量配置售电公司和需求响应聚合商则靠准时报数参与调节。用电量数据分享是把电量时序从电表和网关侧可靠地供到下游系统核心不是“能导出来”而是“能持续、按格式、可校验”。这篇笔记从字段定义、数据管道、质量校验讲到接口开放之后的常见坑适合能源数据后台、网关开发以及集成商工程师对照落地。2. 共享什么字段、按什么粒度先把数据口径定死在源端数据分享不是把采集库整库丢出来而是先把“单位、时间、对象”三个口径讲清楚。这三个口径在实际项目中反复变化尤其涉及分时电费的电表时段切分稍有偏差所有下游计算都会错位。因此我建议在源头就做一个公开的数据字典先解决问题需求这一核心问题。2.1 用电量数据分享的最小数据字典表号、计费点、读数和时标一个用于共享的用电量记录至少要包含下面这些字段少一个都会在后续对账时补不回来。字段名类型/单位说明meter_idstring电表资产编号换表后变化billing_pointstring计费点/户号同一个用电位置保持不变collect_timedatetime采样点所属区间的开始时间统一北京时间并注明时区meter_readingnumber/kWh有功总电量累计读数来自电表面板delta_energynumber/kWh本区间增量电量可由相邻读数算出peak/flat/valley_energynumber/kWh分时电量字段视表型可选demand_max_kwnumber/kW区间最大需量选配这里有个关键设计为什么要同时留 meter_id 和 billing_point因为表号是物理属性计费点才是逻辑属性。换表后表号变了但用电主体没变数据应该按 billing_point 连续拼起来。如果后面查询只认表号换表后的历史就断成两截。我一般把 (billing_point, collect_time) 作为唯一键meter_id 只当说明字段。另一个容易犯的错是只共享增量电量 delta_energy不共享累计读数。增量算起来方便但一旦漏掉一个采集点这段增量就永久丢失而且没有任何手段还原。累计读数不同只要前后两个点都在随时能算出任意区间的电量漏点后还能通过前后值插补。所以对外分享我优先给累计读数把 delta_energy 作为可派生字段一起带上各取所长。2.2 实时功率、电量曲线还是日电量15 分钟粒度才是分享主料功率是某一时刻的瞬时值电表显示多少个千瓦只代表那一刻的负载状态。大电机启动瞬间功率冲得很高一分钟后又回来拿功率去做日电量求和误差大得没法看。电量是区间的积分结果可以逐段相加做审计、预测、分时结算都必须用电量。粒度的选择要看下游到底算什么粒度数据点/天适用场景注意15 分钟96光伏储能调度、需量管理、负荷预测必须按整点固定边界跨日时按区间起始时刻归日1 小时24能源审计、月度结算、分时电量对瞬时冲击不敏感日报更稳日级1综合报告、费用分摊做不了负荷预测和需量分析我的经验是凡是要做负荷预测、储能调度、需量优化的直接给 15 分钟曲线。现在多数智能电表本身就按 15 分钟冻结数据电表存储能力足够网关侧只是多一步读取和转发成本并不高。如果接收方只做月度报数给小时级甚至日级就够了。最忌讳的是不在源头定粒度下游每个系统各拉各的最后口径对不上。分时电价场景还要多留一个心眼尖峰平谷四个时段的划分会随政策、季节调整。如果直接分享四段电量而不传递“时段划分规则”等到季节切换下游用旧的时段边界去套新数据结算必然出错。做法是把时段定义作为配置数据和电量一起分发或者干脆只分享总电量和 15 分钟曲线让下游自己切时段。2.3 源端取数的两种常见做法Modbus 轮询与 DL/T 645 抄表常见做法是在电表和共享平台之间加一层采集网关而不是让下游开发者直接对接电表品牌私有协议。网关对内用统一的方式读数对外输出标准 JSON 或 CSV。Modbus RTU 轮询是大多数厂用电表都支持的方式电表提供 RS485 口网关按点位表配置的寄存器地址用 03 功能码读 32 位寄存器。这里最容易踩坑的是各家点位表不统一A 厂电表 40001 可能是 A 相电压B 厂 40001 就是总有功功率寄存器后面跟的倍率也不一样。所以 Modbus 接入一定要做一张点位映射表从电表说明书抄下来在网关里逐条核对不能拿一份配置套所有表。DL/T 645-2007 是国网、南网系统普遍采用的电表通信协议直接读“正向有功总电能”“实时功率”等标准数据项换表或费率时段切换时一致性更好。代价是协议解析比 Modbus 复杂要按对方提供的数据标识编码手册来调试。我一般建议厂区表计品牌多、后续可能换表计的情况下选择 DL/T 645 做主轴Modbus 做补充两者最终都归一成前面那张数据字典入库存对外发布层完全不感知底层是哪种协议。3. 搭建用电量数据分享链路从采集对齐到对外开放的最小实现链路怎么切直接决定后面的排障成本。我一般切成四层采集层负责从电表读数清洗层处理漏采、尖峰、换表拼接存储层只保存统一格式的时序发布层提供 API 或文件下载。很多人跳过清洗直接建接口等数据出问题才回头补那时候已经污染的存量数据要花几倍精力修复。3.1 四层链路怎么切边界采集、清洗、存储、发布边界划分的标准是“某一层坏了其他层能独立重放”。采集层把原始读数写入一个原始主题库带采集时间 source_ts 和到达时间 recv_ts清洗层从原始库读处理完写入主时序库发布层只从主时序库读不再面对脏数据。这样接口出问题时只要查清洗层日志就能定位不用在发布层做各种临时过滤。主时序库的表结构我建议至少包含billing_point、meter_id、slot_start、meter_reading、delta_kwh、quality、source_system。其中 quality 字段特别重要它标记这一行是正常数据、插值数据还是异常标记数据。下游可以根据质量标签决定要不要采纳这一段这比我方在源头上直接删掉给用户一个“看起来很干净但缺了一段”的假象更诚实。3.2 用 FastAPI 做一个按时间范围查询的用电量分享接口发布层最常见的形式是提供一个时序查询接口让下游按计费点和时间范围取数。下面这个实现是生产里能用、不花哨的最小版本关键点在于做了范围限制和参数校验。from fastapi import FastAPI, HTTPException, Query from datetime import datetime, timedelta app FastAPI() # 伪查询函数生产环境里这里连主时序库按 billing_point 和时段扫描 def query_energy(billing_point: str, start: datetime, end: datetime): rows [] # 示例返回结构实际按你的存储实现替换 return rows app.get(/share/energy) def share_energy( billing_point: str Query(..., min_length6, max_length32), start: datetime Query(...), end: datetime Query(...), ): if end start: raise HTTPException(status_code400, detailstart 必须早于 end) if (end - start) timedelta(days31): raise HTTPException(status_code400, detail单次查询最多覆盖 31 天) rows query_energy(billing_point, start, end) total_kwh round(sum(r[delta_kwh] for r in rows), 3) return { billing_point: billing_point, start: start.isoformat(), end: end.isoformat(), total_kwh: total_kwh, points: rows, }这里两个参数值得说明。第一个是 billing_point 的 min_length 校验防止空字符串打进来扫全表。第二个是 31 天的范围上限避免下游一次拉三年数据把连接池拖死。查询结果里同时返回累计读数派生的 delta_kwh 和 total_kwh方便下游直接核对。真实生产环境还要加上鉴权 token 和单位换算说明但接口的骨架就是这些。3.3 大批量历史数据走文件分发CSV 导出、MD5 清单与增量发布时序接口适合小范围取数一次要导几块表一整年数据时同步返回会让请求卡死。我的做法是走文件分发系统异步生成 CSV附带 MD5 校验文件放到约定的共享目录或对象存储下游自己拉取。import hashlib from pathlib import Path def export_csv_share(billing_point: str, start_day: str, end_day: str) - Path: out Path(f/data/share/{billing_point}_{start_day}_{end_day}.csv) # 按天分段查询避免把整月数据一次性塞进内存 with out.open(w) as f: f.write(slot_start,meter_id,meter_reading,delta_kwh,quality\n) for day_offset in range((end_day - start_day).days 1): day start_day timedelta(daysday_offset) rows query_energy(billing_point, day, day timedelta(days1)) for r in rows: f.write(f{r[slot_start]},{r[meter_id]}, f{r[meter_reading]},{r[delta_kwh]},{r[quality]}\n) checksum hashlib.md5(out.read_bytes()).hexdigest() Path(f{out}.md5).write_text(f{checksum} {out.name}\n) return out配套还要生成一个清单文件记录每批文件的起止时间、记录数、计费点列表。下游拉取后先校验 MD5再按清单核对记录数对不上就直接重拉这一批。增量发布的约定要写明文件按天分割还是按周合并晚了多久的数据会被更新进去历史文件是否允许覆盖。这些约定写得越清楚越能减少双方扯皮。4. 用电量数据的质量关口漏采、尖峰、换表必须在源头处理直接拿电表原始读数去分享真实环境下必然出问题。网关断电、电表通信模块死机、换表清零这些事不是会不会发生而是什么时候发生。这里的三个问题处理得越早接口后期的对账就越轻松。4.1 缺失值处理补 0 是最差的习惯要区分“缺数据”和“无电量”我看到过最糟的处理是把缺失时段直接补 0。一个工厂夜班停产凌晨确实不用电这时候补 0 没问题但如果是采集链路断了把缺的 3 小时补成 0下游做日电量汇总时就会少算一大段月底对账才发现。补 0 把“不知道”伪装成了“确定没有”这是数据分享里最危险的行为。正确做法是区分缺口大小连续缺 1 到 2 个点可以用前后有效读数做线性插值缺口超过两个点插值误差太大不如保留缺失并打上质量标记让下游自己决定是否剔除该时段。def fill_missing(rows): # rows: 按 collect_time 升序的 (collect_time, meter_reading_kwh) 列表 result [] for i, (ts, reading) in enumerate(rows): if reading is not None: result.append((ts, reading, normal)) continue # 向前后找最近的有效读数 prev_i, next_i i - 1, i 1 while prev_i 0 and rows[prev_i][1] is None: prev_i - 1 while next_i len(rows) and rows[next_i][1] is None: next_i 1 gap next_i - prev_i - 1 if prev_i 0 and next_i len(rows) and gap 2: prev_ts, prev_r rows[prev_i] next_ts, next_r rows[next_i] weight (ts - prev_ts) / (next_ts - prev_ts) filled prev_r (next_r - prev_r) * weight result.append((ts, round(filled, 3), interpolated)) else: result.append((ts, None, missing)) return result代码里 gap 2 是插值容忍度超过两个点就返回缺失并带 missing 标签。weight 是时间权重按两个有效点之间的时间比例计算插值位置。这里插值只用于补全序列结算场景要提前和接收方讲清楚哪些点是插值出来的最好在接口文档里约定包含插值数据的查询结果默认带 quality 字段由下游决定是否用于结算。4.2 尖峰与负增量的清洗规则先打标记再决定是否剔除电量序列里偶尔会出现一个离谱的点某个 15 分钟增量比前后大十倍下一分钟又恢复正常。这种尖峰可能来自大功率设备启动的真实冲击也可能是采集通道干扰或寄存器瞬态错误。直接删会低估真实最大需量不删下游做异常检测时又会把正常数据误杀。我的经验是分三种情况处理。真实的大功率设备启动尖峰持续多个点或至少呈现平滑上升不会只跳一个点又瞬间回落这种保留。单点孤立突跳且前后值都正常先标记 quality 为 spike不直接删除让下游在可视化时能看见这段异常并从统计中排除。负增量几乎可以断定是换表清零或采集错位同样打标不参与日电量加总。def flag_spikes(deltas, max_delta_threshold): # deltas: 连续区间的电量增量列表max_delta_threshold 是按表计容量折算的上限 flags [] for i in range(1, len(deltas) - 1): left, mid, right deltas[i - 1], deltas[i], deltas[i 1] # 只有“尖峰后立即回落到阈值以下”的单点才标记为毛刺 if mid max_delta_threshold and left max_delta_threshold and right max_delta_threshold: flags.append(spike) else: flags.append(normal) return flags参数 max_delta_threshold 要按每块表的额定容量来定来自互感器倍率和表计规格不能所有表共用一个值。如果暂时拿不到明确容量可以用历史中位数乘以一个系数估算但这种估算只适合打标不建议直接做删除动作。数据分享的价值在于保真主动删数据是最后的手段标记后让下游过滤才是更稳的路径。4.3 换表导致读数清零拼接的关键是“计费点”而不是“表号”换表是电量数据分享里最容易出乱子的操作。旧表拆走时累计读数停在某个值新表装上后从零开始走字。如果系统只按表号存取数据这条计费点的历史就分成两段前后对不上日电量还会出现负值。处理逻辑是保留 billing_point 作为数据主键换表事件单独记录旧表终值、新表初值和换表时间。新表的数据按“基准偏移”换算到旧表的累计序列上这样下游看到的是一个连续的累计曲线。def stitch_after_replace(old_rows, new_rows): # old_rows/new_rows 都是按时间升序的 (collect_time, meter_reading_kwh) if not old_rows or not new_rows: return old_rows new_rows base old_rows[-1][1] # 旧表最后一条累计读数 first_ts, first_reading new_rows[0] stitched [] for ts, reading in new_rows: # 新表从零走字把读数平移到旧表累计基准上 aligned base reading - first_reading stitched.append((ts, aligned, stitched)) return old_rows[:-1] stitched这里有个细节old_rows[:-1] 把旧表最后一条去掉防止和新表第一条在换表时刻附近重复计算。新表第一条读数作为对齐基线后续读数都相对它做偏移。拼接后每行记录还带了 stitched 标记方便对账时回溯。真实工程里换表时间还要结合拆表、装表两条记录来锁定不能只靠系统时间判断。5. 用电量数据分享常见问题排查五个让我熬夜的真实案例下面的问题都是实战里反复出现的每条按现象、原因、解决写清楚希望能给你省下几个加班的夜晚。5.1 时间错位0 点那根数据被配到前一天的尾巴上现象跨日对账时某天的日电量总是比电表分时计量差几度偏差集中在 0 点那一两个点。接口返回的 96 个点里0 点的数据被分到前一天两侧各错一个点日合计对不上。原因网关侧用本地时间记录采集时间到了存储层没有统一成标准时区。有的设备还把 24:00 记成 0:00处理时按“日内最后一点”还是“次日零点”的口径不一致就出现偏移。这是最像玄学的一类问题因为只看单日数据很难察觉往往要跨月才暴露。解决从采集网关开始就统一时间基准报文里的时间戳一律带时区入库前转成北京时间。对已有脏数据按 slot_start 的整点边界重新对齐0 点数据归到当日第一条。做完后加一条校验每天 96 个点必须连续且最后一点落在 23:45缺一个就告警。5.2 重复抄表同一分钟数据到达两次累计电量悄悄翻倍现象某块表某小时的电量统计明显偏高比如平时 60 度当天接口算出来 120 度。对齐采集日志发现网关在同一采集周期内对同一分钟重复上报两次数据库两条记录都保留了下来。原因网关重试机制没有做幂等。第一次上报超时但实际写入重试时又写了一遍或者数据库表没有对 (billing_point, collect_time) 建唯一约束相同的读数被当作新记录插入。解决存储层建唯一约束用 billing_point 加 collect_time 做联合主键写入时用 upsert 而不是 insert。网关侧把重试逻辑改成“重发同样内容服务端按时间戳覆盖”保证重复到达不会产生重复记录。排查时先查同一 slot_start 下的行数大于 1 就是重复。5.3 互感器变比错配月底对账才发现差 10 倍现象一条线路的电量看着合理单日曲线波形也正常但月底和电网账单一比数值差了 10 倍甚至 100 倍。这种错误最坑人因为曲线形状完全一致没有直观的“异常”可以察觉。原因互感器变比没有落到数字链路里。电表二次侧读数本身是经过变比折算的但配置台账里倍率写错或者漏配采集系统拿原始网关注册值直接入库。设备侧改了一次互感器选型数据链路没有同步更新。解决建一张倍率配置表记录每块表的 CT 变比、PT 变比和生效时间。清洗层读配置统一折算成物理量发布层只输出折算后的数值。配置变更要新增一行而不是就地修改保留历史生效区间这样换互感器后还能按时间回溯。对账脚本每周跑一次把系统总量和电费账单做宏观比对偏差超过 2% 就提示查倍率。5.4 表号作为主键换表后同一计费点历史断成两截现象下游反馈某计费点 6 月之前的曲线查不到一查代码查询条件是 meter_id6 月换表后新表号底下只有新数据旧表号底下的历史被孤立了。原因设计表结构时把物理表号当成业务主键忽视了表号是资产编号会随换表改变。用电主体 billing_point 才是跨时间不变的维度表号只是它的一个属性。解决在所有共享接口里用 billing_point 作为查询主键meter_id 作为可选过滤条件。换表时按第四章的拼接逻辑把新表数据平移到旧表累计基准让下游看到完整的连续曲线。如果历史查询已经乱掉做一次数据迁移按计费点合并新旧两段并用换表事件表核对拼接偏移量。5.5 不设查询上限一次拉三年把接口和数据库一起压垮现象上线初期没问题某天合作方写个脚本一次性拉 50 块表三年的数据数据库扫描上千万行接口超时连接池耗尽后面所有正常查询也跟着超时。原因接口没有做能力和体积设计把“查询接口”当成了“数据下载通道”。同步返回大范围数据既占内存又占带宽一个请求就能把服务拖死。解决时序查询接口限制单次最多 31 天、单次最多 50 个计费点超过就走文件分发。文件分发生成 CSV 后异步通知下游通过预签名 URL 拉取不在线请求里同步返回。还要在网关层加每分钟请求数限制防止合作方脚本失控。接口文档里明确写清限制比事后排查要省心得多。6. 开放用电量数据分享接口前拉一张独立对账表再上线接口自测通过不等于数据正确。我最信任的验证方式不是看接口返回码而是直接拿电表屏幕上的读数来对。6.1 随机抽表做人工读数对账再写一个自动完整性检查随机抽三到五块表一个人去电表前记录屏幕上的“正向有功总”累计值另一个人按同一计费点从接口取同一时刻的数据两者计算偏差。允许范围我压在 0.5% 以内超过就要查采集倍率、时间对齐和换表拼接有没有问题。人工对账虽然笨但能发现接口文档里根本没写的隐性错误。对账脚本可以这样写def daily_consistency_check(billing_points, api_base): # 对每个计费点取前一天 96 个点和采集库基准值比对 for bp in billing_points: data api_energy(billing_pointbp, start前一日0:00, end前一日23:59) if data[point_count] ! 96: print(f{bp}: 点数不足当前 {data[point_count]} 个) continue if abs(data[total_kwh] - baseline[bp]) 0.005 * baseline[bp]: print(f{bp}: 总量偏差超过 0.5%)这段逻辑里 point_count 校验查漏点total_kwh 与 baseline 的比值校验查整体偏差。baseline 来自电表侧独立读取或电网账单不能和接口本身同源否则两边错成一样也发现不了。我把这样的检查放进每天早上的定时任务失败时先看对账单再看清洗日志最后回溯采集链路能少走不少弯路。在用电量数据分享这条路上吃过亏之后我自己养成的习惯是接口上线第一周每天手动抄一次总表比任何监控都诚实。数据管道再顺也先问一句“下游算出来对不对”这比代码跑不跑得通更重要。希望帮到你。本文还有配套的精品资源点击获取