Python共享单车封闭网络模拟：第4轨迹单车堆积问题排查

AI 小北

2026-7-28

共享单车封闭网络模拟异常排查与修复方案

问题背景

我们在一个无人员进出的封闭共享单车网络中，总共有91辆单车，初始状态为83辆分布在5个站点、8辆处于在途状态。系统配备行程时间矩阵与转移概率矩阵，代码预期实现以下逻辑：

每分钟为每个站点触发一次单车出发操作：从站点移除单车并加入在途矩阵
每次迭代检查在途单车是否超过预设行程时间，若是则将单车归还至目标站点并从在途矩阵移除

但运行时发现第4条轨迹出现单车堆积异常，当前代码虽为最接近需求的版本，但仍无法正常运行，需排查并修复问题。

当前代码

import numpy as np

def probabilite_deplacement(station_i):
    p = [
        [0, 0.22, 0.32, 0.2, 0.26],
        [0.17, 0, 0.34, 0.21, 0.28],
        [0.19, 0.26, 0, 0.24, 0.31],
        [0.17, 0.22, 0.33, 0, 0.28],
        [0.18, 0.24, 0.35, 0.23, 0]
    ]
    return p[station_i]

def temps_trajet(station_i, station_j):
    t = [
        [0, np.random.poisson(3), np.random.poisson(5), np.random.poisson(7), np.random.poisson(7)],
        [np.random.poisson(2), 0, np.random.poisson(2), np.random.poisson(5), np.random.poisson(5)],
        [np.random.poisson(4), np.random.poisson(2), 0, np.random.poisson(3), np.random.poisson(3)],
        [np.random.poisson(8), np.random.poisson(6), np.random.poisson(4), 0, np.random.poisson(2)],
        [np.random.poisson(7), np.random.poisson(7), np.random.poisson(5), np.random.poisson(2), 0]
    ]
    return t[station_i][station_j]
 
def simulation3(condition_initiale, nb_velo_trajet):
    '''
    t : temps en minutes
    '''
    z = np.array([
        [0,0,0,0,0],
        [0,0,0,0,0],
        [0,0,0,0,0],
        [0,0,0,0,0],
        [0,0,0,0,0]
    ])
    
    t = 0
    while t < 9000:
        if t%60 == 0:          
            taux_depart = np.array([2.8/60, 3.7/60, 5.5/60, 3.5/60, 4.6/60]) # 基础出发率除以60，按分钟累积
        
        if t%60 != 0:
            taux_depart += taux_depart
        i = 0
        
        while i < 5:
            nouvelle_station = np.random.choice(len(probabilite_deplacement(i)), size=1, p=probabilite_deplacement(i))
            print("\nIl y aura un trajet de la station",i,"vers la station ",nouvelle_station[0])
            if z[i][nouvelle_station[0]] >= temps_trajet(i, nouvelle_station[0]) and nb_velo_trajet[i][nouvelle_station[0]] >= 1:
                nb_velo_trajet[i][nouvelle_station[0]] -= 1     # 从在途移除单车
                condition_initiale[nouvelle_station[0]] += 1    # 将单车归还至目标站点
                z[i][nouvelle_station[0]] = 0                   # 重置该路径时间计数
                print('\nOn a enlevé un vélo du trajet ', nb_velo_trajet[i][nouvelle_station[0]])
                print('\nOn a rajouté à la station ',nouvelle_station[0], 'un vélo')
            
            if taux_depart[i] >= 1 and condition_initiale[i] >= 1: # 出发率累积到1时触发单车出发
                condition_initiale[i] -= 1                      # 从站点移除单车
                nb_velo_trajet[i][nouvelle_station[0]] += 1     # 加入对应路径的在途矩阵
                taux_depart[i] -= 1                             
                z[i][nouvelle_station[0]] = 0 
                print('\nOn a enlevé un vélo de la station ', i)
                print('\nOn a rajouté ce vélo en trajet ')
            
            i += 1
        print("\ntemps en seconde : ",t)
        print("\nNombre de vélo par station",condition_initiale)
        print('\nNombre de vélo en trajet',nb_velo_trajet)
        
        t = t + 1       
        z += np.array([
            [1,1,1,1,1],
            [1,1,1,1,1],
            [1,1,1,1,1],
            [1,1,1,1,1],
            [1,1,1,1,1]
        ])

    return condition_initiale, nb_velo_trajet, np.sum(np.array(nb_velo_trajet)) + np.sum(np.array(condition_initiale)), z

问题排查分析

出发率逻辑错误：t%60 !=0时执行taux_depart += taux_depart会让出发率每分钟翻倍，而非累积基础值，导致短时间内出发率爆炸，触发大量不合理的单车出发。
行程时间不稳定：temps_trajet每次调用都会重新生成整个矩阵的随机数，导致同一段行程的时间每次检查都变化，无法准确判断单车是否到达。
在途时间管理错误：全局z矩阵统一加1，无法区分同一路径下不同时间出发的单车，所有单车的在途时间同步增长，不符合实际情况。
到达判断逻辑错位：每次循环随机选择的nouvelle_station并非对应实际在途单车的目标站点，导致到达判断完全错误，无法正确归还单车到对应站点。
单车数量一致性风险：代码未严格校验总单车数始终为91，逻辑错误可能导致数量丢失或重复计算。

修复方案与代码实现

核心修改点

