# 监测数据怎么才算数

一个自动监测站每分钟往回送一行数：站点、时间、参数、数值、设备状态。运维最怕的不是缺数据，是那行数字明明躺在库里，事后却说不清它能不能算数——设备当时在校准，时间戳被回拨过，同一条还传了两遍。

得在入库前定一套能重跑的判别规则，把判定过程和原始行一起留下：判有效、留痕迹、去重复，全用本地文件跑。

## 先说清楚判什么

有效性判别只管「这行能不能进统计」，不管「这个数准不准」；探头漂移得靠现场核查。

异常归成五类：字段是否齐全、数值是否在量程内、时间有没有回拨、设备是不是维护或校准、相邻两次差值是否跳得太猛。最后一类最容易被跳过——把「待核」和「有效」混在一起，后面所有均值、超标次数都会跟着偏。

## 判别怎么落地

sample.csv 表头是 station,ts,param,value,status；下面逐行判，结果写两个文件。

```python
import csv, json, hashlib
from datetime import datetime

LIMITS = {"pm25": (0, 500), "so2": (0, 1500)}
STEP = {"pm25": 80, "so2": 120}

def check(row, last):
    why = []
    for k in ("station", "ts", "param", "value", "status"):
        if not row.get(k, "").strip():
            why.append("missing:" + k)
    if why:
        return False, why
    t = datetime.fromisoformat(row["ts"])
    v = float(row["value"])
    lo, hi = LIMITS.get(row["param"], (0, 1e9))
    if v < lo or v > hi:
        why.append("out_of_range")
    if row["status"].strip() not in ("", "normal", "ok"):
        why.append("device_state")
    if last and last["param"] == row["param"]:
        if t < last["t"]:
            why.append("time_rollback")
        elif abs(v - last["v"]) > STEP.get(row["param"], 1e9):
            why.append("jump_pending")
    return len(why) == 0, why
```

五类规则写成一串判断，每行攒一个 why 列表，为空才算有效；jump_pending 只标「待核」留人复核，阈值写成字典，改配置不动逻辑。

主循环把三件事收口：

```python
rows = list(csv.DictReader(open("sample.csv", encoding="utf-8")))
valid, bad, ledger, last, seen = [], [], [], None, set()
for row in rows:
    raw = json.dumps(row, sort_keys=True, ensure_ascii=False)
    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
    ok, why = check(row, last)
    key = (row.get("station"), row.get("ts"), row.get("param"))
    if key in seen:
        ok, why = False, ["duplicate"]
    seen.add(key)
    ledger.append({"sha256": digest, "valid": ok, "reason": why, "rule": "v1"})
    (valid if ok else bad).append(row)
    if ok:
        last = {"t": datetime.fromisoformat(row["ts"]),
                "v": float(row["value"]), "param": row["param"]}
with open("ledger.jsonl", "w", encoding="utf-8") as f:
    for x in ledger:
        f.write(json.dumps(x, ensure_ascii=False) + "\n")
print("total=%d valid=%d rejected=%d" % (len(rows), len(valid), len(bad)))
```

每行先算原始内容的 SHA-256（哈希就是给数据算一串唯一指纹，改一个字符指纹就全变），再判有效性、按「站点加时间加参数」去重；结果、理由、规则版本写进 ledger.jsonl，被拦下的另存备查。

跑起来只要一行 `python validate.py`。改 sample.csv 里某个数值再跑一次：sha256 完全不同，valid 和 reason 不该跟着变——原文被钉住了，判定逻辑只依赖规则。

## 为什么留台账

只存判定结果，过三个月没人能回答「当时按哪版规则判的」。台账三条字段就冲着这个：sha256 钉住原始行，reason 说明为什么拦，rule 记规则版本；规则一改版本跟着变，旧数据要重跑就重跑。

别误会两点：哈希只证明「这行没被动过」，不证明「现场那个数是对的」；台账记判定过程，不是采样过程。完整证据链还得配原始报文归档时间和人工复核记录。

## 重跑两遍要一样

网络重传会让同一条回来两次。上面用 seen 挡重复：第一次判完记账，第二次标 duplicate。

这一步决定台账能不能重放：输入不变连跑两遍，输出应一模一样，这叫幂等；做不到，重跑一次就多算一次。去重键别只用时间戳——同一站点、同一分钟、不同参数是三条数据，只按时间会误杀。

## 还差什么

这套东西只解决「入库前能不能算数」。采样真实性、检定有效期、谁签字复核，都不在脚本里。落地时别忘了：规则版本进台账、被拦下那批留着备查、阈值按量程定并留变更记录、台账设只读。

技术链路能做到的，是让每一步留得下痕迹、经得起重放；剩下的判断仍得由人签字。把「这行数能不能算」从口头确认变成能重跑的记录，值钱就在这儿。


一个自动监测站每分钟往回送一行数：站点、时间、参数、数值、设备状态。运维最怕的不是缺数据，是那行数字明明躺在库里，事后却说不清它能不能算数——设备当时在校准，时间戳被回拨过，同一条还传了两遍。

一行监测数据能不能算数：入库前的判别、留痕与去重

自动监测数据入库前怎么判「这行能不能算数」：五类异常判别、SHA-256 原始行留痕、按站点+时间+参数幂等去重，附可直接运行的 Python 校验脚本与重放验证步骤。

数据库

后端

算法

监测数据有效性判别需在入库前执行可重跑的规则：字段齐全、量程内、无时间回拨、设备正常、差值合理，五类异常分类处理；SHA-256哈希钉住原始行，按站点+时间+参数去重，结果写入台账记录规则版本和理由，确保重跑结果一致，形成可追溯的完整证据链。

代码解释

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

一行监测数据能不能算数：入库前的判别、留痕与去重-腾讯云开发者社区-腾讯云

一行监测数据能不能算数：入库前的判别、留痕与去重

一行监测数据能不能算数：入库前的判别、留痕与去重

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