pandapower在配网仿真中的实战应用

发布时间：2026/10/24 7:33:58
pandapower在配网仿真中的实战应用

摘要

随着分布式光伏大量接入中压配电网，配网潮流、电压越限、线路与配变过载风险显著增加，传统手工校核、简易估算方式难以支撑大批量馈线、多时间断面的新能源并网承载力评估。pandapower作为基于Python的开源电力仿真库，具备模型灵活构建、潮流计算稳定、适配自动化流程的优势，非常适合配电网批量仿真场景。本文结合湖南地区中压配网新能源接入评估项目实战代码，介绍基于pandapower搭建自动化配网仿真流程、模型构建、负荷时序赋值、光伏接入修正、指标统计与结果入库的完整实现方案，为配网承载力分析、并网预评估提供工程化落地思路。

关键词：pandapower；配电网仿真；分布式光伏；潮流计算；承载力评估；Python自动化

一、引言

中压配电网结构复杂、馈线数量多、拓扑多变，大量分布式光伏就近接入配变低压侧后，潮流双向流动，容易引发过电压、配变反向过载、线路载流量越限等问题。针对新能源报装批量评估业务，需要一套可循环执行、对接业务数据库、支持多馈线、多时间断面并行仿真的工具链。

主流商用仿真软件存在授权昂贵、二次开发接口受限、难以嵌入自动化业务流程等短板。pandapower依托pandas数据结构管理电网元件，底层调用成熟电力求解器，母线、线路、变压器、负荷、电源、开关等元件都支持代码化创建与参数自定义，天然适配数据库驱动的批量建模场景。本文的实战案例，直接从MySQL业务库读取配网拓扑、设备参数、时序负荷、待接入光伏台账，自动构建pandapower电网模型、执行潮流、统计越限指标并回写数据库，支撑区域新能源接入仿真研判。

二、整体方案架构

整套仿真系统采用"数据库驱动建模 + pandapower潮流内核 + 多进程并行计算 + 结果入库"的架构，整体流程如下：

  1. 参数接收：脚本通过命令行传入数据库连接信息、计算时段、区域编号、并发进程数，便于调度平台调用；

  2. 模型读取：从MySQL读取馈线基础信息、线路、配变、拓扑连接、开关、用户负荷、待接入光伏报装数据；

  3. 电网建模：按照拓扑关系自动生成母线、线路、变压器、负荷、外部电源、开关元件，缺失参数自动填充缺省标准类型，避免仿真报错；

  4. 时序负荷赋值：根据计算断面，给cuspoint用户负荷、配变负荷、光伏负荷注入时序有功、无功；

  5. 光伏功率修正：根据光伏报装容量，对对应配变负荷做功率加减，模拟光伏并网后的反向潮流；

  6. 潮流求解：调用pandapowerrunpp()执行潮流计算；

  7. 指标统计：提取母线电压、线路负载率、配变负载率，识别过电压、过载、反向过载样本；

  8. 结果入库：先删除该馈线+该时段历史结果，再将母线、用户节点、线路、变压器、馈线汇总指标批量写入结果表；

  9. 任务状态管理：在数据库记录算法运行状态、起止时间，便于运维监控。

为提升大批量馈线计算效率，方案使用multiprocessing进程池，对【馈线ID+时间断面】组合任务并行调度，充分利用服务器算力。

三、配网模型构建实战要点

pandapower电网由pp.create_empty_network()初始化空电网对象net，所有元件存储在net内部pandas表（net.bus、net.line、net.trafo、net.load、net.ext_grid、net.switch等），便于查询、修改与导出。工程建模中，标准类型（std_type）是核心，线路、变压器先注册标准参数模板，创建元件时引用模板，兼顾参数灵活性与代码简洁性。

在使用之前，需要导入pandapower算法包

import pandapower as pp

3.1 线路标准类型生成（读mtline参数，建std_type）

源码区分三类线路标准模板：虚拟连接线模板、缺省线路模板、真实设备线路模板。真实线路根据导线材质（铜CU/铝AL）自动计算单位长度电阻，电抗统一预设固定值，载流量从台账读取并转换为kA。台账缺失截面积、载流量、长度等关键字段时，自动降级使用default line缺省模板，防止建模中断。

