用电数据安全分享实践：从分级脱敏到差分隐私

简介：面向制造业用电量时序预测场景的LSTM实践项目，适合希望掌握循环神经网络在电力负荷数据分析中应用的数据科学学习者与能源管理研究者。资源包共106个文件，约4.57MB，包含59个csv用电量数据与预测结果、7个Python建模脚本、2个pth模型权重、1个xlsx训练损失记录，配以jpg可视化图表、xml配置和log日志等，完整覆盖从数据预处理到模型训练、评估与预测的流程，方便对比不同步长下的效果。已有1524人学习下载。通过梳理main脚本与训练损失曲线，可理解LSTM门机制、时间步长和损失函数对预测效果的影响；不同步长csv预测结果便于对比时序建模精度；.idea与iml文件提供可复现的开发环境配置。整体结构清晰，既可作为课程设计参考，也能为实际电力需求预测任务提供可直接改用的代码与调参思路。

1. 用电量数据分享：从“拷个表发过去”到“敢分享才叫能力”

用电量数据分享这几年已经完全变了样：电表每 15 分钟采集一条冻结曲线，叠加户号、地址、容量和行业标签之后，能精确还原出一个企业的开工率、一个家庭的作息规律，甚至能推断屋里有几个人住。很多团队以为把户号和姓名抹掉就安全了，结果分享出去的数据包被合作方拿去做关联分析，用电峰谷特征一匹配，几分钟就把匿名用户重新识别出来。想把这个方向做成稳定能力，不是靠一张脱敏表，而是靠一整套从分级、授权、脱敏到审计的管线设计。这篇笔记面向正要搭数据产品、做征信风控或能效服务的工程师，讲清楚一套能复现的落地路径。

2. 给用电数据分级分类：先知道手里有什么，再谈怎么分享

2.1 分享出去的数据长什么样：把字段清单拉全再谈保护

做用电量数据分享的第一件事，不是写脱敏函数，而是盘点数据形态。电网侧的用电数据通常按采集频率和数据域分成几类，我一般会先让团队把字段清单拉全，再逐字段标注敏感度。

这里面最容易翻车的是原始冻结数据。单看一条“95号表在18:00有功电量0.52”没有任何意义，但连续 96 点拼接成负荷曲线后，它就是一台设备的指纹。某地做过一次测试，把匿名后的负荷曲线和小区变压器日负荷做比对，峰值时段的抬升特征能把大功率用电户锁定到楼栋甚至楼层。所以分级分类不能只看“有没有户号”，还要看“组合之后能不能定位到一个人或一个经营主体”。

2.2 行业通行的数据分级模型：L0 到 L4 的五级划分

行业内做用电数据分级，普遍采用五级模型，跟很多企业数据安全治理的分级方式兼容。我把它简化成下面这张表，落地时可以直接作为评审基线。

很多团队把等级定到字段上就完事了，这是典型误区。L2 和 L3 的界限是动态的：单独看行业分类是 L1，但当某园区只有一家从事电镀的企业，行业+地区就把它变成 L3。所以分级必须有上下文判断环节。我在实际评审里会要求数据负责人回答一个问题：这条数据如果被竞争对手拿到，配合公开信息能不能说出“这是哪家企业”？能说出来的，至少 L3。

2.3 字段级脱敏清单：动手前先过一遍这张表

定完等级之后，要逐字段写脱敏手法。下面这张表是我在多个项目里沉淀出来的检查清单，适合在评审会上直接过。

这份清单的执行原则只有一条：能不放就不放，必须放就做不可逆变换，可逆加密不等于脱敏。真实项目里常见有人用 AES 加密户号当脱敏，这是给自己埋雷——加密密钥一旦和数据集一起交付，等于没脱敏。下一章我会给一套可以照着跑的脚本，把这套清单变成代码。

3. 分享管线落地：用一套 Python 脚本跑通脱敏、授权、审计

3.1 第一步：生成一份模拟用电数据，让整个流程可复现

