摘要

本文深入讲解DolphinDB能耗实时监控技术。从能耗数据采集到实时统计，从能耗分析到可视化展示，从能耗预测到节能优化，全面介绍能耗监控的核心方法。通过丰富的代码示例，帮助读者掌握能耗数据可视化的核心技能。

---

一、能耗监控概述

1.1 能耗监控架构

1.2 能耗类型

类型	单位	说明
电力	kWh	设备用电
燃气	m³	燃气消耗
蒸汽	t	蒸汽消耗
水	m³	用水量

1.3 监控指标

指标	说明
实时功率	当前功率
累计能耗	累计消耗
单位能耗	单位产品能耗
能耗成本	能耗费用

---

二、能耗数据采集

2.1 能耗数据表

// 能耗数据表
share streamTable(100000:0, 
    `meter_id`device_id`timestamp`power`voltage`current`energy,
    [SYMBOL, SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, DOUBLE]) as energy_stream

// 启用持久化
enableTablePersistence(energy_stream, true, true, 1000000)

2.2 分布式存储

// 创建分布式表
db = database("dfs://energy_db", VALUE, 1..100)
schema = table(1:0, 
    `meter_id`device_id`timestamp`power`voltage`current`energy,
    [SYMBOL, SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, DOUBLE])
db.createPartitionedTable(schema, `energy_data, `device_id)

// 订阅写入
subscribeTable(, "energy_stream", "persist", -1,
    def(msg) {
        loadTable("dfs://energy_db", "energy_data").append!(msg)
    }, 10000, 5000)

2.3 数据采集接口

// 能耗数据上报接口
def reportEnergy(meterId, deviceId, power, voltage, current, energy) {
    insert into energy_stream values (
        meterId, deviceId, now(), power, voltage, current, energy
    )
}

---

三、实时统计

3.1 实时功率统计

// 实时功率聚合
share table(1:0, 
    `time_window`device_id`avg_power`max_power`min_power,
    [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as power_agg

// 功率聚合引擎
powerEngine = createTimeSeriesEngine("power_engine", 60000,
    <[avg(power) as avg_power,
      max(power) as max_power,
      min(power) as min_power]>,
    power_agg, `timestamp, `device_id)

subscribeTable(, "energy_stream", "power_agg", -1, powerEngine, true)

3.2 累计能耗统计

// 累计能耗表
share table(1:0, 
    `device_id`total_energy`update_time,
    [SYMBOL, DOUBLE, TIMESTAMP]) as energy_total

// 累计计算
def calculateTotalEnergy(deviceId) {
    data = select last(energy) as last_energy
           from energy_stream
           where device_id = deviceId
    
    if (data.rows() > 0) {
        update energy_total
        set total_energy = data.last_energy[0], update_time = now()
        where device_id = deviceId
    }
}

3.3 单位能耗计算

// 单位能耗计算
def calculateUnitEnergy(deviceId, startTime, endTime) {
    // 获取能耗
    energy = select sum(energy) as total_energy
             from energy_stream
             where device_id = deviceId
             and timestamp between startTime and endTime
    
    // 获取产量
    production = select count(*) as total_production
                 from production_stream
                 where device_id = deviceId
                 and timestamp between startTime and endTime
    
    if (production.total_production[0] == 0) {
        return 0.0
    }
    
    return energy.total_energy[0] / production.total_production[0]
}

---

四、能耗分析

4.1 能耗分布分析

// 能耗分布
def getEnergyDistribution(startTime, endTime) {
    return select device_id,
                  sum(energy) as total_energy,
                  sum(energy) * 100.0 / (select sum(energy) from energy_stream 
                                         where timestamp between startTime and endTime) as percentage
           from energy_stream
           where timestamp between startTime and endTime
           group by device_id
           order by total_energy desc
}

4.2 能耗趋势分析

