数据从哪里来

只显示当前值

设备、蜂箱、采集时间、接收时间全部关联

异常是不是噪声

单点越限即告警

物理范围 + 中值/平滑 + 连续异常

为什么判高风险

只给红色状态

返回模型概率、规则命中与异常原因

告警之后怎么办

推送结束

接单、巡检、结论、标签回流、模型评估

bee/{farmId}/hive/{hiveId}/telemetry
bee/{farmId}/hive/{hiveId}/status
bee/{farmId}/hive/{hiveId}/alarm

// telemetry payload 示例
{
"deviceId": "D-THCO2-027",
"sampleTime": "2026-09-27T02:10:00",
"insideTemp": 34.1,
"insideHumidity": 84.6,
"co2": 1912,
"weight": 31.8,
"activity": 106,
"soundDb": 58.2
}

JSON完整性

缺少 sampleTime

拒绝并记录设备异常

设备身份

deviceId 未注册

进入非法设备日志

绑定关系

设备未绑定蜂箱

拒绝进入业务时序

时间漂移

采集时间晚于接收时间 2 小时

保留原值并标记 timeDrift

重复消息

唯一键已存在

幂等忽略

物理范围

湿度 <0 或 >100

标记无效，不参与模型

缺失数据

CO₂ 缺失

保留 missing 标记，不伪造关键病害特征

@Service
@RequiredArgsConstructor
public class BeeTelemetryConsumer {

private final TelemetryApplicationService applicationService;

public void onMessage(String topic, String payload) {
TelemetryMessage message = TelemetryMessage.parse(payload);

applicationService.accept(
topic,
message.deviceId(),
message.hiveId(),
message.sampleTime(),
message);
}
}

@Transactional
public void accept(String topic,
String deviceId,
String hiveId,
LocalDateTime sampleTime,
TelemetryMessage message) {

deviceService.requireActiveBinding(deviceId, hiveId);

String uniqueKey = hiveId + "|" + deviceId + "|" + sampleTime;
if (!dedupService.tryAcquire(uniqueKey)) {
return;
}

ValidationResult result = telemetryValidator.validate(message);
telemetryRepository.save(result.toRecord());

if (result.canEnterRiskModel()) {
featureQueue.publish(result.recordId());
}
}

短时尖峰

直接触发高风险

保留原值，同时计算过滤值并标记质量

断网 1 小时

丢弃数据

本地缓存，恢复后顺序补传

补传重复

重复入库

服务端业务唯一键去重

传感器恒值

当成稳定正常

检测长时间零方差，产生设备健康告警

设备离线

沉默不处理

LWT/心跳超时产生设备告警

public final class EwmaCalculator {
private final double alpha;
private Double state;

public EwmaCalculator(double alpha) {
if (alpha <= 0.0 || alpha > 1.0) {
throw new IllegalArgumentException("alpha范围错误");
}
this.alpha = alpha;
}

public double update(double value) {
if (!Double.isFinite(value)) {
throw new IllegalArgumentException("观测值必须为有限数");
}
state = state == null
? value
: alpha * value + (1.0 - alpha) * state;
return state;
}
}

public double score(double value) {
if (values.size() < 3) {
add(value);
return 0.0;
}

double mean = values.stream()
.mapToDouble(Double::doubleValue)
.average().orElse(0.0);

double variance = values.stream()
.mapToDouble(v -> (v - mean) * (v - mean))
.average().orElse(0.0);

double sd = Math.sqrt(variance);
double z = sd < 0.000001 ? 0.0 : (value - mean) / sd;

add(value);
return z;
}

public double predict(double humidityZ,
double co2Z,
double weightDropKg,
double activityDropRate) {
double intercept = -2.20;

double z = intercept
+ 0.72 * humidityZ
+ 0.95 * co2Z
+ 0.48 * weightDropKg
+ 1.15 * activityDropRate;

z = Math.max(-30.0, Math.min(30.0, z));
return 1.0 / (1.0 + Math.exp(-z));
}

double score = probability * 60.0;

if (co2 >= 1800.0) score += 15.0;
if (humidity >= 82.0) score += 10.0;
if (activityDropRate >= 0.45) score += 10.0;
if (consecutiveAbnormalCount >= 6) score += 15.0;

if (score >= 70.0) return "高风险";
if (score >= 40.0) return "中风险";
return "低风险";

首次异常

进入观察，不立即高等级通知

连续 N 次

进入待确认或告警

冷却窗

同一蜂箱同类告警 60 分钟内不重复通知

风险升级

中→高不受冷却窗限制

指标恢复

连续正常后关闭，记录恢复时间

再次异常

关闭后重新累计，形成新告警事件

bee:hive:{id}:latest

最新监测状态

可重建、高频读取

bee:device:{id}:heartbeat

最后心跳

TTL 可直接判断离线

bee:alert:{hive}:{rule}:cooldown

告警抑制状态

天然适合过期窗口

bee:feature:{hive}:{name}

短窗口特征状态

在线计算频繁访问

bee:dedup:{messageId}

消息去重标记

短 TTL 幂等