没有真实数据的时候，我先造一份模拟数据，保证脚本能完整跑通。模拟数据要尽量接近真实形态：每个用户一条档案、每天 96 点负荷曲线、行业分类和容量字段。

import pandas as pd import numpy as np from datetime import datetime, timedelta # 固定随机种子，保证结果可复现 np.random.seed(42) users = [] for i in range(2000): # 每户一个档案：户号、用电地址前缀、行业、容量、年用电量 users.append({ "user_id": f"U{100000 + i}", "region": np.random.choice(["A区", "B区", "C区", "D区"], p=[0.4, 0.3, 0.2, 0.1]), "industry": np.random.choice(["制造业", "餐饮", "零售", "教育", "医疗"], p=[0.3, 0.25, 0.2, 0.15, 0.1]), "capacity_kva": np.random.lognormal(mean=3.5, sigma=0.6, size=1)[0].round(1), "annual_kwh": 0, }) df_user = pd.DataFrame(users) # 生成 96 点负荷曲线，用高斯过程近似真实用电波动 date_range = pd.date_range("2025-01-01", periods=30, freq="D") curves = [] for _, row in df_user.iterrows(): base = np.random.lognormal(mean=0.8, sigma=0.4) for ts in date_range: hour = ts.hour + ts.minute / 60 # 白天高、夜间低的负荷形状，叠加随机噪声和用户差异 shape = 0.5 + 0.5 * np.exp(-((hour - 14) ** 2) / 20) noise = np.random.normal(0, 0.05) power = max(0.1, base * shape * (1 + noise)) curves.append({ "user_id": row["user_id"], "timestamp": ts, "power_kw": round(power, 3), }) df_curve = pd.DataFrame(curves) print(f"档案数: {len(df_user)}, 曲线条数: {len(df_curve)}") print(df_user.head(3))

这段代码的关键是随机种子和负荷形状函数。固定np.random.seed(42)是为了让每次跑出来的数据一致，后面验证脱敏效果时不会因为数据变了而误判。shape函数模拟“白天高、夜间低”的典型日负荷形态，这是为了让后面的匿名化和噪声测试有一个真实的分析背景。实际接数时，把df_user和df_curve替换成从数仓读出来的真实表即可。

3.2 第二步：户号与准标识符的不可逆匿名化

第一步先把直接标识符处理掉。户号不能只做普通哈希，因为原始户号空间有限，攻击者可以枚举全部可能值做彩虹表。正确做法是加盐的 HMAC，并且每个分享批次换一个新盐值。

import hashlib import hmac import secrets def anonymize_id(user_id: str, salt: bytes) -> str: """ 户号匿名化：HMAC-SHA256 加盐哈希，截断到 20 位十六进制。 盐值由 secrets.token_bytes 生成，每个批次独立。 """ digest = hmac.new(salt, user_id.encode("utf-8"), hashlib.sha256).hexdigest() return digest[:20] # 每个分享批次生成一次新盐值，避免跨批次关联 salt = secrets.token_bytes(32) df_user["anon_id"] = df_user["user_id"].apply(lambda x: anonymize_id(x, salt)) df_curve["anon_id"] = df_curve["user_id"].apply(lambda x: anonymize_id(x, salt)) # 删除原始标识符 df_user.drop(columns=["user_id"], inplace=True) df_curve.drop(columns=["user_id"], inplace=True) print(df_user.head(3))

为什么用 HMAC 而不是直接把盐拼在字符串后面做sha256？HMAC 在密钥处理上有固定填充流程，能避免长短盐值带来的长度扩展问题，安全性有保障。截断到 20 位十六进制（80 bit）对于 2000 个用户的规模足够，但如果用户量到千万级，建议保留 32 位完整输出，避免生日碰撞。这里的salt要由密钥管理系统保存，落到代码里只是演示，生产环境不允许硬编码。

