How to Add Real-Time Oil Prices to Your Oilfield Software [Developer Guide]

import requests
from functools import lru_cache
import time

class OilPriceClient:
    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}"}

    @lru_cache(maxsize=100)
    def get_price(self, code, cache_key=None):
        """Get price with 5-minute cache."""
        response = requests.get(
            f"{self.base_url}/prices/latest",
            params={"by_code": code},
            headers=self.headers
        )
        return response.json()["data"]

    def get_cached_price(self, code):
        """Return cached price, refreshing every 5 minutes."""
        cache_key = int(time.time() // 300)  # New key every 5 min
        return self.get_price(code, cache_key)

# Usage in your application
client = OilPriceClient("YOUR_API_KEY")

def render_price_widget():
    brent = client.get_cached_price("BRENT_CRUDE_USD")
    wti = client.get_cached_price("WTI_CRUDE_USD")

    return {
        "brent": {"price": brent["price"], "change": brent["change_percent"]},
        "wti": {"price": wti["price"], "change": wti["change_percent"]}
    }

class WellEconomics:
    def __init__(self, api_key):
        self.client = OilPriceClient(api_key)

    def calculate_revenue(self, daily_oil_bbls, daily_gas_mcf):
        """Calculate daily revenue at current prices."""
        oil_price = self.client.get_cached_price("WTI_CRUDE_USD")["price"]
        gas_price = self.client.get_cached_price("NATURAL_GAS_USD")["price"]

        oil_revenue = daily_oil_bbls * oil_price
        gas_revenue = daily_gas_mcf * (gas_price / 10)  # $/MMBtu to $/Mcf

        return {
            "oil_price": oil_price,
            "gas_price": gas_price,
            "daily_oil_revenue": round(oil_revenue, 2),
            "daily_gas_revenue": round(gas_revenue, 2),
            "total_daily_revenue": round(oil_revenue + gas_revenue, 2),
            "monthly_estimate": round((oil_revenue + gas_revenue) * 30, 2)
        }

    def calculate_npv(self, production_profile, discount_rate=0.10):
        """Calculate NPV using current prices."""
        current_price = self.client.get_cached_price("WTI_CRUDE_USD")["price"]
        npv = 0

        for month, production_bbls in enumerate(production_profile):
            revenue = production_bbls * current_price
            discount_factor = 1 / (1 + discount_rate / 12) ** month
            npv += revenue * discount_factor

        return round(npv, 2)

# Usage
economics = WellEconomics("YOUR_API_KEY")
revenue = economics.calculate_revenue(daily_oil_bbls=150, daily_gas_mcf=500)
print(f"Daily Revenue: ${revenue['total_daily_revenue']:,.2f}")

def get_dashboard_data(api_key, daily_production_bbls):
    """Get production value with 30-day price context."""
    client = OilPriceClient(api_key)

    # Current price
    current = client.get_cached_price("WTI_CRUDE_USD")

    # Historical prices
    response = requests.get(
        f"{client.base_url}/prices/past_month",
        params={"by_code": "WTI_CRUDE_USD", "interval": "1d"},
        headers=client.headers
    )
    history = response.json()["data"]["prices"]

    # Calculate metrics
    prices = [p["price"] for p in history]
    avg_30d = sum(prices) / len(prices)
    high_30d = max(prices)
    low_30d = min(prices)

    return {
        "current_price": current["price"],
        "current_change": current["change_percent"],
        "price_30d_avg": round(avg_30d, 2),
        "price_30d_high": high_30d,
        "price_30d_low": low_30d,
        "daily_production": daily_production_bbls,
        "daily_revenue_current": round(daily_production_bbls * current["price"], 2),
        "daily_revenue_avg": round(daily_production_bbls * avg_30d, 2),
        "variance_to_avg": round((current["price"] - avg_30d) / avg_30d * 100, 1)
    }

def get_price_with_fallback(client, code, fallback_price=None):
    """Get price with graceful degradation."""
    try:
        response = client.get_price(code)
        return response["price"]
    except requests.exceptions.RequestException as e:
        # Log the error
        logging.warning(f"API request failed: {e}")

        # Return cached value if available
        if fallback_price:
            return fallback_price

        # Return None and let UI show "Price unavailable"
        return None

import time
from functools import wraps

def retry_with_backoff(max_retries=3):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.HTTPError as e:
                    if e.response.status_code == 429:
                        wait = 2 ** attempt
                        time.sleep(wait)
                    else:
                        raise
            raise Exception("Max retries exceeded")
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3)
def fetch_price(client, code):
    return client.get_price(code)