EOS电价接口：市场电价查询与价格序列

概述

EOS（Energy Optimization System）能源优化系统提供了强大的电价数据接口，支持从多个数据源获取实时和历史电价信息。通过智能算法，系统能够为能源优化决策提供准确的价格序列数据。本文将深入解析EOS的电价接口架构、使用方法和最佳实践。

核心架构设计

EOS的电价模块采用抽象工厂模式，提供了统一的接口规范：

classDiagram
    class ElecPriceProvider {
        <<abstract>>
        +provider_id() str
        +_update_data() None
        +records: List[ElecPriceDataRecord]
    }
    
    class ElecPriceDataRecord {
        +date_time: AwareDatetime
        +elecprice_marketprice_wh: float
        +elecprice_marketprice_kwh: float
    }
    
    class ElecPriceAkkudoktor {
        +_request_forecast() AkkudoktorElecPrice
        +_predict_ets() np.ndarray
    }
    
    class ElecPriceEnergyCharts {
        +_request_forecast() EnergyChartsElecPrice
        +_parse_data() pd.Series
    }
    
    class ElecPriceImport {
        +import_from_file() None
        +import_from_json() None
    }
    
    ElecPriceProvider <|-- ElecPriceAkkudoktor
    ElecPriceProvider <|-- ElecPriceEnergyCharts
    ElecPriceProvider <|-- ElecPriceImport
    ElecPriceProvider "1" *-- "many" ElecPriceDataRecord

支持的数据源

EOS目前支持三种电价数据源：

数据源	提供商ID	特点	更新频率
Akkudoktor	ElecPriceAkkudoktor	市场电价数据	每小时
Energy-Charts	ElecPriceEnergyCharts	电力数据	每天14:00
文件导入	ElecPriceImport	自定义数据源	按需

配置与初始化

基础配置

from akkudoktoreos.prediction.elecprice import ElecPriceCommonSettings
from akkudoktoreos.prediction.prediction import get_prediction

# 配置电价提供商
config = ElecPriceCommonSettings(
    provider="ElecPriceAkkudoktor",  # 选择数据源
    charges_kwh=0.21,  # 电费附加费用(€/kWh)
)

# 获取实例
prediction = get_prediction()

多提供商配置示例

# 配置多个电价数据源
config = ElecPriceCommonSettings(
    provider="ElecPriceEnergyCharts",
    provider_settings=ElecPriceImportCommonSettings(
        import_file_path="/path/to/backup_prices.json"
    )
)

数据获取与处理

实时电价查询

# 获取最新电价数据
prediction.update_data()

# 访问电价记录
for record in prediction.elecprice_akkudoktor.records:
    print(f"时间: {record.date_time}")
    print(f"市场电价: {record.elecprice_marketprice_kwh} €/kWh")
    print(f"含附加费电价: {record.elecprice_marketprice_kwh + config.charges_kwh} €/kWh")

价格序列分析

import pandas as pd
import numpy as np

# 转换为Pandas DataFrame进行分析
price_series = pd.Series({
    record.date_time: record.elecprice_marketprice_kwh 
    for record in prediction.elecprice_akkudoktor.records
})

# 计算统计指标
stats = {
    "平均电价": price_series.mean(),
    "最高电价": price_series.max(),
    "最低电价": price_series.min(),
    "价格波动率": price_series.std()
}

算法详解

EOS采用多重策略确保数据准确性：

1. 平滑算法

def _predict_ets(self, history: np.ndarray, seasonal_periods: int, hours: int) -> np.ndarray:
    """平滑季节算法"""
    clean_history = self._cap_outliers(history)
    model = ExponentialSmoothing(
        clean_history, 
        seasonal="add", 
        seasonal_periods=seasonal_periods
    ).fit()
    return model.forecast(hours)

2. 数据量自适应策略

根据历史数据量选择最优策略：