接下来处理准标识符。地区字段只保留到区级，行业做大类合并，容量做桶化。这一步的目标是让任何一条记录都不能通过“地区+行业+容量”组合被唯一锁定。

def generalize_region(region: str) -> str: # 只保留区级，丢弃更细的街道信息 return region.split("-")[0] if "-" in region else region def generalize_industry(industry: str) -> str: # 行业合并：餐饮和零售合并为"商业服务业" if industry in ["餐饮", "零售"]: return "商业服务业" if industry in ["教育", "医疗"]: return "公共服务" return industry def bucketize_capacity(cap: float) -> str: # 容量桶化：按对数间隔分桶，桶内样本数至少为 k thresholds = [0, 50, 100, 200, 500, 1000, float("inf")] labels = ["0-50", "50-100", "100-200", "200-500", "500-1000", "1000+"] for th, label in zip(thresholds[:-1], labels): if thresholds[thresholds.index(th) + 1] is float("inf"): if cap >= th: return label elif th <= cap < thresholds[thresholds.index(th) + 1]: return label return "unknown"

泛化逻辑看着简单，但桶化函数里藏了一个关键点：分桶边界必须和业务含义对齐。容量 100kVA 在电费计价上是一个台阶，把它拆到 50-100 和 100-200 两个桶，业务上解释得通；如果按分位数切桶，虽然统计上均衡，但合作方拿到数据后无法把它映射回电价体系，数据效用会打折。所以我一般先问业务方“哪些阈值有实际意义”，再定分桶。

3.3 第三步：k-匿名检查，防止组合定位到具体用户

泛化完必须验证每个等价类的样本数。k-匿名的最低要求是任意一组准标识符组合下至少有 k 个用户，通常取 k=5 或 k=10。下面的代码构建准标识符组合并统计最小类大小。

def check_k_anonymity(df, quasi_identifiers: list[str], k: int) -> bool: """ 检查给定准标识符组合的 k-匿名性。 返回是否满足以及最小的等价类大小。 """ grouped = df.groupby(quasi_identifiers).size().reset_index(name="count") min_count = grouped["count"].min() return min_count >= k, min_count df_user["capacity_bucket"] = df_user["capacity_kva"].apply(bucketize_capacity) df_user["industry_general"] = df_user["industry"].apply(generalize_industry) df_user["region_general"] = df_user["region"].apply(generalize_region) quasi = ["region_general", "industry_general", "capacity_bucket"] ok, min_count = check_k_anonymity(df_user, quasi, k=10) print(f"k-匿名检查通过: {ok}, 最小等价类: {min_count}") # 如果不满足，需要进一步合并稀有类别 if not ok: # 常见做法：把出现次数少的行业合并为"其他" small_industries = df_user["industry_general"].value_counts() small_industries = small_industries[small_industries < 10].index.tolist() df_user["industry_general"] = df_user["industry_general"].replace(small_industries, "其他")

这里有一个非常容易踩的坑：只检查全量组合的 k-匿名，但没检查“去掉容量桶只剩地区和行业”的组合。攻击者不需要知道容量，他可能从公开的环评报告里拿到准确容量值，所以必须对多个准标识符子集都做检查。我的做法是把quasi的幂集全部跑一遍，找出其中最脆弱的组合。这个脚本量不大，但能把“看起来满足、实际能被拆穿”的情况抓出来。

3.4 第四步：分享包生成与审计日志

脱敏完成之后，数据的“分享出口”需要一个统一通道。我把它做成一个build_share_package函数，同时产出数据文件、授权 token 和审计日志，三样东西分别落库，任何一次数据交付都可追溯。

