摘要

批发市场的能源计量系统上线后，真正的挑战往往不是「接不上」，而是「数据不能用」。缺报、跳变、时钟漂移、量纲不一致，这些问题会让对账结果失去参考价值。本文从数据治理的角度，给出工业计量场景下数据清洗、多级对账与异常用电检测的工程实现方案。

一、问题定义：三份对不上的报表

批发市场的能源数据通常有三份来源：

外部账单：供电局、供水公司出具的月度费用单总表数据：市场总进线处的计量表分户数据：各档口、各品类区的分表

理想状态下，三份数据应该能互相印证。实际情况是，三份数据经常各说各话。原因通常不是设备故障，而是数据质量问题：

表计离线导致某个时段无数据，被当成 0 计入通信中断后补报，时间戳错位累计值出现跳变（表计复位、换表未清零）不同厂家、不同批次的表计上报字段含义不一致

在没有数据治理的情况下，用这样的数据做对账，得到的结论毫无意义。下面按数据流水线的顺序，逐个环节说明处理方式。

二、接入层：先把数据的「身份」说清楚

数据治理的第一步不是清洗，而是统一身份。

2.1 设备编码的统一

同一个物理表计，在不同环节可能有多套编码（厂家序列号、安装位置编号、资产编号）。如果接入层不做统一映射，后续所有关联分析都会失败。

建议在接入时建立一张映射表：

CREATE TABLE dim_device_mapping (
device_id VARCHAR(64) NOT NULL COMMENT '平台设备ID（主键）',
vendor_sn VARCHAR(64) COMMENT '厂家序列号',
asset_code VARCHAR(64) COMMENT '资产编码',
market_id VARCHAR(32) NOT NULL COMMENT '市场编码',
meter_role VARCHAR(16) NOT NULL COMMENT 'total/zone/user',
zone_code VARCHAR(32) COMMENT '所属分区',
parent_device_id VARCHAR(64) COMMENT '上级设备ID（用于层级对账）',
install_date DATE COMMENT '安装日期',
net_ratio DECIMAL(10,2) COMMENT '互感器倍率，直通表为1',
PRIMARY KEY (device_id),
UNIQUE KEY uk_vendor_sn (vendor_sn)
) COMMENT='设备维度映射表';

net_ratio 这个字段容易被忽略。 互感器接入式表计的读数需要乘以倍率才是实际用电量。如果漏掉这一步，一台 400/5 互感器的表计，数据会差 80 倍，而这种错误在上线初期很常见。

2.2 量纲与精度统一

表计的原始数据可能有不同单位（kWh / Wh，kW / W）、不同小数位。接入层应统一归一到标准量纲，并把精度策略固定下来。精度不要在每一层各自处理——导入时归一，之后一律按统一精度计算，避免多次四舍五入带来的累积误差。

三、数据清洗：四类典型脏数据

3.1 缺报

缺报是最常见的。处理方式不能简单地用 0 填充——那会直接压低用量，造成分区合计偏低的假象。正确做法是标记为缺失，并决定是否插值：

from datetime import datetime, timedelta

def resample_and_flag(rows, interval_minutes=60):
"""
把不规则的时间序列重采样到固定间隔，并标记缺失点。
rows: [{'ts': datetime, 'value': float}, ...] 已按时间升序
返回: [{'ts':..., 'value':..., 'is_missing': bool}, ...]
"""
if not rows:
return []

step = timedelta(minutes=interval_minutes)
by_ts = {r['ts']: r['value'] for r in rows}

start, end = rows[0]['ts'], rows[-1]['ts']
out, cursor = [], start
prev_value = None

while cursor <= end:
if cursor in by_ts:
value = by_ts[cursor]
out.append({'ts': cursor, 'value': value, 'is_missing': False})
prev_value = value
else:

缺失点：累计值用前值占位，is_missing=True 供下游排除

out.append({'ts': cursor, 'value': prev_value, 'is_missing': True})
cursor += step

return out

关键点：缺失点用前值占位并打标，而不是补 0。

对于累计值（单调递增的读数），用前值占位意味着这一段的差分为 0，即「该时段用量不可知」。配合 is_missing 标记，下游对账时可以把包含缺失点的时段整体排除，而不是得到一个偏低的错误结论。

对于瞬时量（功率），缺失点通常直接标记为无效，不做插值。

3.2 跳变

累计电能是单调递增的，出现回落必然异常。成因包括表计复位、换表未清零、通信报文损坏。检测逻辑：

def detect_jumps(series, max_back_ratio=-0.001):
"""
检测累计值序列中的异常跳变。
series: [{'ts':..., 'value':...}, ...]
max_back_ratio: 允许的最大回落比例（负值，接近 0）
"""
anomalies = []
for i in range(1, len(series)):
prev, cur = series[i - 1], series[i]
if prev['value'] == 0:
continue
delta = cur['value'] - prev['value']
ratio = delta / prev['value']
if ratio < max_back_ratio:
anomalies.append({
'ts': cur['ts'],
'prev': prev['value'],
'cur': cur['value'],
'delta': round(delta, 3),
'type': 'reset_or_corrupt'
})
return anomalies

检出后要区分处理：换表（有工单记录的，重设基线）、表计复位（需人工确认）、报文损坏（丢弃该点，等待重报）。不要在清洗层自动「修正」跳变——那会掩盖真实故障。

3.3 时钟漂移

