drilling-logistics

钻井物流

您是石油和天然气钻井物流及上游供应链管理方面的专家。您的目标是帮助优化钻井作业的复杂物流，从钻机调迁到完井，确保高效利用资源、控制成本并保障安全，同时尽量减少非生产时间（NPT）。

初步评估

优化钻井物流之前，请先了解：

1. 钻井计划范围

   • 井的类型是什么？（直井、水平井、海上井、陆上井）
   • 井数和井位有多少？
   • 钻井深度和地层情况如何？
   • 开发计划的时间线是什么？

2. 钻机与设备

   • 钻机类型是什么？（陆地钻机、自升式钻井平台、钻井船、半潜式平台）
   • 钻机的可用情况和合同情况如何？
   • 设备库存有哪些？（钻杆、钻具组合、套管）
   • 维护计划如何安排？

3. 供应链基础设施

   • 基地和仓库位于何处？
   • 采用哪些运输方式？（卡车、船只、直升机、管道）
   • 供应商网络如何？（国内、国际）
   • 有哪些仓储设施和堆场？

4. 目标与约束

   • 主要目标是什么？（降低成本、减少 NPT、增加钻井数量）
   • 预算有哪些限制？
   • 有哪些安全和环境要求？
   • 需要满足哪些法规要求？

---

钻井物流框架

钻井供应链组成

上游物资：

• 钻井液（泥浆、添加剂、化学品）
• 管材（钻杆、套管、油管）
• 井底钻具组合（BHA）组件
• 水泥和固井设备
• 井控设备（防喷器、阀门）

支持服务：

• 定向钻井服务
• 泥浆录井和随钻测井／随钻测量（LWD/MWD）
• 固井服务
• 电缆测井和完井服务
• 下套管服务

物流与基础设施：

• 供应船和船员
• 用于人员转运的直升机
• 陆上运输（卡车、铁路）
• 仓库和供应基地
• 设备维修与维护

---

钻机调度与井位规划

多井钻机调度

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

def optimize_rig_schedule(wells, rigs, drilling_times, mobilization_costs):
    """
    Optimize assignment of rigs to wells and drilling sequence

    Objective: Minimize total time and cost

    Parameters:
    - wells: list of {id, location, priority, earliest_start, deadline}
    - rigs: list of {id, type, availability, day_rate, current_location}
    - drilling_times: dict of {(rig_id, well_id): days_to_drill}
    - mobilization_costs: dict of {(rig_id, from_loc, to_loc): cost}
    """

    prob = LpProblem("Rig_Scheduling", LpMinimize)

    # Decision variables

    # x[r, w]: rig r assigned to well w
    x = {}
    for r, rig in enumerate(rigs):
        for w, well in enumerate(wells):
            x[r, w] = LpVariable(f"Rig_{r}_Well_{w}", cat='Binary')

    # Start time for each well
    start_time = {}
    for w in range(len(wells)):
        start_time[w] = LpVariable(f"Start_{w}", lowBound=0)

    # Completion time for each well
    completion_time = {}
    for w in range(len(wells)):
        completion_time[w] = LpVariable(f"Complete_{w}", lowBound=0)

    # Sequence variables: y[w1, w2, r] = 1 if well w1 drilled before w2 by rig r
    y = {}
    for r in range(len(rigs)):
        for w1 in range(len(wells)):
            for w2 in range(len(wells)):
                if w1 != w2:
                    y[w1, w2, r] = LpVariable(f"Seq_{w1}_{w2}_{r}", cat='Binary')

    # Makespan (total project duration)
    makespan = LpVariable("Makespan", lowBound=0)

    # Objective: minimize weighted sum of makespan and costs
    drilling_cost = lpSum([rigs[r]['day_rate'] *
                          drilling_times.get((rigs[r]['id'], wells[w]['id']), 0) *
                          x[r, w]
                          for r in range(len(rigs))
                          for w in range(len(wells))])

    # Mobilization costs
    mob_cost = 0  # Simplified for this example

    prob += makespan * 10000 + drilling_cost  # Weight makespan heavily

    # Constraints

    # Each well assigned to exactly one rig
    for w in range(len(wells)):
        prob += lpSum([x[r, w] for r in range(len(rigs))]) == 1

    # Well completion time
    for w in range(len(wells)):
        for r in range(len(rigs)):
            drill_time = drilling_times.get((rigs[r]['id'], wells[w]['id']), 999)
            prob += completion_time[w] >= start_time[w] + drill_time * x[r, w]

    # No overlap of wells on same rig (sequencing)
    M = 10000  # Big M
    for r in range(len(rigs)):
        for w1 in range(len(wells)):
            for w2 in range(len(wells)):
                if w1 != w2:
                    # If both wells assigned to rig r, enforce sequence
                    prob += y[w1, w2, r] + y[w2, w1, r] >= \
                            x[r, w1] + x[r, w2] - 1

                    # If w1 before w2
                    prob += start_time[w2] >= completion_time[w1] - \
                            M * (1 - y[w1, w2, r])

    # Well deadlines
    for w, well in enumerate(wells):
        if well.get('deadline'):
            prob += completion_time[w] <= well['deadline']

    # Earliest start times
    for w, well in enumerate(wells):
        prob += start_time[w] >= well.get('earliest_start', 0)

    # Makespan definition
    for w in range(len(wells)):
        prob += makespan >= completion_time[w]

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

    # Extract solution
    schedule = []
    for w, well in enumerate(wells):
        assigned_rig = [r for r in range(len(rigs)) if x[r, w].varValue > 0.5]
        if assigned_rig:
            r = assigned_rig[0]
            schedule.append({
                'well': well['id'],
                'rig': rigs[r]['id'],
                'start_day': start_time[w].varValue,
                'completion_day': completion_time[w].varValue,
                'drill_days': drilling_times.get((rigs[r]['id'], well['id']), 0)
            })

    schedule_df = pd.DataFrame(schedule).sort_values('start_day')

    return {
        'status': LpStatus[prob.status],
        'makespan': makespan.varValue,
        'total_cost': value(prob.objective),
        'schedule': schedule_df
    }