import json import secrets from datetime import datetime, timezone def build_share_package(df: pd.DataFrame, request_id: str, expire_days: int = 30): """ 生成分享包：数据文件 + 授权 token + 审计日志。 - df: 脱敏后的数据 - request_id: 外部请求方提交的申请编号 - expire_days: token 有效期 """ # 1. 生成一次性授权 token，只存哈希 token = secrets.token_urlsafe(32) token_hash = hashlib.sha256(token.encode()).hexdigest() # 2. 准备分享包元信息 package = { "request_id": request_id, "expire_at": (datetime.now(timezone.utc).timestamp() + expire_days * 86400), "rows": len(df), "columns": list(df.columns), "checksum": hashlib.sha256(pd.util.hash_pandas_object(df, index=True).values).hexdigest(), } # 3. 写审计日志：记录是谁、什么时候、取走了哪些字段 audit_log = { "request_id": request_id, "token_hash": token_hash, "access_time": datetime.now(timezone.utc).isoformat(), "fields": list(df.columns), "row_count": len(df), "purpose": "外部合作方征信分析", } with open("audit.jsonl", "a", encoding="utf-8") as f: f.write(json.dumps(audit_log, ensure_ascii=False) + "\n") # 4. 数据落盘 df.to_parquet(f"share_{request_id}.parquet", index=False) # 5. 把 token 单独返回给调用方，不走文件 return token, package token, meta = build_share_package(df_user, req_id := "REQ-2025-001") print(f"分享包生成完成，token 有效期到: {datetime.fromtimestamp(meta['expire_at'])}")

几个参数值得细说。expire_days=30不是拍脑袋定的，它要和合作方的数据使用周期匹配：如果对方是按月结算，30 天刚好一个账期；如果对方做的是年度能效报告，token 要给 90 天但数据内容要按季度更新。checksum用pandas.util.hash_pandas_object，它把每一行内容做哈希再汇总，比单纯对整个文件做 MD5 更灵敏——哪怕只有一行被改动，校验值都会变，方便事后定位是哪个批次、哪个字段出了问题。

审计日志记录的是 token 的哈希而不是 token 本身，这是安全常识：审计文件长期留存，一旦泄露，不应把有效凭证也暴露出去。我见过有人把明文 token 写进日志，结果日志文件被拖库后，所有历史分享包都能被外部访问，那是一次非常被动的翻车经历。

3.5 差分隐私：给汇总接口加一层噪声

除了明细数据的脱敏分享，常见的还有一类“按条件汇总”的查询接口，比如“返回某园区过去 30 天平均日负荷”。这类接口如果直接返回真实数值，反复查询就能差分出单个用户的电量。标准做法是在查询结果上加拉普拉斯噪声。

def laplace_mechanism(value: float, epsilon: float, sensitivity: float = 1.0) -> float: """ 拉普拉斯机制：返回加噪后的值。 epsilon 越小，隐私保护越强，数据精度越低。 sensitivity 是查询函数的最大敏感度，通常取单用户最大电量。 """ scale = sensitivity / epsilon noise = np.random.laplace(0, scale) return max(0.1, value + noise) # 示例：查询某行业日用电总量，ε 取 0.5 industry_daily = df_curve.merge(df_user[["anon_id", "industry_general"]], on="anon_id") total = industry_daily[industry_daily["industry_general"] == "制造业"]["power_kw"].sum() reported = laplace_mechanism(total, epsilon=0.5, sensitivity=1000) print(f"真实值: {total:.2f}, 发布值: {reported:.2f}")

这里的epsilon=0.5是隐私预算，不是噪声幅度。它表示保护强度，每个查询都会消耗预算，累计超过阈值后就不能再发布该用户的任何相关查询。真实部署时，我会把 epsilon 预算做成一个计数器，存 Redis，所有查询都要检查剩余预算。这里不展开，但千万别漏掉。

4. 分享出去之后怎么控得住：数据沙箱与联邦学习的取舍

4.1 数据沙箱：让模型进来，不让数据出去

明细数据哪怕脱敏后，依然有组合攻击风险。对 L3 及以上数据，我一般不建议直接交付文件，而是把合作方的分析作业放进沙箱。沙箱的思路是：数据留在我的环境里，合作方提交代码进来跑，只允许带走计算结果，且计算结果要经过差分隐私审查才能出去。

