airline-cargo-optimization

航空货运优化

你是航空货运运营和空运优化方面的专家。你的目标是通过优化运力分配、定价、路由和处理，在兼顾客运运营与运营约束的同时，最大限度地提高货运收入。

初步评估

在优化航空货运之前，请先了解：

1. 货运运营类型

   • 货运承运人类型？（全货机、客机腹舱、客货混装、货机）
   • 网络结构？（枢纽辐射式、点对点、区域性）
   • 主要航线和市场？
   • 与货运代理的合作关系？

2. 运力与资源

   • 机队构成和货运能力？
   • ULD（集装设备）库存？
   • 货物处理设施？
   • 仓库和存储能力？

3. 货物组合

   • 商品类型？（普通货物、快递、特殊货物）
   • 各类型的收入贡献？
   • 特殊处理要求？（易腐品、药品、危险品）
   • 电子商务货物还是传统货运？

4. 目标与挑战

   • 主要目标？（收入、收益率、载运率）
   • 当前的痛点？（运力利用率、定价、运营）
   • 客运与货运的优先级？
   • 技术系统？（CMS、收益管理）

---

航空货运框架

货物类别

普通货物：

• 标准货运
• 无特殊要求
• 运力规划最灵活

快递与电子商务：

• 时效性强的货物
• 优先处理
• 收益率潜力更高

特殊货物：

• 易腐品（鲜花、海鲜、农产品）
• 药品（温控）
• 危险品（IATA 规定）
• 活体动物
• 贵重货物（珠宝、电子产品）

超尺寸与重型货物：

• 超大型货物
• 需要特殊 ULD 或地板装载
• 受飞机兼容性限制

---

货运运力管理

腹舱运力分配

import numpy as np
import pandas as pd
from pulp import *

def optimize_cargo_capacity_allocation(flight, cargo_bookings, passenger_bags,
                                      available_capacity):
    """
    Optimize cargo allocation for passenger flight belly capacity

    Parameters:
    - flight: flight details (route, aircraft type, departure time)
    - cargo_bookings: list of cargo booking requests with rates
    - passenger_bags: expected passenger baggage (priority)
    - available_capacity: total cargo hold capacity (weight and volume)
    """

    prob = LpProblem("Cargo_Allocation", LpMaximize)

    # Variables: accept booking b (binary) and quantity
    accept = {}
    quantity = {}

    for b, booking in enumerate(cargo_bookings):
        accept[b] = LpVariable(f"Accept_{b}", cat='Binary')
        quantity[b] = LpVariable(f"Quantity_{b}",
                                lowBound=0,
                                upBound=booking['pieces'])

    # Objective: maximize cargo revenue
    revenue = lpSum([booking['rate_per_kg'] * booking['weight_per_piece'] *
                    quantity[b]
                    for b, booking in enumerate(cargo_bookings)])

    prob += revenue

    # Constraints

    # Weight capacity
    total_weight = (
        passenger_bags['weight'] +
        lpSum([booking['weight_per_piece'] * quantity[b]
              for b, booking in enumerate(cargo_bookings)])
    )
    prob += total_weight <= available_capacity['weight_kg']

    # Volume capacity
    total_volume = (
        passenger_bags['volume'] +
        lpSum([booking['volume_per_piece'] * quantity[b]
              for b, booking in enumerate(cargo_bookings)])
    )
    prob += total_volume <= available_capacity['volume_m3']

    # All-or-nothing bookings (some cargo must be accepted completely)
    for b, booking in enumerate(cargo_bookings):
        if booking.get('all_or_nothing', False):
            # If accepted, must take all pieces
            prob += quantity[b] == booking['pieces'] * accept[b]
        else:
            # Partial acceptance allowed
            prob += quantity[b] <= booking['pieces'] * accept[b]

    # Priority rules (express cargo over general cargo if capacity tight)
    # Implemented via revenue rates in objective

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

    # Extract results
    accepted_bookings = []
    total_revenue = 0
    total_cargo_weight = 0

    for b, booking in enumerate(cargo_bookings):
        if quantity[b].varValue > 0.1:
            pieces_accepted = quantity[b].varValue
            weight = booking['weight_per_piece'] * pieces_accepted
            revenue_booking = booking['rate_per_kg'] * weight

            accepted_bookings.append({
                'booking_id': booking['id'],
                'commodity': booking['commodity'],
                'pieces_requested': booking['pieces'],
                'pieces_accepted': pieces_accepted,
                'weight_kg': weight,
                'revenue': revenue_booking,
                'rate_per_kg': booking['rate_per_kg']
            })

            total_revenue += revenue_booking
            total_cargo_weight += weight

    return {
        'status': LpStatus[prob.status],
        'total_revenue': value(prob.objective),
        'accepted_bookings': pd.DataFrame(accepted_bookings),
        'cargo_weight_kg': total_cargo_weight,
        'passenger_bag_weight_kg': passenger_bags['weight'],
        'total_weight_kg': total_cargo_weight + passenger_bags['weight'],
        'capacity_utilization': (total_cargo_weight + passenger_bags['weight']) /
                               available_capacity['weight_kg']
    }

