实用脚本如何融合多源数据进行综合？

wen 实用脚本 2026-10-22 2

实用脚本如何融合多源数据进行综合？

多源数据融合的核心是统一时空基准 → 对齐字段 → 关联实体 → 加权综合 → 校验输出,下面给出一套可落地的脚本设计思路与示例代码。

实用脚本如何融合多源数据进行综合？

融合的整体流程

多源数据（CSV/API/数据库/日志/JSON）
        │
   ①数据接入层：统一读取、编码、时间格式
        │
   ②清洗对齐层：去重、缺失填补、单位统一、时间对齐
        │
   ③实体关联层：主键/地理/时间窗口匹配（join / 空间连接）
        │
   ④融合计算层：加权、投票、置信度、规则引擎
        │
   ⑤输出校验层：质量指标、异常标记、结果落库

关键融合方法（按场景选）

场景	方法	说明
同一实体多来源字段	加权平均 / 中位数	按数据源可信度赋权
分类标签冲突	多数投票 / 置信度加权	结合各源历史准确率
时空数据	时空窗口匹配 + 插值	按时间最近、空间最近对齐
结构化+非结构化	抽取后 join	日志/文本抽字段再关联
缺失补全	规则填补 / 回归 / KNN	先规则后模型
冲突消解	规则优先级 + 置信度打分	明确"谁说了算"

示例脚本（Python / pandas）

场景：融合三个来源的"设备运行数据"——CRM（设备主数据）、IoT 采集（时序指标）、工单系统（故障标签）,输出每台设备的综合状态。

import pandas as pd
import numpy as np
# ---------- ① 数据接入 ----------
crm   = pd.read_csv("crm_devices.csv")        # device_id, model, region, install_date
iot   = pd.read_csv("iot_metrics.csv")        # device_id, ts, temp, vibration
order = pd.read_csv("work_orders.csv")        # device_id, ts, fault_type
# ---------- ② 清洗对齐 ----------
# 统一时间格式
for df in (iot, order):
    df["ts"] = pd.to_datetime(df["ts"], errors="coerce")
    df.dropna(subset=["ts"], inplace=True)
# IoT 去重 + 按小时聚合（降噪）
iot["hour"] = iot["ts"].dt.floor("h")
iot_h = (iot.groupby(["device_id", "hour"])
            .agg(temp=("temp", "median"),
                 vibration=("vibration", "median"))
            .reset_index())
# 工单：统计每个设备近7天故障次数
order["recent"] = order["ts"] >= order["ts"].max() - pd.Timedelta(days=7)
fault_cnt = (order[order["recent"]]
             .groupby("device_id")
             .agg(fault_7d=("fault_type", "count"))
             .reset_index())
# ---------- ③ 实体关联（以 device_id 为主键） ----------
df = crm.merge(iot_h, on="device_id", how="left") \
        .merge(fault_cnt, on="device_id", how="left")
df["fault_7d"] = df["fault_7d"].fillna(0)
# ---------- ④ 融合计算：加权综合得分 ----------
# 各指标归一化（0~1），权重可按业务/可信度设定
def norm(s):
    return (s - s.min()) / (s.max() - s.min() + 1e-9)
df["temp_score"]      = 1 - norm(df["temp"].fillna(df["temp"].median()))
df["vib_score"]       = 1 - norm(df["vibration"].fillna(df["vibration"].median()))
df["fault_score"]     = 1 - norm(df["fault_7d"])
W = {"temp_score": 0.3, "vib_score": 0.4, "fault_score": 0.3}
df["health"] = sum(df[k] * w for k, w in W.items())
# 规则引擎覆盖：多源冲突时按规则兜底
df["status"] = np.select(
    [df["fault_7d"] >= 3, df["health"] < 0.3, df["health"] < 0.6],
    ["故障", "高危", "关注"],
    default="正常"
)
# ---------- ⑤ 输出与质量校验 ----------
quality = {
    "总设备数": len(df),
    "缺失率": df[["temp","vibration"]].isna().mean().round(3).to_dict(),
    "状态分布": df["status"].value_counts().to_dict(),
}
print(quality)
df.to_csv("device_fusion_result.csv", index=False)