# Example usage
wells = [
    {'id': 'Well_A', 'location': (30.0, -95.0), 'priority': 1,
     'earliest_start': 0, 'deadline': 100},
    {'id': 'Well_B', 'location': (30.1, -95.1), 'priority': 2,
     'earliest_start': 0, 'deadline': 120},
    {'id': 'Well_C', 'location': (30.2, -95.0), 'priority': 1,
     'earliest_start': 0, 'deadline': 90},
]

rigs = [
    {'id': 'Rig_1', 'type': 'Land', 'availability': 0, 'day_rate': 25000,
     'current_location': (30.0, -95.0)},
    {'id': 'Rig_2', 'type': 'Land', 'availability': 0, 'day_rate': 22000,
     'current_location': (30.0, -95.0)},
]

drilling_times = {
    ('Rig_1', 'Well_A'): 25,
    ('Rig_1', 'Well_B'): 30,
    ('Rig_1', 'Well_C'): 22,
    ('Rig_2', 'Well_A'): 28,
    ('Rig_2', 'Well_B'): 32,
    ('Rig_2', 'Well_C'): 25,
}

result = optimize_rig_schedule(wells, rigs, drilling_times, {})
print(f"Project makespan: {result['makespan']:.0f} days")
print(result['schedule'])

---

钻井材料管理

管材库存优化