# 虚拟连接线、缺省线路标准参数 line_data = {"c_nf_per_km": 0, "r_ohm_per_km": 0.01, "x_ohm_per_km": 0.01, "max_i_ka": 0.5, "type": "xnljx", "q_mm2": 240, "alpha": 0 } pp.create_std_type(net, line_data, "virtual line", element='line') pp.create_std_type(net, line_data, "switch line", element='line') # 真实线路：按材质计算单位长度电阻 for row in mtline.itertuples(): if getattr(row, 'conductor') == 'CU': if pd.isnull(getattr(row, 'jmj')): R = 31.2 / 500 else: R = 31.2 / getattr(row, 'jmj') elif getattr(row, 'conductor') == 'AL': if pd.isnull(getattr(row, 'jmj')): R = 18.5 / 500 else: R = 18.5 / getattr(row, 'jmj') else: R = 0.001 if pd.isnull(getattr(row, 'zll')): max_i_ka = 0.5 else: max_i_ka = getattr(row, 'zll') / 1000 line_data = {"r_ohm_per_km": R, "x_ohm_per_km": 0.1, "max_i_ka": max_i_ka, "type": getattr(row, 'linekind'), "q_mm2": getattr(row, 'jmj')} pp.create_std_type(net, line_data, getattr(row, 'id'), element='line')

开关节点之间采用虚拟短线（switch line）建模：pandapower原生开关支持直接连接母线，但工程中为方便观测开关支路电流，采用"母线-线路-开关"的l型开关模型，新建一条极短虚拟线路，再绑定开关元件，开关et参数设为'l'绑定线路。

3.2 配电变压器标准类型生成（读mttransformer参数，建std_type）

配变标准模板包含空载损耗pfe_kw、空载电流i0_percent、短路损耗对应的vkr_percent、短路电压vk_percent、额定容量、接线组别Dyn5、分接头配置等。台账参数校验容错逻辑：对i0、p0、uk、pk、sn做上下限判断，超出合理区间、空值时自动切换为default_trafo缺省模板。

# 缺省配变参数，台账缺失时使用 default_data = {"i0_percent": 0.5, "pfe_kw": 0.5, "vkr_percent": 1, "sn_mva": 0.4, "vn_lv_kv": 0.4, "vn_hv_kv": 10, "vk_percent": 4, "shift_degree": 150, "vector_group": 'Dyn5', "tap_side": 'hv', "tap_neutral": 0, "tap_min": -2, "tap_max": 2, "tap_step_percent": 2.5} pp.create_std_type(net, default_data, 'default_trafo', element='trafo') # 真实配变：参数合理性校验，越界则降级 for row in mttransformer.itertuples(): if pd.isnull(getattr(row, 'i0')) or getattr(row, 'i0') > 1 or getattr(row, 'i0') < 0.1: i0_percent = 0.5 else: i0_percent = getattr(row, 'i0') if pd.isnull(getattr(row, 'uk')) or getattr(row, 'uk') > 6 or getattr(row, 'uk') < 2: vk_percent = 4 else: vk_percent = getattr(row, 'uk') if pd.isnull(getattr(row, 'sn')) or pd.isnull(getattr(row, 'pk')) \ or getattr(row, 'pk') > 5 or getattr(row, 'pk') < 0.05: vkr_percent = 1 else: vkr_percent = getattr(row, 'pk') / getattr(row, 'sn') * 100 trafo_data = {"i0_percent": i0_percent, "pfe_kw": pfe_kw, "vkr_percent": vkr_percent, "sn_mva": sn_mva, "vn_lv_kv": 0.4, "vn_hv_kv": 10, "vk_percent": vk_percent, "vector_group": 'Dyn5'} pp.create_std_type(net, trafo_data, getattr(row, 'id'), element='trafo')

建模时根据两端母线额定电压自动区分高压侧、低压侧，保证create_transformer的hv_bus/lv_bus方向正确，避免变压器电压方向颠倒带来潮流异常。

3.3 母线与馈线电源点建模（读取mttopological表）

拓扑连接表connection1、connection2为节点编号，先去重批量生成母线，设置母线额定电压、上下限电压约束；再区分四类负荷：用户接入点cuspoint负荷、光伏站点pvstation负荷、配变低压侧负荷，初始化有功无功为0，后续由时序负荷表覆盖赋值。

# 拓扑连接点去重后批量建母线 cn1 = mttopological[['connection1', 'vol1']][mttopological['connection1'].notnull()] cn2 = mttopological[['connection2', 'vol2']][mttopological['connection2'].notnull()] cn1.columns = ['cn', 'vol'] cn2.columns = ['cn', 'vol'] topo_bus = pd.concat([cn1, cn2], ignore_index=True) cn = topo_bus.drop_duplicates() for row in cn.itertuples(): pp.create_bus(net, name=getattr(row, 'cn'), vn_kv=getattr(row, 'vol'), min_vm_pu=0.9, max_vm_pu=1.1) # 馈线电源点：拓扑中仅出现一次的节点判定为首端 grid = pd.merge(mtfeedgraph, switch, left_on=['cxkgbh'], right_on=['id']) for row in grid.itertuples(): bus1 = getattr(row, 'connection1') bus2 = getattr(row, 'connection2') if topo_bus['cn'].value_counts()[bus1] == 1: pp.create_ext_grid(net, net.bus[net.bus['name'] == bus1].index[0], name=getattr(row, 'id_x'), min_p_mw=0, max_p_mw=10, min_q_mvar=0, max_q_mvar=10) else: pp.create_ext_grid(net, net.bus[net.bus['name'] == bus2].index[0], name=getattr(row, 'id_x'))