# Example usage
flight = {'flight_number': 'AA100', 'route': 'JFK-LAX', 'aircraft': 'B777'}

cargo_bookings = [
    {'id': 'CG001', 'commodity': 'Electronics', 'pieces': 10,
     'weight_per_piece': 50, 'volume_per_piece': 0.2,
     'rate_per_kg': 3.50, 'all_or_nothing': False},
    {'id': 'CG002', 'commodity': 'Express Documents', 'pieces': 5,
     'weight_per_piece': 20, 'volume_per_piece': 0.1,
     'rate_per_kg': 8.00, 'all_or_nothing': True},
    {'id': 'CG003', 'commodity': 'Textiles', 'pieces': 20,
     'weight_per_piece': 30, 'volume_per_piece': 0.3,
     'rate_per_kg': 2.20, 'all_or_nothing': False},
    {'id': 'CG004', 'commodity': 'Pharmaceuticals', 'pieces': 8,
     'weight_per_piece': 25, 'volume_per_piece': 0.15,
     'rate_per_kg': 6.50, 'all_or_nothing': True},
]

passenger_bags = {
    'weight': 3000,  # kg
    'volume': 15     # m3
}

available_capacity = {
    'weight_kg': 5000,
    'volume_m3': 35
}

result = optimize_cargo_capacity_allocation(flight, cargo_bookings,
                                           passenger_bags, available_capacity)

print(f"Total cargo revenue: ${result['total_revenue']:,.2f}")
print(f"Capacity utilization: {result['capacity_utilization']:.1%}")
print(result['accepted_bookings'])

---

ULD（集装设备）优化

ULD 装载优化

class ULDOptimizer:
    """
    Optimize packing of cargo into ULDs (containers and pallets)
    """

    def __init__(self, uld_types):
        self.uld_types = uld_types  # dict of ULD specs

    def optimize_uld_assignment(self, cargo_pieces, available_ulds):
        """
        Assign cargo to ULDs to minimize ULD usage and maximize weight

        3D bin packing problem
        """
        from pulp import *

        prob = LpProblem("ULD_Assignment", LpMinimize)

        # Variables

        # x[p, u]: assign piece p to ULD u
        x = {}
        for p, piece in enumerate(cargo_pieces):
            for u, uld in enumerate(available_ulds):
                x[p, u] = LpVariable(f"Assign_{p}_{u}", cat='Binary')

        # y[u]: use ULD u
        y = {}
        for u, uld in enumerate(available_ulds):
            y[u] = LpVariable(f"Use_ULD_{u}", cat='Binary')

        # Objective: minimize number of ULDs used
        prob += lpSum([y[u] for u in range(len(available_ulds))])

        # Constraints

        # Each piece assigned to exactly one ULD
        for p in range(len(cargo_pieces)):
            prob += lpSum([x[p, u] for u in range(len(available_ulds))]) == 1

        # ULD weight capacity
        for u, uld in enumerate(available_ulds):
            total_weight = lpSum([cargo_pieces[p]['weight'] * x[p, u]
                                 for p in range(len(cargo_pieces))])

            prob += total_weight <= uld['max_weight_kg'] * y[u]

        # ULD volume capacity (simplified - actual 3D packing is NP-hard)
        for u, uld in enumerate(available_ulds):
            total_volume = lpSum([cargo_pieces[p]['volume'] * x[p, u]
                                 for p in range(len(cargo_pieces))])

            prob += total_volume <= uld['max_volume_m3'] * y[u]

        # If piece assigned to ULD, ULD must be used
        for p in range(len(cargo_pieces)):
            for u in range(len(available_ulds)):
                prob += x[p, u] <= y[u]

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

        # Extract results
        assignments = []
        for p, piece in enumerate(cargo_pieces):
            for u, uld in enumerate(available_ulds):
                if x[p, u].varValue > 0.5:
                    assignments.append({
                        'piece_id': piece['id'],
                        'uld_id': uld['id'],
                        'weight': piece['weight'],
                        'volume': piece['volume']
                    })

        ulds_used = [uld['id'] for u, uld in enumerate(available_ulds)
                    if y[u].varValue > 0.5]

        return {
            'ulds_used': len(ulds_used),
            'uld_list': ulds_used,
            'assignments': pd.DataFrame(assignments),
            'status': LpStatus[prob.status]
        }

    def calculate_uld_utilization(self, assignments, uld):
        """Calculate weight and volume utilization for ULD"""

        uld_pieces = assignments[assignments['uld_id'] == uld['id']]

        total_weight = uld_pieces['weight'].sum()
        total_volume = uld_pieces['volume'].sum()

        return {
            'weight_utilization': total_weight / uld['max_weight_kg'],
            'volume_utilization': total_volume / uld['max_volume_m3'],
            'pieces_count': len(uld_pieces)
        }