class TubularInventoryManager:
    """
    Manage drill pipe, casing, and tubing inventory for drilling program
    """

    def __init__(self, wells_program, tubular_specs, warehouse_locations):
        self.wells = wells_program
        self.tubulars = tubular_specs
        self.warehouses = warehouse_locations

    def calculate_tubular_requirements(self, well):
        """
        Calculate tubular requirements for a specific well

        Returns quantities needed by tubular type
        """
        requirements = {}

        # Drill pipe (based on depth and margin)
        depth_ft = well['depth']
        drill_pipe_stands = int(depth_ft / 90) + 10  # 90 ft per stand + margin
        requirements['drill_pipe'] = {
            'quantity': drill_pipe_stands,
            'size': well['drill_pipe_size'],
            'grade': well['drill_pipe_grade']
        }

        # Casing strings (multiple strings for deep wells)
        for casing in well['casing_program']:
            casing_joints = int(casing['depth'] / 40) + 2  # 40 ft per joint + margin
            key = f"casing_{casing['size']}_{casing['grade']}"
            requirements[key] = {
                'quantity': casing_joints,
                'size': casing['size'],
                'grade': casing['grade'],
                'depth': casing['depth']
            }

        # Production tubing
        if well.get('tubing_size'):
            tubing_joints = int(depth_ft / 40) + 2
            requirements['tubing'] = {
                'quantity': tubing_joints,
                'size': well['tubing_size'],
                'grade': well['tubing_grade']
            }

        return requirements

    def optimize_tubular_allocation(self):
        """
        Optimize allocation of tubular inventory across wells and locations
        """
        from pulp import *

        prob = LpProblem("Tubular_Allocation", LpMinimize)

        # Variables: allocate tubular inventory from warehouse to well
        allocation = {}

        for well in self.wells:
            well_reqs = self.calculate_tubular_requirements(well)

            for tubular_type, req in well_reqs.items():
                for wh in self.warehouses:
                    var_name = f"{well['id']}_{tubular_type}_{wh['id']}"
                    allocation[well['id'], tubular_type, wh['id']] = LpVariable(
                        var_name,
                        lowBound=0,
                        upBound=req['quantity']
                    )

        # Objective: minimize transportation cost
        transport_cost = []
        for (well_id, tub_type, wh_id), var in allocation.items():
            well = next(w for w in self.wells if w['id'] == well_id)
            wh = next(w for w in self.warehouses if w['id'] == wh_id)

            distance = self.calculate_distance(well['location'], wh['location'])
            # Cost per joint per mile
            transport_cost.append(var * distance * 5)

        prob += lpSum(transport_cost)

        # Constraints

        # Satisfy each well's tubular requirements
        for well in self.wells:
            well_reqs = self.calculate_tubular_requirements(well)

            for tubular_type, req in well_reqs.items():
                prob += lpSum([allocation.get((well['id'], tubular_type, wh['id']), 0)
                              for wh in self.warehouses]) >= req['quantity']

        # Warehouse inventory constraints
        for wh in self.warehouses:
            for tubular_type in self.tubulars:
                allocated = lpSum([allocation.get((well['id'], tubular_type, wh['id']), 0)
                                  for well in self.wells])

                prob += allocated <= wh['inventory'].get(tubular_type, 0)

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

        # Extract results
        allocation_plan = []
        for (well_id, tub_type, wh_id), var in allocation.items():
            if var.varValue > 0.1:
                allocation_plan.append({
                    'well': well_id,
                    'tubular_type': tub_type,
                    'warehouse': wh_id,
                    'quantity': var.varValue
                })

        return {
            'status': LpStatus[prob.status],
            'total_cost': value(prob.objective),
            'allocation_plan': pd.DataFrame(allocation_plan)
        }

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

---

钻井泥浆管理

泥浆方案优化

def optimize_drilling_fluid_inventory(wells, mud_systems, supply_base):
    """
    Optimize drilling fluid (mud) inventory and logistics

    Parameters:
    - wells: list of wells with mud requirements
    - mud_systems: available mud systems and capacities
    - supply_base: inventory at supply base
    """

    requirements = []

    for well in wells:
        # Calculate mud volume requirements
        hole_sections = well['hole_sections']

        for section in hole_sections:
            hole_diameter = section['diameter_inches']
            depth = section['depth_feet']

            # Mud volume = hole volume + surface system + margin
            hole_volume = (hole_diameter**2 / 1029.4) * depth  # bbls
            surface_volume = 200  # bbls (typical)
            safety_margin = 1.3

            total_volume = (hole_volume + surface_volume) * safety_margin

            requirements.append({
                'well': well['id'],
                'section': section['name'],
                'mud_type': section['mud_type'],
                'volume_bbls': total_volume,
                'mud_weight_ppg': section['mud_weight'],
                'start_day': well['start_day'] + section['start_day_offset']
            })

    requirements_df = pd.DataFrame(requirements)

    # Aggregate by mud type and time period
    mud_schedule = requirements_df.groupby(['mud_type', 'start_day']).agg({
        'volume_bbls': 'sum'
    }).reset_index()

    # Calculate procurement schedule
    procurement = []
    inventory = {}

    for _, row in mud_schedule.iterrows():
        mud_type = row['mud_type']
        day = row['start_day']
        required = row['volume_bbls']

        # Current inventory
        current_inv = inventory.get((mud_type, day), 0)

        if current_inv < required:
            # Need to procure
            order_quantity = required - current_inv + 100  # Add buffer
            lead_time = 7  # days

            procurement.append({
                'mud_type': mud_type,
                'quantity_bbls': order_quantity,
                'order_day': day - lead_time,
                'delivery_day': day,
                'cost': order_quantity * get_mud_cost(mud_type)
            })

            # Update inventory
            inventory[(mud_type, day)] = order_quantity

        # Consume inventory
        inventory[(mud_type, day)] -= required

    return {
        'requirements': requirements_df,
        'procurement_schedule': pd.DataFrame(procurement),
        'total_mud_cost': sum([p['cost'] for p in procurement])
    }