// 能耗趋势
def getEnergyTrend(deviceId, startTime, endTime, interval = 3600000) {
    return select bar(timestamp, interval) as time_window,
                  sum(energy) as total_energy,
                  avg(power) as avg_power
           from energy_stream
           where device_id = deviceId
           and timestamp between startTime and endTime
           group by bar(timestamp, interval)
}

4.3 能耗对比分析

// 同比对比
def compareYoY(deviceId) {
    now = now()
    
    current = getEnergyTotal(deviceId, now - 86400000, now)
    lastYear = getEnergyTotal(deviceId, now - 365*86400000, now - 364*86400000)
    
    return dict(STRING, ANY, [
        ["current", current],
        ["lastYear", lastYear],
        ["change", (current - lastYear) * 100.0 / lastYear]
    ])
}

// 环比对比
def compareMoM(deviceId) {
    now = now()
    
    current = getEnergyTotal(deviceId, now - 86400000, now)
    lastMonth = getEnergyTotal(deviceId, now - 60*86400000, now - 59*86400000)
    
    return dict(STRING, ANY, [
        ["current", current],
        ["lastMonth", lastMonth],
        ["change", (current - lastMonth) * 100.0 / lastMonth]
    ])
}

---

五、能耗预测

5.1 简单预测

// 移动平均预测
def predictEnergy(deviceId, periods = 7) {
    data = select sum(energy) as daily_energy
           from energy_stream
           where device_id = deviceId
           group by date(timestamp)
           order by date(timestamp) desc
           limit periods
    
    if (data.rows() == 0) {
        return 0.0
    }
    
    return avg(data.daily_energy)
}

5.2 趋势预测

// 线性趋势预测
def predictEnergyTrend(deviceId, futureDays = 7) {
    data = select sum(energy) as daily_energy
           from energy_stream
           where device_id = deviceId
           group by date(timestamp)
           order by date(timestamp)
    
    if (data.rows() < 2) {
        return 0.0
    }
    
    // 简单线性回归
    n = data.rows()
    x = 1..n
    y = data.daily_energy
    
    sumX = sum(x)
    sumY = sum(y)
    sumXY = sum(x * y)
    sumX2 = sum(x * x)
    
    slope = (n * sumXY - sumX * sumY) / (n * sumX2 - sumX * sumX)
    intercept = (sumY - slope * sumX) / n
    
    return intercept + slope * (n + futureDays)
}

---

六、可视化展示

6.1 能耗大屏数据

// 能耗大屏
def getEnergyDashboard() {
    now = now()
    
    return dict(STRING, ANY, [
        ["totalEnergy", getTotalEnergy(now - 86400000, now)],
        ["avgPower", getAvgPower(now - 3600000, now)],
        ["peakPower", getPeakPower(now - 86400000, now)],
        ["unitEnergy", getUnitEnergy(now - 86400000, now)],
        ["energyCost", getEnergyCost(now - 86400000, now)]
    ])
}

def getTotalEnergy(startTime, endTime) {
    return exec sum(energy) from energy_stream
           where timestamp between startTime and endTime
}

def getAvgPower(startTime, endTime) {
    return exec avg(power) from energy_stream
           where timestamp between startTime and endTime
}

def getPeakPower(startTime, endTime) {
    return exec max(power) from energy_stream
           where timestamp between startTime and endTime
}

6.2 能耗排名

// 能耗排名
def getEnergyRanking(limit = 10) {
    now = now()
    
    return select device_id,
                  sum(energy) as total_energy,
                  rank() over order by sum(energy) desc as rank
           from energy_stream
           where timestamp > now - 86400000
           group by device_id
           limit limit
}

6.3 能耗曲线

// 能耗曲线数据
def getEnergyCurve(deviceId, startTime, endTime) {
    return select timestamp as time,
                  power,
                  energy
           from energy_stream
           where device_id = deviceId
           and timestamp between startTime and endTime
           order by timestamp
}

---

七、节能优化

7.1 能耗异常检测

