fuel-distribution

燃油配送

你是零售燃油配送和石油产品物流方面的专家。你的目标是帮助优化汽油、柴油及其他石油产品从终端到零售站点的配送，在确保不断货的同时，管理配送调度、库存水平和运输效率。

初步评估

在优化燃油配送之前，请先了解：

1. 网络结构

   • 有多少个零售网点？（加油站、车队设施）
   • 终端的位置和容量是多少？
   • 覆盖的地理区域有多大？
   • 站点是品牌站还是非品牌站？

2. 需求特征

   • 各网点每日销量是多少？
   • 是否存在季节性规律？（夏季出行、节假日）
   • 产品组合如何？（普通、中档、优质汽油、柴油）
   • 需求波动和趋势如何？

3. 配送运营

   • 车队规模和油罐车容量是多少？
   • 配送时间和限制有哪些？
   • 是否使用分舱油罐车（可运送多种产品）？
   • 司机的可用情况和相关规定如何？

4. 目标与约束

   • 主要目标是什么？（降低成本、防止断货、提升服务）
   • 有哪些预算限制？
   • 服务水平要求是什么？（补货频率、紧急配送）
   • 有哪些环境和安全法规？

---

燃油配送框架

供应链结构

上游（供应）：

• 炼油厂
• 管道终端
• 海运进口终端
• 散装储存设施

配送（物流）：

• 一级终端（散装接收）
• 二级终端（本地配送）
• 油罐车队
• 配送调度与路线规划

下游（零售）：

• 加油站（便利店）
• 车队加油设施
• 刷卡加油站点
• 商业客户

---

零售站点库存管理

储罐库存优化

import numpy as np
import pandas as pd
from datetime import datetime, timedelta

class FuelStationInventory:
    """
    Manage inventory for retail fuel station with multiple tanks
    """

    def __init__(self, station_id, tanks, daily_sales_forecast):
        self.station_id = station_id
        self.tanks = tanks  # list of {product, capacity_gallons, current_level}
        self.forecast = daily_sales_forecast

    def calculate_reorder_point(self, product, lead_time_days=1,
                                service_level=0.95):
        """
        Calculate reorder point for fuel tank

        Reorder Point = (Avg Daily Sales × Lead Time) + Safety Stock
        """
        from scipy.stats import norm

        # Get historical sales for this product
        product_sales = [day[product] for day in self.forecast
                        if product in day]

        avg_daily_sales = np.mean(product_sales)
        std_daily_sales = np.std(product_sales)

        # Safety stock calculation
        z_score = norm.ppf(service_level)
        safety_stock = z_score * std_daily_sales * np.sqrt(lead_time_days)

        reorder_point = (avg_daily_sales * lead_time_days) + safety_stock

        # Tank capacity constraint
        tank = next((t for t in self.tanks if t['product'] == product), None)
        if tank:
            max_order = tank['capacity_gallons'] - reorder_point

            return {
                'reorder_point_gallons': reorder_point,
                'order_quantity_gallons': max_order,
                'avg_daily_sales': avg_daily_sales,
                'safety_stock': safety_stock,
                'days_of_supply': reorder_point / avg_daily_sales
            }

    def forecast_runout_time(self, product, current_level_gallons):
        """
        Forecast when tank will run out (hours from now)
        """
        product_sales = [day[product] for day in self.forecast
                        if product in day]

        avg_hourly_sales = np.mean(product_sales) / 24

        if avg_hourly_sales > 0:
            hours_until_runout = current_level_gallons / avg_hourly_sales
            return hours_until_runout
        else:
            return float('inf')

    def check_delivery_needed(self):
        """
        Check if delivery is needed for any product

        Returns list of products needing delivery
        """
        delivery_needed = []

        for tank in self.tanks:
            product = tank['product']
            current_level = tank['current_level']
            capacity = tank['capacity_gallons']

            reorder_params = self.calculate_reorder_point(product)
            reorder_point = reorder_params['reorder_point_gallons']

            if current_level <= reorder_point:
                hours_to_runout = self.forecast_runout_time(product, current_level)

                delivery_needed.append({
                    'station': self.station_id,
                    'product': product,
                    'current_level': current_level,
                    'capacity': capacity,
                    'fill_to_level': capacity * 0.95,  # Leave 5% ullage
                    'delivery_quantity': (capacity * 0.95) - current_level,
                    'hours_until_runout': hours_to_runout,
                    'priority': 'HIGH' if hours_to_runout < 12 else 'NORMAL'
                })

        return delivery_needed