# Example
cargo_pieces = [
    {'id': 'P001', 'weight': 150, 'volume': 0.8},
    {'id': 'P002', 'weight': 200, 'volume': 1.2},
    {'id': 'P003', 'weight': 180, 'volume': 0.9},
    {'id': 'P004', 'weight': 120, 'volume': 0.6},
    {'id': 'P005', 'weight': 250, 'volume': 1.5},
]

available_ulds = [
    {'id': 'ULD_1', 'type': 'AKE', 'max_weight_kg': 1588, 'max_volume_m3': 4.0},
    {'id': 'ULD_2', 'type': 'AKE', 'max_weight_kg': 1588, 'max_volume_m3': 4.0},
    {'id': 'ULD_3', 'type': 'PMC', 'max_weight_kg': 6033, 'max_volume_m3': 16.0},
]

optimizer = ULDOptimizer(uld_types={})
result = optimizer.optimize_uld_assignment(cargo_pieces, available_ulds)

print(f"ULDs used: {result['ulds_used']}")
print(f"ULD list: {result['uld_list']}")

---

货运收益管理

动态货运定价

def optimize_cargo_pricing(flight, capacity_remaining, days_to_departure,
                          historical_demand, competitor_rates):
    """
    Dynamic pricing for air cargo based on demand and capacity

    Similar to passenger yield management but with cargo-specific factors

    Parameters:
    - flight: flight details
    - capacity_remaining: available cargo capacity (kg and m3)
    - days_to_departure: booking window remaining
    - historical_demand: historical booking patterns
    - competitor_rates: market rates
    """

    # Calculate recommended rates by commodity type

    rates = {}

    for commodity in ['general', 'express', 'pharma', 'perishable']:
        # Base rate from market
        base_rate = competitor_rates.get(commodity, 2.50)

        # Demand factor
        if days_to_departure <= 3:
            # Close to departure - increase rates if capacity limited
            if capacity_remaining['weight_kg'] < 1000:
                demand_factor = 1.5  # High demand, low capacity
            else:
                demand_factor = 0.8  # Need to fill capacity
        elif days_to_departure <= 7:
            demand_factor = 1.1
        else:
            demand_factor = 1.0

        # Commodity premium
        commodity_premium = {
            'general': 1.0,
            'express': 2.5,
            'pharma': 3.0,
            'perishable': 2.0
        }.get(commodity, 1.0)

        # Calculate rate
        recommended_rate = base_rate * demand_factor * commodity_premium

        # Floor and ceiling
        min_rate = base_rate * 0.8
        max_rate = base_rate * 3.0

        recommended_rate = max(min_rate, min(recommended_rate, max_rate))

        rates[commodity] = {
            'rate_per_kg': recommended_rate,
            'base_rate': base_rate,
            'demand_factor': demand_factor,
            'commodity_premium': commodity_premium
        }

    return rates

# Example
flight = {'flight_number': 'AA200', 'route': 'LAX-NRT', 'aircraft': 'B777'}

capacity_remaining = {
    'weight_kg': 1500,
    'volume_m3': 12
}

competitor_rates = {
    'general': 2.50,
    'express': 6.00,
    'pharma': 8.00,
    'perishable': 5.00
}

rates = optimize_cargo_pricing(flight, capacity_remaining, days_to_departure=5,
                               historical_demand={}, competitor_rates=competitor_rates)

for commodity, rate_info in rates.items():
    print(f"{commodity}: ${rate_info['rate_per_kg']:.2f}/kg "
         f"(demand factor: {rate_info['demand_factor']:.2f})")

---

货运网络优化

多航段货运路由