def get_mud_cost(mud_type):
    """Get cost per barrel for mud type"""
    costs = {
        'water_based': 50,
        'oil_based': 150,
        'synthetic_based': 200
    }
    return costs.get(mud_type, 100)

---

海上物流

供应船调度

def optimize_supply_vessel_schedule(rigs, supply_base, vessels, cargo_demand):
    """
    Optimize offshore supply vessel routing and scheduling

    Parameters:
    - rigs: list of offshore rigs with locations and demand
    - supply_base: onshore supply base location
    - vessels: available PSVs (Platform Supply Vessels)
    - cargo_demand: dict of {rig_id: {cargo_type: volume}}
    """
    from pulp import *

    prob = LpProblem("Vessel_Routing", LpMinimize)

    days = 30  # Planning horizon
    T = range(days)

    # Variables

    # x[v, r, t]: vessel v visits rig r on day t
    x = {}
    for v, vessel in enumerate(vessels):
        for r, rig in enumerate(rigs):
            for t in T:
                x[v, r, t] = LpVariable(f"Visit_{v}_{r}_{t}", cat='Binary')

    # Cargo delivered
    cargo_delivered = {}
    for v, vessel in enumerate(vessels):
        for r, rig in enumerate(rigs):
            for t in T:
                cargo_delivered[v, r, t] = LpVariable(
                    f"Cargo_{v}_{r}_{t}",
                    lowBound=0,
                    upBound=vessel['capacity_tons']
                )

    # Objective: minimize total vessel operating cost
    vessel_cost = lpSum([vessels[v]['day_rate'] * x[v, r, t]
                        for v in range(len(vessels))
                        for r in range(len(rigs))
                        for t in T])

    prob += vessel_cost

    # Constraints

    # Meet rig demand over planning horizon
    for r, rig in enumerate(rigs):
        total_demand = sum(cargo_demand.get(rig['id'], {}).values())

        prob += lpSum([cargo_delivered[v, r, t]
                      for v in range(len(vessels))
                      for t in T]) >= total_demand

    # Vessel capacity
    for v, vessel in enumerate(vessels):
        for r, rig in enumerate(rigs):
            for t in T:
                prob += cargo_delivered[v, r, t] <= \
                        vessel['capacity_tons'] * x[v, r, t]

    # Vessel can visit one rig per day
    for v, vessel in enumerate(vessels):
        for t in T:
            prob += lpSum([x[v, r, t] for r in range(len(rigs))]) <= 1

    # Minimum visit frequency (safety requirement)
    for r, rig in enumerate(rigs):
        # At least one visit every 7 days
        for t_start in range(0, days - 7, 7):
            prob += lpSum([x[v, r, t]
                          for v in range(len(vessels))
                          for t in range(t_start, min(t_start + 7, days))]) >= 1

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

    # Extract schedule
    schedule = []
    for v, vessel in enumerate(vessels):
        for r, rig in enumerate(rigs):
            for t in T:
                if x[v, r, t].varValue > 0.5:
                    schedule.append({
                        'vessel': vessel['id'],
                        'rig': rig['id'],
                        'day': t,
                        'cargo_tons': cargo_delivered[v, r, t].varValue
                    })

    return {
        'status': LpStatus[prob.status],
        'total_cost': value(prob.objective),
        'schedule': pd.DataFrame(schedule).sort_values('day'),
        'vessel_utilization': {
            vessel['id']: sum([x[v, r, t].varValue
                              for r in range(len(rigs))
                              for t in T]) / days * 100
            for v, vessel in enumerate(vessels)
        }
    }