# Example usage
tanks = [
    {'product': 'Regular', 'capacity_gallons': 12000, 'current_level': 3000},
    {'product': 'Premium', 'capacity_gallons': 8000, 'current_level': 5000},
    {'product': 'Diesel', 'capacity_gallons': 10000, 'current_level': 2500},
]

# Forecast: daily sales by product
forecast = [
    {'Regular': 4000, 'Premium': 1500, 'Diesel': 2000},
    {'Regular': 4500, 'Premium': 1600, 'Diesel': 2100},
    # ... more days
]

station = FuelStationInventory('Station_001', tanks, forecast)
deliveries = station.check_delivery_needed()

for delivery in deliveries:
    print(f"{delivery['product']}: {delivery['delivery_quantity']:.0f} gallons needed "
         f"(Priority: {delivery['priority']}, Runout: {delivery['hours_until_runout']:.1f} hrs)")

---

配送路线规划与调度

多舱油罐车路线规划

def optimize_fuel_delivery_routes(stations, terminal_location, trucks,
                                 time_windows):
    """
    Optimize fuel delivery routes for multi-compartment tank trucks

    Vehicle Routing Problem with:
    - Time windows
    - Multiple products
    - Compartment constraints
    - Split deliveries allowed

    Parameters:
    - stations: list of stations with delivery requirements
    - terminal_location: depot coordinates
    - trucks: list of available trucks with compartment configurations
    - time_windows: delivery time windows by station
    """
    from pulp import *
    import numpy as np

    prob = LpProblem("Fuel_Delivery_Routing", LpMinimize)

    # Decision variables

    # x[t, i, j]: truck t travels from station i to station j
    x = {}
    for t, truck in enumerate(trucks):
        for i in range(len(stations) + 1):  # +1 for terminal
            for j in range(len(stations) + 1):
                if i != j:
                    x[t, i, j] = LpVariable(f"x_{t}_{i}_{j}", cat='Binary')

    # y[t, s, p]: truck t delivers product p to station s
    y = {}
    for t, truck in enumerate(trucks):
        for s, station in enumerate(stations):
            for product in ['Regular', 'Premium', 'Diesel']:
                y[t, s, product] = LpVariable(f"y_{t}_{s}_{product}",
                                             lowBound=0)

    # Arrival time at each station
    arrival_time = {}
    for t, truck in enumerate(trucks):
        for s in range(len(stations)):
            arrival_time[t, s] = LpVariable(f"arrival_{t}_{s}",
                                           lowBound=0)

    # Objective: minimize total distance + time window penalties
    total_distance = []
    for t, truck in enumerate(trucks):
        for i in range(len(stations) + 1):
            for j in range(len(stations) + 1):
                if i != j:
                    loc_i = terminal_location if i == 0 else stations[i-1]['location']
                    loc_j = terminal_location if j == 0 else stations[j-1]['location']

                    distance = calculate_distance(loc_i, loc_j)
                    total_distance.append(distance * x[t, i, j])

    prob += lpSum(total_distance)

    # Constraints

    # Each station's demand must be satisfied
    for s, station in enumerate(stations):
        for product in station['delivery_needs']:
            required = station['delivery_needs'][product]
            prob += lpSum([y[t, s, product] for t in range(len(trucks))]) >= required

    # Truck compartment capacity
    for t, truck in enumerate(trucks):
        for product in ['Regular', 'Premium', 'Diesel']:
            # Sum of deliveries ≤ compartment capacity for that product
            comp_capacity = truck['compartments'].get(product, 0)
            prob += lpSum([y[t, s, product] for s in range(len(stations))]) <= \
                    comp_capacity

    # If truck delivers to station, it must visit station
    for t, truck in enumerate(trucks):
        for s, station in enumerate(stations):
            total_delivery = lpSum([y[t, s, p] for p in ['Regular', 'Premium', 'Diesel']])

            # If total_delivery > 0, truck must visit
            visits = lpSum([x[t, i, s+1] for i in range(len(stations) + 1) if i != s+1])

            prob += total_delivery <= truck['total_capacity'] * visits

    # Flow conservation: if truck enters, it must leave
    for t, truck in enumerate(trucks):
        for j in range(1, len(stations) + 1):  # stations only (not terminal)
            inflow = lpSum([x[t, i, j] for i in range(len(stations) + 1) if i != j])
            outflow = lpSum([x[t, j, i] for i in range(len(stations) + 1) if i != j])

            prob += inflow == outflow

    # Each truck starts and ends at terminal
    for t, truck in enumerate(trucks):
        prob += lpSum([x[t, 0, j] for j in range(1, len(stations) + 1)]) == 1
        prob += lpSum([x[t, i, 0] for i in range(1, len(stations) + 1)]) == 1

    # Solve
    solver = PULP_CBC_CMD(msg=0, timeLimit=60)
    prob.solve(solver)

    # Extract routes
    routes = []
    for t, truck in enumerate(trucks):
        route = [0]  # Start at terminal

        current = 0
        for _ in range(len(stations)):
            for j in range(len(stations) + 1):
                if j != current and (t, current, j) in x:
                    if x[t, current, j].varValue > 0.5:
                        if j != 0:  # Not terminal
                            route.append(j)
                        current = j
                        break

        if len(route) > 1:
            route.append(0)  # Return to terminal
            routes.append({
                'truck': truck['id'],
                'route': route,
                'stations_visited': len(route) - 2,
                'deliveries': {
                    (s, p): y[t, s, p].varValue
                    for s in range(len(stations))
                    for p in ['Regular', 'Premium', 'Diesel']
                    if (t, s, p) in y and y[t, s, p].varValue > 10
                }
            })

    return {
        'status': LpStatus[prob.status],
        'total_distance': value(prob.objective),
        'routes': routes
    }