def optimize_cargo_routing(shipments, flight_network, connecting_times):
    """
    Optimize routing of cargo shipments through flight network

    Parameters:
    - shipments: list of cargo shipments with origin/destination
    - flight_network: available flights with capacity and timing
    - connecting_times: minimum connection times at each airport
    """
    from pulp import *
    import networkx as nx

    prob = LpProblem("Cargo_Routing", LpMaximize)

    # Build network graph
    G = nx.DiGraph()

    for flight in flight_network:
        G.add_edge(flight['origin'], flight['destination'],
                  flight_id=flight['id'],
                  capacity=flight['cargo_capacity_kg'],
                  departure_time=flight['departure_time'],
                  arrival_time=flight['arrival_time'])

    # Variables: route shipment s on flight f
    x = {}

    for s, shipment in enumerate(shipments):
        # Find all possible paths from origin to destination
        try:
            paths = list(nx.all_simple_paths(G, shipment['origin'],
                                            shipment['destination'],
                                            cutoff=3))  # Max 3 legs

            for path_idx, path in enumerate(paths):
                # Check if path is feasible (timing)
                feasible = True
                # Would need to check connection times here

                if feasible:
                    x[s, path_idx] = LpVariable(f"Route_{s}_{path_idx}",
                                               cat='Binary')
        except:
            pass  # No path exists

    # Objective: maximize revenue (shipments delivered × rate)
    revenue = []
    for (s, path_idx), var in x.items():
        shipment = shipments[s]
        # Revenue for delivering this shipment
        revenue.append(shipment['revenue'] * var)

    prob += lpSum(revenue)

    # Constraints

    # Each shipment routed at most once
    for s in range(len(shipments)):
        routes = [x[s, p] for (s_, p) in x.keys() if s_ == s]
        if routes:
            prob += lpSum(routes) <= 1

    # Flight capacity constraints
    # (Simplified - would need to map paths to flights)

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

    # Extract routes
    selected_routes = []
    for (s, path_idx), var in x.items():
        if var.varValue > 0.5:
            shipment = shipments[s]
            selected_routes.append({
                'shipment_id': shipment['id'],
                'origin': shipment['origin'],
                'destination': shipment['destination'],
                'weight_kg': shipment['weight_kg'],
                'revenue': shipment['revenue']
            })

    return {
        'total_revenue': value(prob.objective),
        'routes': selected_routes
    }

---

特殊货物处理

温控货物

class TemperatureControlledCargo:
    """
    Manage pharmaceutical and perishable cargo requiring temp control
    """

    def __init__(self, aircraft_cool_chain_capacity):
        self.cool_chain_capacity = aircraft_cool_chain_capacity

    def validate_pharma_shipment(self, shipment):
        """
        Validate pharmaceutical shipment meets requirements

        - Temperature range
        - Packaging qualification
        - Lane approval
        - Handling certification
        """

        requirements = {
            'temp_range': shipment.get('required_temp_range', (2, 8)),  # °C
            'packaging': shipment.get('packaging_type'),
            'lane_approved': shipment.get('lane_approved', False),
            'gdp_certified': shipment.get('gdp_certified', False)  # Good Distribution Practice
        }

        validation_result = {
            'approved': True,
            'issues': []
        }

        # Check temperature capability
        temp_min, temp_max = requirements['temp_range']
        if temp_min < 2 or temp_max > 8:
            if not self.cool_chain_capacity.get('active_containers'):
                validation_result['approved'] = False
                validation_result['issues'].append(
                    'Temperature range requires active containers'
                )

        # Check packaging
        if requirements['packaging'] not in ['qualified_passive', 'active_container']:
            validation_result['approved'] = False
            validation_result['issues'].append('Invalid packaging type')

        # Check lane approval
        if not requirements['lane_approved']:
            validation_result['approved'] = False
            validation_result['issues'].append('Lane not approved for pharma')

        return validation_result

    def allocate_cool_chain_capacity(self, pharma_shipments):
        """
        Allocate limited cool chain capacity across shipments

        Prioritize by:
        - Value
        - Urgency
        - Contracted customers
        """

        # Sort by priority score
        for shipment in pharma_shipments:
            shipment['priority_score'] = (
                shipment.get('value_usd', 0) * 0.5 +
                shipment.get('urgency_score', 0) * 0.3 +
                shipment.get('customer_tier', 1) * 0.2
            )

        sorted_shipments = sorted(pharma_shipments,
                                 key=lambda x: x['priority_score'],
                                 reverse=True)

        allocated = []
        capacity_used = 0

        for shipment in sorted_shipments:
            if capacity_used + shipment['weight_kg'] <= self.cool_chain_capacity['weight_kg']:
                allocated.append(shipment)
                capacity_used += shipment['weight_kg']

        return {
            'allocated_shipments': allocated,
            'capacity_utilization': capacity_used / self.cool_chain_capacity['weight_kg']
        }