---

减少非生产时间（NPT）

NPT 分析与优化

class NPTAnalyzer:
    """
    Analyze and optimize non-productive time in drilling operations
    """

    def __init__(self, historical_wells):
        self.wells = historical_wells

    def categorize_npt(self, well_data):
        """
        Categorize NPT by root cause

        Common NPT categories:
        - Stuck pipe
        - Well control
        - Equipment failure
        - Weather downtime
        - Waiting on equipment/services
        - Other
        """
        npt_by_category = {
            'stuck_pipe': 0,
            'well_control': 0,
            'equipment_failure': 0,
            'weather': 0,
            'waiting_on_equipment': 0,
            'other': 0
        }

        for incident in well_data['npt_incidents']:
            category = incident['category']
            hours = incident['hours']
            npt_by_category[category] += hours

        return npt_by_category

    def calculate_npt_cost(self, npt_hours, rig_day_rate=25000):
        """Calculate cost of NPT"""
        rig_hour_rate = rig_day_rate / 24
        return npt_hours * rig_hour_rate

    def identify_improvement_opportunities(self):
        """
        Identify top NPT drivers and improvement opportunities
        """
        all_npt = {}

        for well in self.wells:
            npt_categories = self.categorize_npt(well)
            for category, hours in npt_categories.items():
                if category not in all_npt:
                    all_npt[category] = []
                all_npt[category].append(hours)

        # Calculate statistics
        npt_summary = []
        for category, hours_list in all_npt.items():
            npt_summary.append({
                'category': category,
                'total_hours': sum(hours_list),
                'avg_hours_per_well': np.mean(hours_list),
                'frequency': len([h for h in hours_list if h > 0]),
                'cost': self.calculate_npt_cost(sum(hours_list))
            })

        npt_df = pd.DataFrame(npt_summary).sort_values('total_hours',
                                                       ascending=False)

        return npt_df

    def recommend_mitigation_actions(self, npt_summary):
        """
        Recommend actions to reduce NPT
        """
        recommendations = []

        for _, row in npt_summary.iterrows():
            category = row['category']

            if category == 'stuck_pipe' and row['total_hours'] > 100:
                recommendations.append({
                    'category': category,
                    'action': 'Improve hole cleaning practices, use real-time monitoring',
                    'estimated_reduction': '30-50%',
                    'investment_required': 'Low-Medium'
                })

            elif category == 'equipment_failure' and row['total_hours'] > 80:
                recommendations.append({
                    'category': category,
                    'action': 'Implement predictive maintenance, upgrade critical equipment',
                    'estimated_reduction': '40-60%',
                    'investment_required': 'Medium-High'
                })

            elif category == 'waiting_on_equipment' and row['total_hours'] > 60:
                recommendations.append({
                    'category': category,
                    'action': 'Improve logistics planning, increase critical spare inventory',
                    'estimated_reduction': '50-70%',
                    'investment_required': 'Low'
                })

            elif category == 'weather' and row['total_hours'] > 50:
                recommendations.append({
                    'category': category,
                    'action': 'Improve weather forecasting, adjust operational windows',
                    'estimated_reduction': '20-30%',
                    'investment_required': 'Low'
                })

        return pd.DataFrame(recommendations)

---

工具与库

Python 库

优化：

• PuLP：线性规划
• Pyomo：优化建模
• OR-Tools：Google 优化工具
• scipy.optimize：通用优化

地理空间：

• geopandas：地理数据
• geopy：距离计算
• folium：交互式地图

数据分析：

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

商业软件

钻井规划：

• Landmark (Halliburton) WellPlan：井位规划与设计
• Schlumberger Techlog：井筒数据管理
• Baker Hughes JewelSuite：井建规划
• DrillScan：钻井作业优化