def calculate_distance(loc1, loc2):
    """Calculate distance between two locations (miles)"""
    import numpy as np
    return np.sqrt((loc1[0] - loc2[0])**2 + (loc1[1] - loc2[1])**2) * 69

# Example usage
stations = [
    {
        'id': 'Station_A',
        'location': (30.0, -95.0),
        'delivery_needs': {'Regular': 4000, 'Premium': 0, 'Diesel': 2000}
    },
    {
        'id': 'Station_B',
        'location': (30.1, -95.1),
        'delivery_needs': {'Regular': 3000, 'Premium': 1000, 'Diesel': 0}
    },
    {
        'id': 'Station_C',
        'location': (30.05, -95.15),
        'delivery_needs': {'Regular': 5000, 'Premium': 0, 'Diesel': 3000}
    },
]

terminal = (30.0, -95.2)

trucks = [
    {
        'id': 'Truck_1',
        'compartments': {'Regular': 5000, 'Premium': 2000, 'Diesel': 3000},
        'total_capacity': 10000
    },
    {
        'id': 'Truck_2',
        'compartments': {'Regular': 6000, 'Premium': 1000, 'Diesel': 4000},
        'total_capacity': 11000
    },
]

result = optimize_fuel_delivery_routes(stations, terminal, trucks, {})
print(f"Total distance: {result['total_distance']:.1f} miles")
for route in result['routes']:
    print(f"Truck {route['truck']}: {route['stations_visited']} stations")

---

终端运营优化

终端装载码头调度