表计本地时钟和云端时间不一致，会让跨设备对账错位。处理方式有两种：一是接入层统一采用平台侧接收时间戳，保留设备原始时间戳备查；二是定期下发校时。前者实现简单，推荐作为默认策略。

3.4 重复上报

通信重传会导致同一时间点出现多条记录。去重规则要明确：同一 device_id + 同一时间戳，保留最后一条（或第一条，需固定策略，不能随机）。

四、多级对账：从「差额」到「可归因的差额」

数据清洗完成后，对账才有意义。批发市场的对账比单一市场场景更复杂，因为品类区之间的用能特征差异极大。

4.1 对账的层级关系

总表
├── 品类区表 A（水产区）
│ ├── 档口表 A-001
│ └── 档口表 A-002 ...
├── 品类区表 B（冻品区）
└── 品类区表 C（干货区）

对账要逐层做，而不是只做顶层：

总表用量 ≈ Σ(品类区表用量) + 该层合理损耗
品类区表用量 ≈ Σ(该区档口表用量) + 该区合理损耗

逐层做的好处是归因收敛。 如果总表与品类区合计对不上，但各品类区内部都对得上，问题就锁定在主干管网；如果某个品类区内部对不上，问题就在这个片区的公共部分。

4.2 容差要分层标定，不能一刀切

这一点在批发市场尤其重要。冻品区的管网结构与干货区不同，损耗比例也不同。用统一的容差会出现「冻品区正常却被判异常、干货区异常却被放过」的情况。

建议按分区历史数据标定各自的基线容差，并定期复核。

五、异常用电检测：从阈值到模式

基础阈值告警（超容、低压、离线）只能捕获显性问题。批发市场更值得关注的是模式异常。

5.1 夜间基线

闭市时段应当接近零负荷。下面这段逻辑用于找出「夜间不该有电却在用电」的回路：

def night_baseline_check(series, night_hours=(1, 2, 3, 4, 5), ratio_threshold=0.15):
"""
series: [{'ts': datetime, 'usage': float}, ...] 小时级用量
返回夜间用量占全天比例异常的时间点
"""
from collections import defaultdict

daily = defaultdict(float)
daily_night = defaultdict(float)
for r in series:
    d = r['ts'].date()
    daily[d] += r['usage']
    if r['ts'].hour in night_hours:
        daily_night[d] += r['usage']

out = []
for d, total in daily.items():
    if total <= 0:
        continue
    night_ratio = daily_night[d] / total
    if night_ratio > ratio_threshold:
        out.append({'date': d, 'night_ratio': round(night_ratio, 4),
                    'night_usage': round(daily_night[d], 2)})
return out

对冻品区这类需要 24 小时制冷的场景，夜间基线不适用，规则要按品类分别配置——这也是为什么层级结构必须先建对，否则规则没法按品类生效。

5.2 趋势偏离

同一回路在环境条件没有明显变化时，用量曲线持续抬升，通常指向设备效率衰减（保温层老化、门封失效、制冷剂不足）。做法是用同期对比（同比、环比）而不是绝对值：

-- 按分区计算同比用量变化率，识别效率衰减趋势
SELECT
zone_code,
DATE_FORMAT(ts, '%Y-%m') AS ym,
SUM(usage) AS month_usage,
LAG(SUM(usage), 12) OVER (
PARTITION BY zone_code
ORDER BY DATE_FORMAT(ts, '%Y-%m')
) AS same_month_last_year
FROM fact_meter_usage
WHERE meter_role = 'zone'
GROUP BY zone_code, ym
HAVING same_month_last_year IS NOT NULL
AND month_usage / same_month_last_year > 1.15;

这条查询会挑出「比去年同期高出 15% 以上」的品类区。这类分区值得去看一看设备状况——趋势偏离是设备劣化最早期的信号，比等到故障停机再处理成本低得多。

5.3 曲线形状识别

冷库类负荷有明显的周期性（压缩机启停）。如果曲线的启停频率突然变密，可能指向温度设定过窄或制冷剂泄漏。这类检测需要把原始曲线保留下来做形状分析，因此原始数据不能只做聚合就丢弃。

六、存储与查询：几个工程取舍

原始层与聚合层分表。 原始上报数据用于追溯与审计，保留期可以较短；小时级、日级聚合表用于对账与报表，长期保留。两者分开可以让查询效率差出数量级。

对账结果要落固化成表。 每次对账都实时计算是浪费——对账结果一旦生成就应该落表，附带对账口径版本号。这样历史争议可以追溯「当时是按哪个口径算的」。

分区裁剪。 按 market_id + 时间范围分区是基本要求。批发市场单场的计量点常达数千个，小时级数据量不小，不做分区查询会很难受。

七、小结

批发市场能源计量的工程难度，主要不在「把表接上云」，而在「让云上的数据可用」。三条经验：

先统一身份，再谈清洗。 设备映射、量纲、精度这些「元数据」问题不解决，后面每一步都在错的基础上算。
缺失要打标，不要补零。 补零会制造「用量偏低」的假象，比缺失本身更危险。
对账要逐层做，容差要分层标定。 只有逐层对账，差额才能从标量变成可归因的结构化信息。

数据治理的投入不产生直接产出，但它决定了这套系统最终是「能看数」还是「能用数」。

（本文方法在批发市场能耗管理项目中实践，涵盖总表—品类区—档口三层对账与夜间基线、同比偏离检测；三层结构对应 3T-UEM 统一能源计量管理模型。相关项目由合众致达实施。）