数据点数	策略	周期
>800	完整算法	168小时
168-800	简化算法	24小时
<168	中位数计算	无季节效应

3. 异常值处理

def _cap_outliers(self, data: np.ndarray, sigma: int = 2) -> np.ndarray:
    """基于标准差的价格异常值处理"""
    mean = data.mean()
    std = data.std()
    lower_bound = mean - sigma * std
    upper_bound = mean + sigma * std
    return data.clip(min=lower_bound, max=upper_bound)

高级功能

自定义数据导入

# 从JSON文件导入电价数据
import_config = ElecPriceImportCommonSettings(
    import_file_path="custom_prices.json",
    import_json='{"elecprice_marketprice_wh": [0.0003384, 0.0003318, 0.0003284]}'
)

# 从API响应创建电价记录
api_response = {
    "meta": {"start": "2024-01-01", "end": "2024-01-02"},
    "values": [
        {"start": "2024-01-01T00:00:00", "marketpriceEurocentPerKWh": 28.45},
        {"start": "2024-01-01T01:00:00", "marketpriceEurocentPerKWh": 26.32}
    ]
}

缓存机制

@cache_in_file(with_ttl="1 hour")
def _request_forecast(self) -> AkkudoktorElecPrice:
    """带缓存的API请求，减少外部调用"""
    # API请求逻辑...
    return response_data

性能优化建议

1. 数据更新策略

flowchart TD
    A[检查数据更新需求] --> B{需要更新?}
    B -->|是| C[调用API获取数据]
    B -->|否| D[使用缓存数据]
    C --> E[数据验证和处理]
    E --> F[异常值检测]
    F --> G[计算价格]
    G --> H[更新数据记录]
    D --> I[返回现有数据]

2. 内存管理

# 限制历史数据保留
config = PredictionCommonSettings(
    hours=48,          # 计算未来48小时
    historic_hours=168 # 保留过去7天数据
)

错误处理与监控

异常处理策略

try:
    prediction.update_data()
except ValueError as e:
    logger.error(f"电价数据更新失败: {e}")
    # 降级到备用数据源
    fallback_provider = ElecPriceImport()
    fallback_provider._update_data()
except requests.Timeout:
    logger.warning("API请求超时，使用缓存数据")

数据质量监控

def monitor_data_quality(price_series):
    """监控电价数据质量"""
    quality_metrics = {
        "数据完整性": len(price_series) / expected_count,
        "价格合理性": (price_series < 1.0).all(),  # 电价应小于1€/kWh
        "波动性": price_series.std() / price_series.mean()
    }
    return quality_metrics

实际应用案例

家庭能源优化

# 基于电价的用电优化
def optimize_energy_usage(prices, consumption_pattern):
    """根据电价优化用电时间"""
    low_price_hours = prices.nsmallest(8).index  # 选择8个最低电价时段
    optimized_schedule = {
        hour: "高能耗设备运行" if hour in low_price_hours else "低功耗模式"
        for hour in prices.index
    }
    return optimized_schedule

商业场景应用

# 工商业电价分析
def analyze_business_costs(prices, production_schedule):
    """分析生产成本与电价关系"""
    hourly_costs = prices * production_schedule
    total_cost = hourly_costs.sum()
    cost_savings = (prices.max() - prices.min()) * production_schedule.sum()
    
    return {
        "总电费成本": total_cost,
        "潜在节省": cost_savings,
        "最优生产时段": prices.nsmallest(6).index
    }

最佳实践总结

1. 多数据源备份：配置主备数据源确保服务连续性
2. 定期数据验证：监控数据质量并及时处理异常
3. 缓存策略优化：合理设置缓存时间减少API调用
4. 算法调优：根据数据特征选择合适的模型
5. 错误恢复机制：实现优雅降级和自动恢复

EOS的电价接口为能源优化系统提供了可靠的价格数据支撑，通过灵活的架构设计和智能的算法，能够满足从家庭到工商业等各种场景的电价数据需求。