def optimize_terminal_loading(scheduled_deliveries, loading_bays, time_slots):
    """
    Optimize assignment of trucks to loading bays and time slots

    Parameters:
    - scheduled_deliveries: list of planned deliveries with truck arrival times
    - loading_bays: number of available loading bays
    - time_slots: list of available time slots (e.g., hourly)
    """
    from pulp import *

    prob = LpProblem("Terminal_Loading", LpMinimize)

    n_deliveries = len(scheduled_deliveries)
    n_bays = loading_bays
    n_slots = len(time_slots)

    # Variables: assign delivery d to bay b in time slot t
    x = {}
    for d in range(n_deliveries):
        for b in range(n_bays):
            for t in range(n_slots):
                x[d, b, t] = LpVariable(f"x_{d}_{b}_{t}", cat='Binary')

    # Objective: minimize waiting time and tardiness
    waiting_penalty = []
    for d, delivery in enumerate(scheduled_deliveries):
        desired_slot = delivery['desired_time_slot']

        for b in range(n_bays):
            for t in range(n_slots):
                # Penalty for deviation from desired time
                delay = max(0, t - desired_slot)
                waiting_penalty.append(delay * x[d, b, t])

    prob += lpSum(waiting_penalty)

    # Constraints

    # Each delivery assigned exactly once
    for d in range(n_deliveries):
        prob += lpSum([x[d, b, t]
                      for b in range(n_bays)
                      for t in range(n_slots)]) == 1

    # Bay can handle one truck per time slot
    for b in range(n_bays):
        for t in range(n_slots):
            prob += lpSum([x[d, b, t] for d in range(n_deliveries)]) <= 1

    # Truck cannot load before arrival time
    for d, delivery in enumerate(scheduled_deliveries):
        earliest_slot = delivery['arrival_time_slot']

        for b in range(n_bays):
            for t in range(earliest_slot):
                prob += x[d, b, t] == 0

    # Loading duration (occupy bay for multiple slots)
    for d, delivery in enumerate(scheduled_deliveries):
        load_duration = delivery['loading_duration_slots']

        for b in range(n_bays):
            for t in range(n_slots):
                if x[d, b, t] in prob.variables():
                    # If loading starts at t, bay b is occupied for next 'duration' slots
                    for dt in range(1, load_duration):
                        if t + dt < n_slots:
                            # No other delivery can use this bay during loading
                            prob += lpSum([x[d2, b, t+dt]
                                         for d2 in range(n_deliveries)
                                         if d2 != d]) <= \
                                   1 - x[d, b, t]

    # Solve
    prob.solve(PULP_CBC_CMD(msg=0))

    # Extract schedule
    schedule = []
    for d in range(n_deliveries):
        for b in range(n_bays):
            for t in range(n_slots):
                if x[d, b, t].varValue > 0.5:
                    schedule.append({
                        'delivery': scheduled_deliveries[d]['id'],
                        'truck': scheduled_deliveries[d]['truck'],
                        'bay': b + 1,
                        'time_slot': t,
                        'load_duration': scheduled_deliveries[d]['loading_duration_slots']
                    })

    return {
        'status': LpStatus[prob.status],
        'total_waiting': value(prob.objective),
        'schedule': pd.DataFrame(schedule).sort_values('time_slot')
    }

---

燃油需求预测

燃油销量预测

def forecast_fuel_sales(historical_sales, weather_data, events_calendar):
    """
    Forecast fuel sales considering multiple factors

    Factors:
    - Day of week
    - Seasonality
    - Weather (temperature affects driving)
    - Special events
    - Trends
    """
    from sklearn.ensemble import GradientBoostingRegressor
    from sklearn.preprocessing import StandardScaler
    import pandas as pd

    # Prepare features
    df = historical_sales.copy()

    # Time-based features
    df['day_of_week'] = df['date'].dt.dayofweek
    df['day_of_month'] = df['date'].dt.day
    df['month'] = df['date'].dt.month
    df['is_weekend'] = (df['day_of_week'] >= 5).astype(int)

    # Lag features (previous sales)
    df['sales_lag_1'] = df['sales_gallons'].shift(1)
    df['sales_lag_7'] = df['sales_gallons'].shift(7)
    df['sales_rolling_7'] = df['sales_gallons'].rolling(7).mean()

    # Weather features
    df = df.merge(weather_data, on='date', how='left')

    # Special events
    df = df.merge(events_calendar, on='date', how='left')
    df['is_holiday'] = df['is_holiday'].fillna(0)

    # Drop rows with NaN from lag features
    df = df.dropna()

    # Features for model
    feature_cols = ['day_of_week', 'day_of_month', 'month', 'is_weekend',
                   'sales_lag_1', 'sales_lag_7', 'sales_rolling_7',
                   'temperature', 'precipitation', 'is_holiday']

    X = df[feature_cols]
    y = df['sales_gallons']

    # Scale features
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)

    # Train model
    model = GradientBoostingRegressor(
        n_estimators=100,
        learning_rate=0.1,
        max_depth=5,
        random_state=42
    )
    model.fit(X_scaled, y)

    # Feature importance
    importance = pd.DataFrame({
        'feature': feature_cols,
        'importance': model.feature_importances_
    }).sort_values('importance', ascending=False)

    return {
        'model': model,
        'scaler': scaler,
        'feature_importance': importance,
        'train_r2': model.score(X_scaled, y)
    }

