DolphinDB物联网实测手记：用环境传感器数据跑通时序分析的完整链路

// 按日期 + 节点ID复合分区
db1 = database("", VALUE, 2024.01.01..2025.12.31)
db2 = database("", HASH, [SYMBOL, 5])
db = database("dfs://env_monitor", COMPO, [db1, db2])

sensor = table(1:0, `ts`nodeId`temperature`humidity`co2`pm25`lightLevel`,
               [DATETIME, SYMBOL, DOUBLE, DOUBLE, DOUBLE, DOUBLE, DOUBLE])
db.createPartitionedTable(sensor, "sensor", `ts`nodeId)

import dolphindb as ddb
import pandas as pd

sess = ddb.Session()
sess.connect("localhost", 8848, "admin", "123456")

# 读取传感器CSV，解析时间列
df = pd.read_csv("env_sensor_20240615.csv", parse_dates=["ts"])

# 写入DolphinDB
sess.run("tableInsert{loadTable('dfs://env_monitor', 'sensor')}", df)
sess.close()

// 找出某个节点数据中断超过5分钟的时刻
select ts, nodeId,
       double(ts - prev(ts)) / 1000 as gapSeconds
from sensor
where nodeId = "N001" and date(ts) = 2024.06.15
context by nodeId
having double(ts - prev(ts)) / 1000 > 300

// 对温度和湿度做前值填充（缺失时间不超过2分钟的）
select ts, nodeId,
       iif(isNull(temperature), ffill(temperature), temperature) as temperature,
       iif(isNull(humidity), ffill(humidity), humidity) as humidity,
       co2, pm25, lightLevel
from sensor
where nodeId = "N001" and date(ts) = 2024.06.15
context by nodeId

// 每个节点每分钟的平均值
select nodeId,
       bar(ts, 60000) as minuteTs,  // 按60秒对齐
       avg(temperature) as avgTemp,
       avg(humidity) as avgHumidity,
       avg(co2) as avgCo2,
       avg(pm25) as avgPm25,
       max(pm25) as maxPm25
from sensor
where date(ts) = 2024.06.15
group by nodeId, bar(ts, 1m)

// 一天内的温度小时趋势
select hour(ts) as hour,
       avg(temperature) as avgTemp,
       min(temperature) as minTemp,
       max(temperature) as maxTemp,
       max(temperature) - min(temperature) as tempRange
from sensor
where nodeId = "N001" and date(ts) = 2024.06.15
group by hour(ts)
order by hour

// PM2.5的小时分位数分布
select hour(ts) as hour,
       percentile(pm25, 25) as p25,
       percentile(pm25, 50) as median,
       percentile(pm25, 75) as p75,
       percentile(pm25, 95) as p95,
       percentile(pm25, 99) as p99
from sensor
where date(ts) = 2024.06.15
group by hour(ts)
order by hour

// 找出温度超过30℃或CO2超过1000ppm的记录
select ts, nodeId, temperature, co2, pm25
from sensor
where date(ts) = 2024.06.15
    and (temperature > 30 or co2 > 1000)
order by ts

// 找出CO2连续超标超过10分钟的时段
select nodeId,
       first(ts) as anomalyStart,
       last(ts) as anomalyEnd,
       count(*) as durationSeconds,
       avg(co2) as avgCo2,
       max(co2) as peakCo2
from (
    select ts, nodeId, co2,
           segment(ts, 1m) as seg  // 间隔超过1分钟则断开
    from sensor
    where date(ts) = 2024.06.15 and co2 > 1000
    context by nodeId
)
group by nodeId, segment(ts, 1m) as seg
having count(*) > 600  // 持续超过600秒（10分钟）
order by anomalyStart

// 用3-sigma原则找出每个节点的温度离群点
select ts, nodeId, temperature,
       avgTemp, stdTemp,
       abs(temperature - avgTemp) / stdTemp as zscore
from (
    select ts, nodeId, temperature,
           mavg(temperature, 300) as avgTemp,
           mstd(temperature, 300) as stdTemp
    from sensor
    where nodeId = "N001" and date(ts) = 2024.06.15
    context by nodeId
)
where stdTemp > 0 and abs(temperature - avgTemp) / stdTemp > 3
order by ts

// 计算温度、湿度、CO2、PM2.5之间的相关系数矩阵
select corr(temperature, humidity) as temp_humidity,
       corr(temperature, co2) as temp_co2,
       corr(humidity, pm25) as humidity_pm25,
       corr(co2, pm25) as co2_pm25
from sensor
where date(ts) = 2024.06.15

// 工作时间 vs 非工作时间的温湿度 - PM2.5相关性 SELECT
  CASE
    WHEN
      HOUR(ts) BETWEEN 8
      AND 18 THEN
      "工作时间"
    ELSE
      "非工作时间"
  END AS period,
  corr (temperature, pm25) AS temp_pm25,
  corr (humidity, pm25) AS humidity_pm25,
  avg(pm25) AS avgPm25
FROM
  sensor
WHERE
  DATE(ts) = 2024.06.15
GROUP BY
  CASE
    WHEN HOUR(ts) BETWEEN 8
      AND 18 THEN
      "工作时间"
    ELSE
      "非工作时间"
  END

@state
def envAlert(temperature, co2, pm25, tempThreshold, co2Threshold, pm25Threshold){
    tempAlert = iif(temperature > tempThreshold, 1, 0)
    co2Alert = iif(co2 > co2Threshold, 1, 0)
    pm25Alert = iif(pm25 > pm25Threshold, 1, 0)
    return (tempAlert + co2Alert + pm25Alert) > 0
}

// 流数据表
share streamTable(1:0, `ts`nodeId`temperature`humidity`co2`pm25`lightLevel`,
                  [DATETIME, SYMBOL, DOUBLE, DOUBLE, DOUBLE, DOUBLE, DOUBLE]) as envStream

// 告警输出表
alerts = table(1:0, `ts`nodeId`temperature`co2`pm25`isAlert`,
               [DATETIME, SYMBOL, DOUBLE, DOUBLE, DOUBLE, INT])

factors = <[temperature, co2, pm25, envAlert(temperature, co2, pm25, 30, 1000, 75)]>
engine = createReactiveStateEngine(
    name="envAlert",
    metrics=factors,
    dummyTable=envStream,
    outputTable=alerts,
    keyColumn="nodeId"
)

subscribeTable(tableName=`envStream, actionName="alert",
               handler=engine)

inputDS = replayDS(
    <select * from loadTable("dfs://env_monitor", "sensor")
     where date(ts) = 2024.06.15>,
    `ts
)
replay(inputDS, envStream, `ts, 1000, true, 2)