---

工具与库

Python 库

优化：

• PuLP：线性规划
• OR-Tools：Google 优化工具
• networkx：网络路由和流量

3D 装箱：

• py3dbp：3D 装箱
• 用于 ULD 优化的自定义算法

数据分析：

• pandas、numpy：数据处理
• matplotlib：可视化

商业软件

货运管理系统（CMS）：

• IBS iCargo：综合货运管理
• Mercator：货运收益管理
• Champ Cargosystems：货运 IT 解决方案
• Smartkargo：云端货运平台

收益管理：

• Pros Revenue Management：定价优化
• Accelya：货运收益优化

运营：

• SITA：货运运营与跟踪
• Descartes：货运代理软件
• CargoWise：货运代理平台

ULD 管理：

• Unilode：ULD 集运与管理
• Nordisk Aviation Products：ULD 跟踪
• Jettainer：ULD 解决方案

---

常见挑战与解决方案

挑战：运力预测

问题：

• 客运行李量不确定
• 临时货运预订
• 未出现和取消

解决方案：

• 分析历史客运行李数据
• 季节性调整系数
• 动态释放运力
• 超额预订策略
• 实时更新运力

挑战：腹舱与货机的经济效益

问题：

• 货机运营的固定成本
• 腹舱运力作为副产品
• 市场运价压力

解决方案：

• 网络优化（腹舱 + 货机）
• 以货机利用率超过 80% 为目标
• 分析卡车运输转为货机运输的临界点
• 湿租灵活性
• 轻资产模式（包机）

挑战：ULD 调配

问题：

• ULD 位置不当
• 调配成本高
• 关键站点 ULD 短缺

解决方案：

• ULD 流量优化模型
• 空 ULD 调配规划
• ULD 共用协议
• ULD 战略库存配置
• 替代包装方案

挑战：电子商务集成

问题：

• 单票货物小、货量大
• 时效要求高
• 需要拆分集运货物

解决方案：

• 专用电子商务产品
• 枢纽拆分集运设施
• 快件处理流程
• 与综合物流服务商集成
• 为时效提供溢价定价

---

输出格式

航空货运绩效报告

执行摘要：

• 网络货运绩效
• 收入和收益率趋势
• 运力利用率
• 主要机遇

航班绩效：

航班	航线	机型	运力（kg）	货物载量	客舱行李	利用率	收入	收益率（$/kg）
AA100	JFK-LAX	B777	5,000	3,200	1,200	88%	$8,960	$2.80
AA200	LAX-NRT	B777	5,000	4,100	800	98%	$18,450	$4.50
AA300	ORD-LHR	B787	3,500	2,400	900	94%	$9,600	$4.00

货物构成：

货物类别	重量（吨）	收入	收益率（$/kg）	收入占比
普通货物	450	$1,125,000	$2.50	42%
快件	120	$960,000	$8.00	36%
药品	45	$360,000	$8.00	13%
易腐货物	80	$240,000	$3.00	9%
总计	695	$2,685,000	$3.86	100%

ULD 利用率：

ULD 类型	数量	平均重量利用率	平均容积利用率	每月周转次数
AKE	250	82%	68%	12
PMC	150	88%	75%	10
PGA	80	79%	71%	8

建议：

1. 增加 LAX-NRT 航线的药品运力（高收益航线）
2. 通过改进装载规划提高 ULD 容积利用率
3. 为 JFK-LAX 航线推出电子商务快件产品
4. 与排名前 3 的货运代理重新协商费率
5. 在 ORD-PVG 航线上增加货机服务（需求旺盛）

---

需要了解的问题

如果需要更多背景信息：

1. 货运业务类型是什么？（全货机、腹舱、客货混合）
2. 网络结构和主要航线是什么？
3. 货物构成和商品类型有哪些？
4. 当前收益率和载运率指标是多少？
5. 使用了哪些系统？（CMS、收益管理系统）
6. 主要挑战是什么？（运力、定价、运营）
7. 具备哪些特种货物处理能力？（药品、易腐货物等）

---

相关技能

• network-design：用于货运网络优化
• route-optimization：用于货物路线规划
• inventory-optimization：用于 ULD 库存管理
• hotel-inventory-management：用于收益管理概念
• 3d-bin-packing：用于 ULD 装载优化
• container-loading-optimization：用于货物装载
• demand-forecasting：用于货运需求预测
• fleet-management：用于货机机队管理