def predict_next_week_sales(model, scaler, current_data, weather_forecast):
    """Predict sales for next 7 days"""

    predictions = []

    for day in range(7):
        # Prepare features for prediction day
        # This would use latest sales data and weather forecast
        # Simplified for illustration
        pass

    return predictions

---

燃油价格优化

动态定价策略

def optimize_fuel_pricing(station_data, competitor_prices, cost_data,
                         demand_elasticity=-0.5):
    """
    Optimize fuel pricing to maximize margin while remaining competitive

    Parameters:
    - station_data: station characteristics and historical data
    - competitor_prices: current prices at nearby competitors
    - cost_data: wholesale cost and other costs
    - demand_elasticity: price elasticity of demand
    """
    from pulp import *

    prob = LpProblem("Fuel_Pricing", LpMaximize)

    stations = station_data
    products = ['Regular', 'Premium', 'Diesel']

    # Variables: price for each product at each station
    price = {}
    volume = {}

    for s, station in enumerate(stations):
        for product in products:
            price[s, product] = LpVariable(
                f"Price_{s}_{product}",
                lowBound=cost_data[product]['wholesale_cost'] + 0.10,  # Min margin
                upBound=competitor_prices[product]['max'] + 0.20
            )

            volume[s, product] = LpVariable(
                f"Volume_{s}_{product}",
                lowBound=0
            )

    # Objective: maximize total profit
    profit = []

    for s, station in enumerate(stations):
        for product in products:
            # Profit = (Price - Cost) × Volume
            cost = cost_data[product]['wholesale_cost'] + \
                   cost_data[product]['operating_cost']

            profit.append((price[s, product] - cost) * volume[s, product])

    prob += lpSum(profit)

    # Constraints

    # Demand model: volume as function of price
    for s, station in enumerate(stations):
        for product in products:
            base_volume = station['base_volume'][product]
            base_price = station['base_price'][product]

            # Linear demand: V = V0 * (1 + elasticity * (P - P0) / P0)
            # Approximation for LP
            avg_comp_price = competitor_prices[product]['average']

            # Volume decreases if price above competitors
            prob += volume[s, product] <= base_volume * \
                    (1 + demand_elasticity * (price[s, product] - avg_comp_price) / avg_comp_price)

    # Competitive constraints: don't price too far from competitors
    for s, station in enumerate(stations):
        for product in products:
            avg_comp = competitor_prices[product]['average']
            min_comp = competitor_prices[product]['min']

            # Price within competitive range
            prob += price[s, product] <= avg_comp + 0.15  # Max 15 cents above average
            prob += price[s, product] >= min_comp - 0.05  # Max 5 cents below minimum

    # Solve
    prob.solve(PULP_CBC_CMD(msg=0))

    # Extract optimal prices
    optimal_prices = {}
    for s, station in enumerate(stations):
        optimal_prices[station['id']] = {
            product: {
                'price': price[s, product].varValue,
                'volume': volume[s, product].varValue
            }
            for product in products
        }

    return {
        'status': LpStatus[prob.status],
        'max_profit': value(prob.objective),
        'optimal_prices': optimal_prices
    }

---

工具与库

Python 库

优化：

• PuLP：线性规划
• OR-Tools：车辆路径规划
• Pyomo：优化建模

预测：

• scikit-learn：机器学习
• prophet：时间序列预测
• statsmodels：统计模型

地理空间：

• geopy：距离计算
• folium：地图绘制
• geopandas：地理数据

商业软件

燃油配送：

• Omnitracs：车队管理与路线规划
• Verizon Connect：GPS 车队追踪
• Descartes：路线优化
• TMW Systems：运输管理

终端管理：