3.4 线路建模（读取mttopological表，type为line）

线路标准类型注册完成后，按拓扑连接表将母线两两连通创建线路实例。真实线路通过拓扑表id关联设备台账，按参数完整度选择引用哪种标准类型：字段缺失或越界时降级用default line缺省模板；虚拟连接线统一用virtual line模板，长度设为极短值；开关节点同样新建极短虚拟线路作为开关支路。母线通过名称精确索引，从net.bus表取对应的索引号传入from_bus/to_bus。

# 新建真实线路 topo_line = mttopological[mttopological['type'] == 'line'] line = topo_line.merge(mtline, how='left', on='id') for row in line.itertuples(): bus1 = net.bus[net.bus['name'] == getattr(row, 'connection1')].index[0] bus2 = net.bus[net.bus['name'] == getattr(row, 'connection2')].index[0] # 参数缺失/越界时降级使用default line if pd.isnull(getattr(row, 'jmj')) or pd.isnull(getattr(row, 'zll')) \ or pd.isnull(getattr(row, 'length')): type, length = 'default line', 0.1 else: type, length = getattr(row, 'id'), getattr(row, 'length') / 1000 pp.create_line(net, name=getattr(row, 'id'), from_bus=bus1, to_bus=bus2, length_km=length, std_type=type) # 新建虚拟连接线 lj_line = mttopological[mttopological['type'] == 'xnljx'] for row in lj_line.itertuples(): bus1 = net.bus[net.bus['name'] == getattr(row, 'connection1')].index[0] bus2 = net.bus[net.bus['name'] == getattr(row, 'connection2')].index[0] pp.create_line(net, name=getattr(row, 'id'), from_bus=bus1, to_bus=bus2, length_km=0.01, std_type='virtual line') # 开关支路：新建极短开关线路，再由switch元件绑定 switchline = pp.create_line(net, name=getattr(row, 'id'), from_bus=bus1, to_bus=bus2, length_km=0.01, std_type='switch line')

开关建模采用"母线-线路-开关"的l型模型：先创建开关虚拟线路，再调用pp.create_switch，et参数设为'l'绑定该线路，初始状态closed=True合闸。相比直接把开关挂在母线上，这种方式能在结果表中观测到开关支路的电流与负载，便于统计开关潮流。

3.5 变压器建模（读取mttopological表，type为trafo）

变压器通过拓扑表id关联配变台账，创建实例时同样先判断关键参数（空载损耗p0、空载电流i0、短路损耗pk、短路电压uk）是否完整，缺失则降级引用default_trafo缺省模板。与线路不同，变压器是两端口元件，必须区分高压侧hv_bus与低压侧lv_bus：源码依据拓扑两端母线电压vol1、vol2的大小关系判定，电压高的一侧作为高压侧，避免变压器高低压侧颠倒导致潮流方向错误。

topo_trafo = mttopological[mttopological['type'] == 'trafo'] trafo = topo_trafo.merge(mttransformer, how='left', on='id') for row in trafo.itertuples(): bus1 = net.bus[net.bus['name'] == getattr(row, 'connection1')].index[0] bus2 = net.bus[net.bus['name'] == getattr(row, 'connection2')].index[0] # 关键参数缺失时降级default_trafo if pd.isnull(getattr(row, 'p0')) or pd.isnull(getattr(row, 'i0')) \ or pd.isnull(getattr(row, 'pk')) or pd.isnull(getattr(row, 'uk')): typename = 'default_trafo' else: typename = getattr(row, 'id') # 按母线电压区分高/低压侧，保证方向正确 if getattr(row, 'vol1') > getattr(row, 'vol2'): pp.create_transformer(net, name=getattr(row, 'id'), hv_bus=bus1, lv_bus=bus2, std_type=typename) else: pp.create_transformer(net, name=getattr(row, 'id'), hv_bus=bus2, lv_bus=bus1, std_type=typename)

配变低压侧通常为0.4kV，建模后在低压母线上生成配变负荷（type='trafo'），作为配变容量与负载率统计的挂载点；配变本身通过res_trafo结果表输出高低压侧功率，支撑配变正向/反向过载判断。

四、时序负荷与分布式光伏接入处理

4.1 时序负荷赋值

独立开发update_mv_load函数，读取pp_mv_load时序负荷表，按moment时间断面筛选数据，通过name关联net.load表，将kW、kvar转换为MW、MVar写入p_mw、q_mvar。pandapower所有潮流变量基准单位为MW/Mvar/kV，单位转换是高频出错点，工程中必须统一换算。

