PPC 电力交易引擎
电价预测 | 充放电优化 | 报价 | 偏差考核
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
PPC 电力交易引擎
功能: 电价预测 / 充放电策略优化 / 报价生成 / 偏差考核
对接: 电力交易中心API + EMS调度
"""
import json, requests, logging
from datetime import datetime, timedelta
from typing import List, Dict
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("PPC")
class PPCTrader:
"""电力交易引擎"""
def __init__(self, center_url, ess_id="ESS-001"):
self.url = center_url
self.ess_id = ess_id
self.session = requests.Session()
self.session.headers.update({"Content-Type": "application/json"})
def fetch_price_forecast(self) -> List[Dict]:
"""获取未来24h电价预测"""
try:
resp = self.session.get(f"{self.url}/api/price/forecast", timeout=10)
data = resp.json()
return data.get("prices", [])
except Exception as e:
log.error("电价获取失败: %s", e)
return [{"hour": h, "price": 1.08 if (9<=h<=12 or 14<=h<=19) else 0.35 if (h>=23 or h<=7) else 0.66,
"type": "peak" if (9<=h<=12 or 14<=h<=19) else "valley" if (h>=23 or h<=7) else "flat"}
for h in range(24)]
def optimize_schedule(self, bess_kwh, soc, prices) -> List[Dict]:
"""生成24h充放电计划"""
schedule = []
current_soc = soc
power_max = bess_kwh * 0.5
for p in prices:
hour = p["hour"]
price = p["price"]
if price <= 0.40 and current_soc < 90:
charge_kwh = min(power_max, (90 - current_soc) / 100 * bess_kwh)
schedule.append({"hour": hour, "action": "charge", "power": charge_kwh, "soc_before": current_soc})
current_soc += charge_kwh / bess_kwh * 100
elif price >= 0.80 and current_soc > 20:
discharge_kwh = min(power_max, (current_soc - 20) / 100 * bess_kwh)
schedule.append({"hour": hour, "action": "discharge", "power": discharge_kwh, "soc_before": current_soc})
current_soc -= discharge_kwh / bess_kwh * 100
else:
schedule.append({"hour": hour, "action": "idle", "power": 0, "soc_before": current_soc})
return schedule
def generate_bid(self, schedule) -> Dict:
"""生成报价单 (日前市场)"""
today = datetime.now().strftime("%Y-%m-%d")
bid = {"ess_id": self.ess_id, "date": today, "plan": []}
for s in schedule:
bid["plan"].append({
"hour": s["hour"],
"power_kw": s["power"],
"direction": 1 if s["action"] == "discharge" else (-1 if s["action"] == "charge" else 0)
})
return bid
def submit_bid(self, bid) -> Dict:
"""向交易中心提交报价"""
try:
resp = self.session.post(f"{self.url}/api/bid/submit", json=bid, timeout=10)
result = resp.json()
log.info("报价提交成功: %s", result.get("bid_id"))
return result
except Exception as e:
log.error("报价提交失败: %s", e)
return {"error": str(e)}
def calc_deviation(self, planned_kw, actual_kw) -> Dict:
"""偏差考核计算"""
if planned_kw == 0:
return {"deviation_pct": 0, "penalty": 0}
dev = abs(actual_kw - planned_kw) / abs(planned_kw) * 100
penalty = 0
if dev > 5:
penalty = (dev - 5) * 0.1
return {"deviation_pct": round(dev, 2), "penalty": round(penalty, 2)}
def run_daily(self, bess_kwh, soc):
"""每日交易流程"""
prices = self.fetch_price_forecast()
log.info("获取电价: 峰=%.2f 谷=%.2f", max(p["price"] for p in prices), min(p["price"] for p in prices))
schedule = self.optimize_schedule(bess_kwh, soc, prices)
bid = self.generate_bid(schedule)
result = self.submit_bid(bid)
return {"schedule": schedule, "bid": bid, "result": result}
if __name__ == "__main__":
ppc = PPCTrader("http://grid-center.example.com")
result = ppc.run_daily(bess_kwh=5000, soc=85)
print(json.dumps(result, indent=2, ensure_ascii=False))
BMS 数据采集程序
CAN J1939 | SOC/SOH | 告警 | Modbus
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
BMS 电池数据采集程序
功能: CAN J1939解析 / SOC/SOH估算 / 告警管理 / Modbus转发
硬件: STM32F407 + CAN2.0B + RS485
"""
import struct, time, logging
from collections import deque
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("BMS")
# J1939 PGN定义
PGN_MAP = {
0xE001: {"name": "电芯电压温度", "fields": ["V_min", "V_max", "T_min", "T_max", "V_avg"]},
0x1806: {"name": "充电参数", "fields": ["I_charge", "V_charge", "SOC", "SOH"]},
0x1807: {"name": "放电参数", "fields": ["I_discharge", "V_discharge", "P_discharge", "T_batt"]},
0x1808: {"name": "告警状态", "fields": ["alarm_word", "protect_word", "fault_code", "reserve"]},
0x1809: {"name": "均衡状态", "fields": ["balance_flag", "V_diff", "cycle_count", "capacity_ah"]},
0x1810: {"name": "绝缘检测", "fields": ["R_iso_pos", "R_iso_neg", "status", "reserve"]},
}
class BMSCollector:
"""BMS数据采集与解析"""
def __init__(self):
self.data = {
"soc": 50.0, "soh": 100.0,
"v_min": 3.20, "v_max": 3.35, "v_avg": 3.28,
"t_min": 25.0, "t_max": 35.0,
"i_charge": 0, "i_discharge": 0,
"p_charge": 0, "p_discharge": 0,
"alarm": 0, "protect": 0, "fault": 0,
"cycle_count": 0, "capacity_ah": 100,
"r_iso_pos": 1000, "r_iso_neg": 1000
}
self.history = deque(maxlen=8640)
self.alarms = []
def parse_can_frame(self, can_id, data):
"""解析CAN帧"""
pgn = (can_id >> 8) & 0x3FFFF
source = can_id & 0xFF
if pgn not in PGN_MAP:
return
if pgn == 0xE001 and len(data) >= 8:
self.data["v_min"] = struct.unpack(">H", data[0:2])[0] / 1000
self.data["v_max"] = struct.unpack(">H", data[2:4])[0] / 1000
self.data["t_min"] = data[4] / 2 - 40
self.data["t_max"] = data[5] / 2 - 40
self.data["v_avg"] = (self.data["v_min"] + self.data["v_max"]) / 2
elif pgn == 0x1806 and len(data) >= 8:
self.data["i_charge"] = struct.unpack(">H", data[0:2])[0] / 10
self.data["v_charge"] = struct.unpack(">H", data[2:4])[0] / 10
self.data["soc"] = data[4] / 2
self.data["soh"] = data[5] / 2
elif pgn == 0x1807 and len(data) >= 8:
self.data["i_discharge"] = struct.unpack(">H", data[0:2])[0] / 10
self.data["p_discharge"] = struct.unpack(">H", data[4:6])[0]
elif pgn == 0x1808 and len(data) >= 4:
self.data["alarm"] = data[0]
self.data["protect"] = data[1]
self.data["fault"] = data[2]
elif pgn == 0x1809 and len(data) >= 8:
self.data["cycle_count"] = struct.unpack(">H", data[4:6])[0]
self.data["capacity_ah"] = struct.unpack(">H", data[6:8])[0] / 10
elif pgn == 0x1810 and len(data) >= 6:
self.data["r_iso_pos"] = struct.unpack(">H", data[0:2])[0]
self.data["r_iso_neg"] = struct.unpack(">H", data[2:4])[0]
def check_alarms(self):
"""告警检测"""
alarms = []
if self.data["v_max"] > 3.65:
alarms.append("OVERVOLT_MAX")
if self.data["v_min"] < 2.80:
alarms.append("UNDERVOLT_MIN")
if self.data["t_max"] > 55:
alarms.append("OVERTEMP")
if self.data["t_min"] < 0:
alarms.append("UNDERTEMP")
if self.data["soc"] < 10:
alarms.append("SOC_LOW")
if self.data["soc"] > 95:
alarms.append("SOC_HIGH")
if self.data["r_iso_pos"] < 500 or self.data["r_iso_neg"] < 500:
alarms.append("ISO_LOW")
self.alarms = alarms
return alarms
def to_modbus_registers(self):
"""转换为Modbus寄存器 (供EMS读取)"""
regs = [
int(self.data["soc"] * 10),
int(self.data["v_min"] * 1000),
int(self.data["v_max"] * 1000),
int((self.data["t_max"] + 40) * 2),
int(self.data["i_charge"] * 10),
int(self.data["i_discharge"] * 10),
self.data["alarm"],
self.data["protect"],
self.data["cycle_count"],
int(self.data["capacity_ah"] * 10),
self.data["r_iso_pos"],
self.data["r_iso_neg"],
]
return regs
def get_status(self):
"""获取BMS状态摘要"""
return {
"soc": self.data["soc"],
"soh": self.data["soh"],
"v_range": "{:.3f}~{:.3f}V".format(self.data["v_min"], self.data["v_max"]),
"t_range": "{:.1f}~{:.1f}C".format(self.data["t_min"], self.data["t_max"]),
"alarms": self.alarms,
"cycles": self.data["cycle_count"]
}
if __name__ == "__main__":
bms = BMSCollector()
bms.parse_can_frame(0x18FE0001, bytes([0x41,0x83,0x43,0x87,0x3C,0x46,0,0]))
bms.parse_can_frame(0x18060001, bytes([0x01,0xF4,0x01,0x2C,0xA2,0xC8,0,0]))
alarms = bms.check_alarms()
print("BMS状态:", bms.get_status())
print("告警:", alarms)
print("Modbus寄存器:", bms.to_modbus_registers())