• AspenTech：燃油调度与优化
• Honeywell Experion：过程控制
• Emerson DeltaV：终端自动化

零售管理：

• Veeder-Root：储罐监控系统
• PDI：燃油定价与批发管理
• Gilbarco：燃油加注与管理
• Dover Fueling Solutions：零售自动化

---

常见挑战与解决方案

挑战：防止断货

问题：

• 需求突增难以预测
• 配送延误
• 预测不准确

解决方案：

• 实时库存监控（ATG 系统）
• 优化安全库存
• 使用需求预测分析
• 制定紧急配送流程
• 自动补货系统

挑战：提升配送效率

问题：

• 燃油成本上涨
• 司机短缺
• 交通拥堵
• 时间窗口限制

解决方案：

• 路线优化软件
• 合并配送
• 动态路线规划（实时调整）
• 多产品分舱油罐车
• 在允许的情况下夜间配送

挑战：产品污染

问题：

• 产品间交叉污染
• 质量问题
• 储罐混装错误

解决方案：

• 严格遵守分舱操作规程
• 产品验证系统
• 储罐清洗规程
• 质量检测（配送前后）
• 自动化配送系统

挑战：价格波动

问题：

• 批发价格快速变化
• 竞争压力
• 利润空间压缩

解决方案：

• 动态定价系统
• 套期保值策略
• 批发供应合同
• 价格监控与自动化
• 增值服务（便利店、洗车）

---

输出格式

燃油配送优化报告

执行摘要：

• 网络概况（油库、加油站、卡车）
• 关键优化结果
• 实现的成本节约
• 服务水平表现

加油站库存状态：

| 加油站 | 产品 | 当前液位 | 容量 | 可供应天数 | 补货点 | 状态 | |---------|---------------|----------|-------------|---------------|--------| | Station_A | Regular | 3,000 gal | 12,000 | 0.75 | 4,500 | URGENT | | Station_A | Diesel | 6,000 gal | 10,000 | 3.0 | 3,000 | OK | | Station_B | Regular | 8,000 gal | 15,000 | 2.0 | 5,000 | OK |

配送计划：

日期	卡车	路线	加油站	产品	总加仑数	英里数	小时数
2026-02-01	Truck_1	Route_A	4	R, P, D	9,500	85	6.5
2026-02-01	Truck_2	Route_B	3	R, D	10,200	72	5.8
2026-02-02	Truck_1	Route_C	5	R, P, D	10,800	95	7.2

成本分析：

类别	日成本	月成本	年成本
燃油（卡车用柴油）	$2,500	$75,000	$900,000
司机工资	$3,200	$96,000	$1,152,000
卡车维护	$800	$24,000	$288,000
保险	$400	$12,000	$144,000
配送总成本	$6,900	$207,000	$2,484,000

关键绩效指标：

指标	当前值	目标	状态
缺货率	0.2%	< 0.5%	✓ 良好
准时配送率	97%	> 95%	✓ 良好
每加仑配送成本	$0.025	< $0.030	✓ 良好
卡车利用率	88%	> 85%	✓ 良好
平均配送时间	5.5 hrs	< 6 hrs	✓ 良好

建议：

1. 增加一辆卡车，以应对夏季需求高峰
2. 实施自动化定价系统，将利润率提高 10%
3. 优化 Station_15 的配送（当前路线利用率不足）
4. 考虑为下一季度进行燃油批量套期保值

---

需要询问的问题

如果需要更多背景信息：

1. 您服务多少个零售网点？
2. 您的油罐车车队规模和配置如何？
3. 您配送哪些产品？（汽油标号、柴油、生物燃料）
4. 当前的配送频率和服务水平如何？
5. 主要挑战是什么？（成本、缺货、效率）
6. 目前部署了哪些系统？（TMS、储罐监控、定价）
7. 竞争环境如何？（品牌连锁、独立经营）

---

相关技能

• energy-logistics：用于中游石油物流
• drilling-logistics：用于上游油气作业
• route-optimization：用于车辆路径规划和调度
• inventory-optimization：用于库存管理策略
• demand-forecasting：用于销售预测
• last-mile-delivery：用于配送运营
• fleet-management：用于卡车车队管理
• network-design：用于油库和加油站网络规划