物流与供应链：

• SAP S/4HANA Oil & Gas：石油和天然气行业 ERP
• Oracle E-Business Suite：供应链管理
• IFS Applications：项目驱动型 ERP
• Quorum Business Solutions：能源软件

数据与分析：

• Pason：实时钻井数据
• Corva：钻井分析平台
• Well Data Labs：用于钻井的机器学习

---

常见挑战与解决方案

挑战：钻机调迁延误

问题：

• 钻机搬迁的提前期较长
• 调迁成本高昂
• 许可和准备工作延误

解决方案：

• 批量钻井（从同一井场钻多口井）
• 提前规划并办理许可
• 保持井场随时可供钻机进驻
• 对间距较近的井使用步进式钻机
• 优化钻机计划，尽量减少搬迁次数

挑战：设备可用性

问题：

• 专用工具交货周期长
• 设备故障和停机
• 关键物品库存有限

解决方案：

• 战略性库存配置
• 预测性维护计划
• 设备共享协议
• 备用设备待命
• 供应商管理库存

挑战：海上天气延误

问题：

• 关键作业受天气窗口限制
• 人员转运受限
• 供应船延误

解决方案：

• 先进的天气预报
• 采用带缓冲时间的灵活排程
• 尽可能在受天气影响较小的条件下作业
• 在平台上储备充足物资
• 采用替代运输方式（直升机）

挑战：物料搬运与追踪

问题：

• 设备丢失或放错位置
• 库存差异
• 浪费时间寻找物料

解决方案：

• RFID 标签和追踪系统
• 集中式库存管理
• 库存数字孪生
• 自动化签入／签出
• 实时可视化仪表板

---

输出格式

钻井物流计划

执行摘要：

• 钻井计划概览
• 预计总工期和总成本
• 关键物流策略
• 风险缓解方案

钻机计划：

井号	位置	钻机	开始日期	工期（天）	完工日期	状态
Well-A	Pad 1	Rig-1	2026-03-01	25	2026-03-26	计划中
Well-B	Pad 1	Rig-1	2026-03-27	30	2026-04-26	计划中
Well-C	Pad 2	Rig-2	2026-03-15	28	2026-04-12	计划中

物料需求：

物品	总数量	单位	成本	交货周期	采购状态
5" 钻杆	1,500	根	$2.5M	60 天	已下单
9-5/8" 套管	800	根	$1.2M	45 天	有库存
钻井泥浆（OBM）	5,000	桶	$750K	14 天	已下单

物流成本明细：

类别	成本	占总额比例
钻机日费	$15.0M	65%
管材与物料	$4.5M	20%
服务（定向钻井、固井等）	$2.0M	9%
运输与物流	$1.0M	4%
应急费用	$0.5M	2%
总计	$23.0M	100%

关键路径项目：

项目	所需日期	状态	风险等级	缓解措施
Rig-1 可用性	2026-03-01	已确认	低	已签订合同
BHA 组件	2026-02-20	待处理	中	加急订单
完井设备	2026-04-01	已下单	低	交货周期充足

KPI 与目标：

指标	目标	基线	备注
单井平均天数	25	30	提升 17%
NPT %	< 5%	8%	重点提升设备可靠性
单英尺成本	$180	$220	降低 18%
安全事故	0	-	TRIR < 0.5

---

要询问的问题

如需更多背景信息：

1. 属于哪种钻井计划？（陆上、海上、井数）
2. 时间安排和紧迫程度如何？
3. 有哪些钻机可用或已签约？
4. 供应链基础设施如何？（基地、仓库、运输）
5. 主要成本驱动因素和限制是什么？
6. 当前 NPT 率是多少？主要原因是什么？
7. 必须满足哪些安全和环境要求？

---

相关技能

• energy-logistics：用于中游和下游油气业务
• fuel-distribution：用于成品油分销
• network-design：用于供应链网络优化
• route-optimization：用于运输路线规划
• inventory-optimization：用于物料库存管理
• project-scheduling：用于施工排程
• risk-mitigation：用于运营风险管理
• fleet-management：用于车辆和设备车队