沙箱方案的关键参数是三个：数据读写权限、出网白名单、作业审计。一个最小可用的沙箱配置长这样：

sandbox: data_mount: read_only allowed_egress: ["result_bucket"] network: blocked compute_limit: cpu_cores: 4 memory_gb: 16 max_runtime_min: 180 audit: log_every_api_call: true preserve_job_output: 30d

data_mount: read_only是硬底线，很多泄露事故是数据目录可写，合作方直接把原始数据拷到结果目录带走。network: blocked保证数据不能通过外传通道泄露。max_runtime_min: 180用于防止恶意长任务反复逼近原始数据。我在生产环境里还会加一个检查点：作业输出的每一行结果都必须通过规则引擎检查，字段维度不能超过授权范围。

4.2 纵向联邦：不同机构联合建模但数据不出域

如果合作方不是“拿来算”，而是“联合建模”——比如银行用工商业用户的电量特征做授信评分，这时候沙箱算起来很别扭，因为银行的特征变量在银行侧，电量特征在电网侧，两边数据形式不同，谁也不愿意把原始数据给对方。纵向联邦就是为这个场景设计的：双方在本地计算中间统计量，只交换加密后的梯度或参数。

最小参数的联邦训练配置，我一般这样写：

federated_learning: mode: vertical algorithm: logistic_regression sample_alignment: method: private_set_intersection key: ecdh_secp256k1 encryption: type: paillier key_length: 3072 training: batch_size: 2048 epochs: 10 learning_rate: 0.02 privacy: max_epsilon_per_epoch: 0.1 total_epsilon_budget: 1.0

纵向联邦里有三个参数最容易被忽略。第一个是sample_alignment，两边的用户 ID 不能直接交换明文，要用私密集合求交，否则样本对齐过程就把“哪些用户在两边都有数据”泄露了。第二个是paillier密钥长度，3072 位是当前安全基线，低于 2048 位坚决不用。第三个是total_epsilon_budget，它决定了整个联邦训练过程对个体隐私的总消耗，预算耗尽后模型要继续训练，只能重新初始化或者用更强的加密方案。

4.3 沙箱还是联邦：一张表定取舍

沙箱和联邦不是非此即彼，它们的取舍取决于合作方的技术能力和数据形态。我在项目里用下面这张表做决策：

从一线经验看：如果合作方只是想“看一张分析报告”，沙箱半天就能搭完方案；如果对方要把电量特征并入生产线做每日评分，纵向联邦才是正解。两者的共同前提是分级分类先做完，否则你不知道哪些数据该进沙箱、哪些字段能进联邦训练。

5. 用电量数据分享避坑手册：5 条最常见的翻车现场

5.1 现象：定时任务把含户号的明细直接发给了合作方

第一次做分享时，我搭了一个定时任务，每天生成前一天的用电汇总表发给合作方。上线一周后被安全团队抽查发现，表头里有user_id，而且是从数仓原表直接 SELECT 出来的，根本没走脱敏流程。

原因：任务复用了数据分析用的旧脚本，只加了过滤条件，没加脱敏函数。

解决：把分享管线的入口收口成唯一的服务，数据只能通过build_share_package出去，不允许任何人直接连数仓导出文件。对定时任务增加硬校验，在写出前检查user_id、户名等黑名单字段，一旦检测到就报错并终止发送。

5.2 现象：脱敏数据被反推出用户身份，合作方还拿出了证据

某次交付完 30 天负荷曲线，合作方的模型团队说他们用“行业+容量+地区”把数据跟公开的工商信息做了匹配，识别出了多家重点企业的真实负荷。

原因：分组太少。某地区只有一家大型制造企业，行业分到大类、容量桶是“500-1000”，组合起来就是唯一记录，k-匿名根本没满足。