def update_mv_load(net, mvload, moment): if len(mvload) == 0: return net mvload = mvload[mvload['moment'] == moment] mvload['p'] = mvload['p'].astype(float) mvload['q'] = mvload['q'].astype(float) newload = pd.merge(net.load, mvload, left_on=['name'], right_on=['id'], how='left') newload.fillna(0, inplace=True) # kW -> MW，统一单位 net.load['p_mw'] = newload['p'] / 1000 net.load['q_mvar'] = newload['q'] / 1000 return net

4.2 分布式光伏并网模拟

从project_mv_powerplant光伏报装表读取待接入项目，通过blg_trafo_id绑定对应配变负荷节点：类型2光伏接入抵消配变原有负荷，负荷有功减去光伏容量；其余光伏类型叠加出力，负荷有功增加光伏容量。容量字段原始单位为W，除以1e6转为MW，和pandapower内部单位对齐。

if len(newproject) > 0: newproject['capacity'] = newproject['capacity'].astype(float) for row in newproject.itertuples(): if getattr(row, 'type') == 2: # 光伏接入：抵消配变原有负荷 net.load.loc[(net.load['name'] == getattr(row, 'blg_trafo_id')), 'p_mw'] = \ net.load.loc[(net.load['name'] == getattr(row, 'blg_trafo_id')), 'p_mw'] \ - getattr(row, 'capacity') / 1000000 else: # 其他新能源类型：叠加出力 net.load.loc[(net.load['name'] == getattr(row, 'blg_trafo_id')), 'p_mw'] = \ net.load.loc[(net.load['name'] == getattr(row, 'blg_trafo_id')), 'p_mw'] \ + getattr(row, 'capacity') / 1000000

这种基于负荷节点修正的方式适合快速批量评估，适合大量报装项目预校核；精细化仿真可改用pp.create_sgen静态发电机元件。

五、潮流计算、越限指标统计与入库

调用pp.runpp(net)执行牛顿-拉夫逊潮流求解，求解成功后读取res系列结果表（res_bus、res_line、res_trafo、res_load、res_ext_grid），按业务规则统计越限标记。

try: pp.runpp(net) blg_subarea_id = net.ext_grid['blg_subarea_id'].values[0] # 母线电压越限：vm_pu > 1.07 标记过电压 net.res_bus['name'] = net.bus['name'] plan_mv_connection[['connection_id', 'voltage_rate']] = \ net.res_bus[['name', 'vm_pu']] plan_mv_connection.loc[plan_mv_connection['voltage_rate'] > 1.07, 'overvoltage'] = 1 plan_mv_connection.loc[plan_mv_connection['voltage_rate'] <= 1.07, 'overvoltage'] = 0 # 线路负载率越限：loading_percent > 80 标记过载 net.res_line['max_i_ka'] = net.line['max_i_ka'] * 1000 plan_mv_line['load_rate'] = net.res_line['loading_percent'] plan_mv_line.loc[plan_mv_line['load_rate'] > 80, 'overload'] = 1 # 配变负载率：>80正向过载，<-80反向过载 plan_mv_transformer['load_rate'] = net.res_trafo['loading_percent'] plan_mv_feeder['overload_zheng_trans_count'] = \ (plan_mv_transformer['load_rate'] > 80).sum() plan_mv_feeder['overload_fan_trans_count'] = \ (plan_mv_transformer['load_rate'] < -80).sum() except Exception as e: print(f"{feederid}潮流计算失败: {e}")

入库策略采用先删后写：对同一馈线、同一时刻、同一区域，先删除历史仿真结果，再追加写入，避免重复脏数据；使用SQLAlchemy引擎批量to_sql写入，相比单行insert大幅提升入库速度。增加全局异常捕获，单条馈线潮流发散、拓扑错误、索引报错不会导致整个批次任务崩溃，仅打印失败馈线信息，其余馈线继续计算。

# 先删除该馈线该时段历史结果 sql1 = "delete from plan_mv_feeder where feeder_id = %s and blg_area_id = %s and moment = %s" sql2 = "delete from plan_mv_connection where blg_feeder_id = %s and blg_area_id = %s and moment = %s" cursor.execute(sql1, (feederid, BLG_AREA_ID, moment)) cursor.execute(sql2, (feederid, BLG_AREA_ID, moment)) db.commit() # SQLAlchemy批量追加写入 engine = create_engine('mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset={charset}' .format(user=USER, password=parse.quote_plus(PASSWORD), host=HOST, port=PORT, database=DATABASE, charset=CHARSET)) plan_mv_feeder = plan_mv_feeder.drop_duplicates(subset=['feeder_id'], keep='first') plan_mv_feeder.to_sql(name='plan_mv_feeder', con=engine, if_exists='append', index=False)