CREATE TABLE bee_telemetry (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
farm_id VARCHAR(32) NOT NULL,
hive_id VARCHAR(32) NOT NULL,
device_id VARCHAR(64) NOT NULL,
sample_time DATETIME(3) NOT NULL,
receive_time DATETIME(3) NOT NULL,
inside_temp DECIMAL(6,2),
inside_humidity DECIMAL(6,2),
co2 DECIMAL(10,2),
weight_kg DECIMAL(8,2),
activity_count INT,
sound_db DECIMAL(6,2),
quality_flag VARCHAR(32) NOT NULL,
message_id VARCHAR(96) NOT NULL,
UNIQUE KEY uk_message (message_id),
KEY idx_hive_time (hive_id, sample_time)
);

CREATE TABLE bee_alert (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
hive_id VARCHAR(32) NOT NULL,
risk_score DECIMAL(6,2) NOT NULL,
risk_level VARCHAR(16) NOT NULL,
model_version VARCHAR(32) NOT NULL,
reasons JSON NOT NULL,
status VARCHAR(24) NOT NULL,
triggered_at DATETIME(3) NOT NULL,
acknowledged_at DATETIME(3),
closed_at DATETIME(3),
KEY idx_hive_triggered (hive_id, triggered_at)
);

<template>
<div class="risk-card" v-for="item in highRiskHives" :key="item.hiveId">
<strong>{{ item.hiveId }} · {{ item.score }} 分</strong>

<ul>
<li v-for="reason in item.reasons" :key="reason.code">
{{ reason.message }}
</li>
</ul>

<button @click="openHive(item.hiveId)">查看趋势与巡检记录</button>
</div>
</template>

private static final int RECORD_COUNT = 50000;
private static final int HIVE_COUNT = 100;
private static final Random RANDOM = new Random(20250308L);

boolean rain = RANDOM.nextDouble() < 0.12;
boolean ventilationFault = RANDOM.nextDouble() < 0.045;
boolean diseaseTrend = RANDOM.nextDouble() < 0.035;

double riskScore = clamp(
(insideHumidity - 70.0) * 1.2
+ (co2 - 1300.0) / 22.0
+ (220.0 - activity) / 4.0
+ (diseaseTrend ? 22.0 : 0.0),
0.0, 100.0);

double precision = tp + fp == 0
? 0.0 : (double) tp / (tp + fp);

double recall = tp + fn == 0
? 0.0 : (double) tp / (tp + fn);

double f1 = precision + recall == 0.0
? 0.0
: 2.0 * precision * recall / (precision + recall);

Precision

发出的告警有多少是真的？

过低会造成告警疲劳

Recall

真实异常有多少被抓到？

过低意味着漏掉应巡检蜂箱

F1

Precision 与 Recall 是否平衡？

便于比较不同阈值/版本

混淆矩阵

错在哪里？

区分误报 FP 与漏报 FN

确认蜂群状态异常

VALID_ALERT

正样本

现场正常，模型误报

FALSE_ALERT

误报分析与阈值校准

传感器故障

DEVICE_FAULT

排除出病害训练标签

需要送检

PENDING_LAB

暂不作为确定标签

管理操作导致变化

MANAGEMENT_EVENT

作为上下文特征

重复投递

同一 messageId 连续发布 2 次

历史表只新增 1 条

离线补传

断网缓存 6 个周期再恢复

按 sampleTime 补传，receiveTime 保留真实接收时间

单点尖峰

湿度从 68 突变 96 后恢复

质量标记/滤波抑制，不直接形成持续高风险

持续异常

湿度+CO₂异常 6 个周期

风险逐步升级并形成一个告警事件

告警风暴

同规则持续命中

冷却窗内不重复通知

设备恒值

温度长时间完全不变

产生设备质量告警

人工误报

巡检标记 FALSE_ALERT

进入模型评估数据集

MQTT瞬时重连

大量补传同时进入

消费解耦、批量入库、背压

历史曲线

长时间范围查询慢

按小时/日预聚合

特征计算

重复扫描窗口

Redis/内存维护短窗口状态

告警扫描

全表定时轮询

事件驱动增量计算

ECharts大曲线

浏览器点数过多

服务端降采样/聚合

长期数据

单表持续膨胀

按时间分区或冷热归档

伪造设备

每设备独立凭证或证书

跨蜂场发布

Broker Topic ACL

明文传输

TLS

重放旧消息

messageId + sampleTime + 时间窗

异常高频

设备级限流

凭证泄漏

可吊销、可轮换，不写死在前端

一个阈值判断病害

天气和噪声导致误报

连续异常 + 多因子联合

MQTT QoS 1 就认为不重复

至少一次可能重复

业务幂等

只保存入库时间

补传数据时间错位

采集时间 + 接收时间

缺失值一律均值填充

可能虚构关键病害信号

关键字段保留 missing 标记

风险只返回分数

无法解释

返回 reasons + modelVersion

模拟数据算出高准确率

存在标签泄漏/自证

模拟用于链路测试，真实标签用于评估

告警一触发就结束

没有反馈数据

巡检、复核、标签回流