// 能耗异常检测
def detectEnergyAnomaly(deviceId) {
    data = select avg(power) as avg_power
           from energy_stream
           where device_id = deviceId
           and timestamp > now() - 86400000
    
    avgPower = data.avg_power[0]
    stdPower = exec std(power) from energy_stream
               where device_id = deviceId
               and timestamp > now() - 86400000
    
    currentPower = exec last(power) from energy_stream
                   where device_id = deviceId
    
    if (abs(currentPower - avgPower) > 3 * stdPower) {
        return true
    }
    
    return false
}

7.2 节能建议

// 节能建议
def generateEnergyAdvice(deviceId) {
    advice = array(STRING, 0)
    
    // 检查峰谷用电
    peakUsage = getPeakUsage(deviceId)
    valleyUsage = getValleyUsage(deviceId)
    
    if (peakUsage > valleyUsage * 1.5) {
        advice.append!("建议将部分生产转移到谷电时段")
    }
    
    // 检查设备效率
    unitEnergy = calculateUnitEnergy(deviceId, now() - 86400000, now())
    targetEnergy = getTargetEnergy(deviceId)
    
    if (unitEnergy > targetEnergy * 1.2) {
        advice.append!("设备能耗偏高，建议检查设备运行状态")
    }
    
    return advice
}

---

八、实战案例

7.1 完整能耗监控系统

// ========== 能耗实时监控系统 ==========

// 1. 创建数据表
share streamTable(100000:0, 
    `meter_id`device_id`timestamp`power`voltage`current`energy,
    [SYMBOL, SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, DOUBLE]) as energy_stream

enableTablePersistence(energy_stream, true, true, 1000000)

// 2. 创建分布式表
db = database("dfs://energy_db", VALUE, 1..100)
schema = table(1:0, 
    `meter_id`device_id`timestamp`power`voltage`current`energy,
    [SYMBOL, SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, DOUBLE])
db.createPartitionedTable(schema, `energy_data, `device_id)

// 3. 订阅写入
subscribeTable(, "energy_stream", "persist", -1,
    def(msg) {
        loadTable("dfs://energy_db", "energy_data").append!(msg)
    }, 10000, 5000)

// 4. 功率聚合
share table(1:0, 
    `time_window`device_id`avg_power`max_power`min_power,
    [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as power_agg

powerEngine = createTimeSeriesEngine("power_engine", 60000,
    <[avg(power) as avg_power,
      max(power) as max_power,
      min(power) as min_power]>,
    power_agg, `timestamp, `device_id)

subscribeTable(, "energy_stream", "power_agg", -1, powerEngine, true)

// 5. 模拟数据
def generateMockEnergy() {
    while (true) {
        data = table(
            "M" + string(rand(100, 10)) as meter_id,
            take(1..10, 10) as device_id,
            take(now(), 10) as timestamp,
            rand(50.0..100.0, 10) as power,
            rand(220.0..240.0, 10) as voltage,
            rand(10.0..50.0, 10) as current,
            rand(1000.0..2000.0, 10) as energy
        )
        energy_stream.append!(data)
        sleep(5000)
    }
}

submitJob("mock_energy", "模拟能耗数据", generateMockEnergy)

// 6. 能耗看板接口
def getEnergyDashboard() {
    now = now()
    return select sum(energy) as total_energy,
                  avg(power) as avg_power,
                  max(power) as max_power
           from energy_stream
           where timestamp > now - 86400000
}

addFunctionView(getEnergyDashboard)

print("能耗实时监控系统启动完成")

---

八、总结

本文详细介绍了DolphinDB能耗实时监控：

1. 数据采集：能耗数据表、分布式存储
2. 实时统计：功率统计、累计能耗、单位能耗
3. 能耗分析：分布分析、趋势分析、对比分析
4. 能耗预测：简单预测、趋势预测
5. 可视化展示：能耗大屏、能耗排名、能耗曲线
6. 节能优化：异常检测、节能建议

思考题：

1. 如何提高能耗预测的准确性？
2. 如何设计有效的节能策略？
3. 如何实现能耗成本优化？

---

参考资料

• DolphinDB时序数据处理
• DolphinDB聚合计算

---