Maritime Bunker Fuel API

import requests
from dataclasses import dataclass
from typing import List

@dataclass
class Port:
    code: str
    name: str
    eta_days: int

class BunkerOptimizer:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = 'https://api.oilpriceapi.com/v1'
        self.headers = {'Authorization': f'Token {api_key}'}

    def get_port_prices(self, port_codes: List[str], fuel_grade='VLSFO'):
        """Get current bunker prices for specified ports."""
        codes = [f'{fuel_grade}_{code}_USD' for code in port_codes]
        response = requests.get(
            f'{self.base_url}/prices/latest',
            params={'by_code': ','.join(codes)},
            headers=self.headers
        )
        return response.json()

    def compare_bunkering_options(self, voyage_ports: List[Port],
                                   fuel_needed_mt: float,
                                   fuel_grade='VLSFO'):
        """Compare bunkering costs across voyage ports."""
        port_codes = [p.code for p in voyage_ports]
        prices_response = self.get_port_prices(port_codes, fuel_grade)

        options = []
        for port in voyage_ports:
            code = f'{fuel_grade}_{port.code}_USD'
            # Find price for this port
            price_per_mt = self._find_price(prices_response, code)
            if price_per_mt:
                total_cost = price_per_mt * fuel_needed_mt
                options.append({
                    'port': port.name,
                    'port_code': port.code,
                    'eta_days': port.eta_days,
                    'price_per_mt': price_per_mt,
                    'total_cost': round(total_cost, 2),
                    'fuel_grade': fuel_grade
                })

        # Sort by total cost
        options.sort(key=lambda x: x['total_cost'])
        return options

    def _find_price(self, response, commodity_code):
        """Extract price from API response."""
        if response.get('status') == 'success':
            data = response.get('data', {})
            if isinstance(data, dict) and data.get('code') == commodity_code:
                return data.get('price')
        return None

    def calculate_voyage_fuel_cost(self, voyage_legs, vessel_consumption_mt_per_day):
        """Calculate total fuel cost for a voyage with optimized bunkering."""
        total_fuel_needed = sum(
            leg['distance_nm'] / leg['speed_knots'] / 24 * vessel_consumption_mt_per_day
            for leg in voyage_legs
        )

        # Get all port prices
        port_codes = list(set(leg['bunker_port'] for leg in voyage_legs if leg.get('bunker_port')))
        prices = self.get_port_prices(port_codes)

        return {
            'total_fuel_mt': round(total_fuel_needed, 2),
            'bunker_options': self.compare_bunkering_options(
                [Port(code=c, name=c, eta_days=0) for c in port_codes],
                total_fuel_needed
            )
        }

# Usage example
optimizer = BunkerOptimizer('YOUR_API_KEY')

voyage_ports = [
    Port(code='SGSIN', name='Singapore', eta_days=0),
    Port(code='AEFUJ', name='Fujairah', eta_days=5),
    Port(code='NLRTM', name='Rotterdam', eta_days=18)
]

options = optimizer.compare_bunkering_options(voyage_ports, fuel_needed_mt=2500)
print("Bunkering options (sorted by cost):")
for opt in options:
    print(f"  {opt['port']}: ${opt['total_cost']:,.0f} (${opt['price_per_mt']}/MT)")