解决：给所有可能的准标识符组合做幂集检查。代码里加一段循环，把quasi的 2 的 n 次方个子集全部跑一遍check_k_anonymity，任何一个组合的等价类小于 k 就重新泛化。这个操作耗时不到一分钟，却能避免最严重的这类翻车。

5.3 现象：过期授权 token 还能下载历史分享包

合作方反馈下载链接失效，我排查时发现存储桶里的对象没设置生命周期策略，token 虽然过期了，但文件仍然可以直接访问。

原因：只做了应用层的 token 校验，没有做存储层的最小权限控制。

解决：给每个分享包生成独立的存储路径和独立的访问凭证，token 过期时同时删除对应的临时凭证。更稳妥的做法是给存储桶开启对象级保留策略，所有分享包默认 7 天自动清理，需要延期的必须走重新授权流程。

5.4 现象：差分隐私噪声加上去后，月度汇总数据对不上账

一个对外发布的用电指数，连续三个月和电网侧内部统计对不上，误差达到 20%。业务方投诉数据“失真”。

原因：每次都独立加噪声，且 epsilon 设得太小，噪声方差过大，累计误差没有约束。

解决：改用固定种子的噪声生成，让同一个查询在预算内可重复；对月度汇总这类低敏感查询单独放开 epsilon，从 0.1 提到 0.5 到 1.0。另外加一个一致性修正步骤：加噪后的小计强制向总量对齐，即“先加噪出小计，再用小计差异修正总量”，保证账面上始终自洽。

5.5 现象：审计日志只记了“谁下载了文件”，查不出查了哪些用户

泄露事件复盘时，我打开审计日志发现只有三列：时间、操作人、文件名。安全团队问“对方跑了哪些查询、筛了哪些条件”，日志里完全没有。

原因：设计的初始只考虑了文件型下载，忽略了接口型查询的审计。

解决：对每一次查询请求记录入参、条件哈希、返回行数。条件哈希用sha256(request_body)实现，既能判断两次请求是否相同，又不会把具体筛选条件明文落盘。这样出问题时可以对比不同请求的重叠度，快速定位异常行为模式。

6. 验证你的分享方案靠不靠谱：一套自测清单和两个必调参数

方案搭完之后，真正决定能不能上线的是验证环节。我每次都会做一轮“攻击者视角”测试：把脱敏数据包当成已经完全公开，尝试用各种外部数据源重新识别用户。具体做法是准备一份模拟的工商登记地址表，和脱敏数据做模糊匹配，看能否对上。识别成功率超过 1%，方案就必须打回重做。之前某次做社区级数据分享时，这个测试就抓出过问题——泛化后的地址前缀依然存在多户共用一个地址的情况，看起来像是不同用户，实际上是同一地址下的多个户号，直接暴露了内部房间数量。

两个必调参数值得单独说。第一是 k 值。k=5 是数据采样和质量研究的下限，k=10 适合常规业务，如果数据稀疏导致无法泛化，不要盲目把 k 往下压，而是先考虑减少字段维度。第二是隐私预算 epsilon 的分配。我一般会把它分成三份：50% 给核心报表接口，30% 给临时查询，20% 作为应急储备。预算一旦耗尽，宁可拒绝查询也不要调大 epsilon，否则隐私保护会变成一句空话。

上线后我还会做一次“双跑验证”：同一个分享任务，用原始数据直接算一遍报表，再用脱敏数据跑一遍，两个结果做差异分析。如果差异超过业务口径的容忍阈值，说明脱敏过度或噪声不当，要回调参数。这个验证会一直持续到数据下线，因为每一次分享批次的数据形态不同，参数也需要跟着微调。有一次我把容量桶边界改细后忘了重新验证报表，结果差异率直接跳到 8%，那是被业务方找上门才发现的教训。现在我的习惯是：任何脱敏逻辑变更，都必须触发双跑验证流程，不能只做单元测试。这套流程看起来繁琐，但能把“分享出去之后才发现问题”的成本降到最低，希望帮到你。

本文还有配套的精品资源，点击获取