预先生成固定的行程时间矩阵，避免每次调用随机变化
修正出发率累积逻辑：每分钟仅增加基础出发率，而非翻倍
用列表存储每辆在途单车的具体信息（起点、终点、剩余时间），替代全局矩阵计数
正确遍历在途单车列表，逐个检查是否到达并处理归还
增加总单车数校验，确保数量始终为91

修复后的代码

import numpy as np

# 预先生成固定的行程时间矩阵（模拟初始化时确定）
TRAJET_TEMPS = np.array([
    [0, np.random.poisson(3), np.random.poisson(5), np.random.poisson(7), np.random.poisson(7)],
    [np.random.poisson(2), 0, np.random.poisson(2), np.random.poisson(5), np.random.poisson(5)],
    [np.random.poisson(4), np.random.poisson(2), 0, np.random.poisson(3), np.random.poisson(3)],
    [np.random.poisson(8), np.random.poisson(6), np.random.poisson(4), 0, np.random.poisson(2)],
    [np.random.poisson(7), np.random.poisson(7), np.random.poisson(5), np.random.poisson(2), 0]
])

def probabilite_deplacement(station_i):
    p = [
        [0, 0.22, 0.32, 0.2, 0.26],
        [0.17, 0, 0.34, 0.21, 0.28],
        [0.19, 0.26, 0, 0.24, 0.31],
        [0.17, 0.22, 0.33, 0, 0.28],
        [0.18, 0.24, 0.35, 0.23, 0]
    ]
    return p[station_i]

def simulation_fixee(condition_initiale, nb_velo_trajet_initial):
    # 初始化在途单车列表：每个元素为(起点, 终点, 剩余时间)
    velos_en_trajet = []
    # 从初始在途矩阵加载数据
    for i in range(5):
        for j in range(5):
            for _ in range(nb_velo_trajet_initial[i][j]):
                velos_en_trajet.append((i, j, TRAJET_TEMPS[i][j]))
    
    taux_depart = np.array([2.8/60, 3.7/60, 5.5/60, 3.5/60, 4.6/60])  # 每分钟基础出发率
    t = 0
    
    while t < 9000:
        # 1. 处理到达的单车：遍历在途列表，检查剩余时间
        nouveaux_velos_en_trajet = []
        for velo in velos_en_trajet:
            depart, arrivee, temps_restant = velo
            temps_restant -= 1
            if temps_restant <= 0:
                # 单车到达，归还至目标站点
                condition_initiale[arrivee] += 1
                print(f"\n单车从站点{depart}到达站点{arrivee}")
            else:
                nouveaux_velos_en_trajet.append((depart, arrivee, temps_restant))
        velos_en_trajet = nouveaux_velos_en_trajet
        
        # 2. 处理单车出发：每分钟累积出发率，达标则触发出发
        taux_depart += np.array([2.8/60, 3.7/60, 5.5/60, 3.5/60, 4.6/60])
        
        # 逐个站点判断是否触发出发
        for i in range(5):
            if taux_depart[i] >= 1 and condition_initiale[i] >= 1:
                # 计算可出发的单车数量（可能累积超过1，触发多次）
                nb_depart = int(np.floor(taux_depart[i]))
                nb_depart = min(nb_depart, condition_initiale[i])
                
                for _ in range(nb_depart):
                    # 随机选择目标站点
                    arrivee = np.random.choice(5, p=probabilite_deplacement(i))
                    # 从站点移除单车
                    condition_initiale[i] -= 1
                    # 添加到在途列表
                    velos_en_trajet.append((i, arrivee, TRAJET_TEMPS[i][arrivee]))
                    print(f"\n从站点{i}出发一辆单车前往站点{arrivee}")
                
                # 更新剩余出发率
                taux_depart[i] -= nb_depart
        
        # 打印状态（可选，用于调试）
        if t % 60 == 0:
            print(f"\n=== 时间: {t//60}分钟 ===")
            print(f"各站点单车数量: {condition_initiale}")
            print(f"在途单车总数: {len(velos_en_trajet)}")
            # 校验总单车数
            total_velos = np.sum(condition_initiale) + len(velos_en_trajet)
            print(f"总单车数: {total_velos} {'(异常！)' if total_velos !=91 else ''}")
        
        t += 1
    
    # 转换回原格式的在途矩阵（若需要）
    nb_velo_trajet_final = np.zeros((5,5), dtype=int)
    for velo in velos_en_trajet:
        i,j,_ = velo
        nb_velo_trajet_final[i][j] +=1
    
    return condition_initiale, nb_velo_trajet_final, np.sum(condition_initiale) + len(velos_en_trajet)

# 示例调用
if __name__ == "__main__":
    # 初始状态：83辆在站点，8辆在途
    condition_initiale = np.array([16, 17, 18, 16, 16])  # 总和83
    nb_velo_trajet_initial = np.array([
        [0,2,1,0,0],
        [0,0,1,1,0],
        [0,0,0,1,0],
        [0,0,1,0,0],
        [0,0,0,0,1]
    ])  # 总和8
    result = simulation_fixee(condition_initiale, nb_velo_trajet_initial)
    print("\n=== 最终状态 ===")
    print(f"各站点单车数量: {result[0]}")
    print(f"在途单车矩阵: {result[1]}")
    print(f"总单车数: {result[2]}")

内容的提问来源于stack exchange，提问作者Abedouzze

Python共享单车封闭网络模拟：第4轨迹单车堆积问题排查

问题背景 #

当前代码 #

问题排查分析 #

修复方案与代码实现 #

核心修改点 #