Commodities Trading API

import requests
from datetime import datetime, timedelta
from typing import Dict, List
import statistics

class CommodityTradingAPI:
    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_current_prices(self, commodities: List[str]) -> Dict:
        """Get current prices for specified commodities."""
        response = requests.get(
            f'{self.base_url}/prices/latest',
            params={'by_code': ','.join(commodities)},
            headers=self.headers
        )
        return response.json()

    def get_spread(self, commodity_a: str, commodity_b: str) -> Dict:
        """Calculate spread between two commodities."""
        data = self.get_current_prices([commodity_a, commodity_b])

        # Parse response for both commodities
        prices = self._parse_prices(data)

        if commodity_a in prices and commodity_b in prices:
            spread = prices[commodity_a] - prices[commodity_b]
            return {
                'commodity_a': commodity_a,
                'commodity_b': commodity_b,
                'price_a': prices[commodity_a],
                'price_b': prices[commodity_b],
                'spread': round(spread, 2),
                'spread_pct': round((spread / prices[commodity_b]) * 100, 2)
            }
        return {'error': 'Could not calculate spread'}

    def _parse_prices(self, data) -> Dict[str, float]:
        """Parse API response to extract prices by code."""
        prices = {}
        if data.get('status') == 'success':
            d = data.get('data', {})
            if isinstance(d, dict) and 'code' in d:
                prices[d['code']] = d['price']
            elif isinstance(d, list):
                for item in d:
                    prices[item['code']] = item['price']
        return prices

    def analyze_trend(self, commodity: str, days: int = 30) -> Dict:
        """Analyze price trend for a commodity."""
        endpoint = 'past_week' if days <= 7 else 'past_month' if days <= 30 else 'past_year'
        response = requests.get(
            f'{self.base_url}/prices/{endpoint}',
            params={'by_code': commodity, 'interval': '1d'},
            headers=self.headers
        )
        data = response.json()
        prices = [p['price'] for p in data['data']['prices']]

        current = prices[0]
        avg = statistics.mean(prices)
        std_dev = statistics.stdev(prices) if len(prices) > 1 else 0

        # Calculate momentum
        if len(prices) >= 5:
            short_ma = statistics.mean(prices[:5])
            long_ma = statistics.mean(prices)
            momentum = 'bullish' if short_ma > long_ma else 'bearish'
        else:
            momentum = 'insufficient_data'

        return {
            'commodity': commodity,
            'current_price': current,
            'average': round(avg, 2),
            'std_deviation': round(std_dev, 3),
            'min': min(prices),
            'max': max(prices),
            'volatility_pct': round((std_dev / avg) * 100, 2),
            'momentum': momentum,
            'z_score': round((current - avg) / std_dev, 2) if std_dev > 0 else 0
        }

    def get_correlation_data(self, commodities: List[str], days: int = 30) -> Dict:
        """Get historical data for correlation analysis."""
        endpoint = 'past_month' if days <= 30 else 'past_year'
        results = {}

        for commodity in commodities:
            response = requests.get(
                f'{self.base_url}/prices/{endpoint}',
                params={'by_code': commodity, 'interval': '1d'},
                headers=self.headers
            )
            data = response.json()
            results[commodity] = [
                {'date': p['created_at'], 'price': p['price']}
                for p in data['data']['prices']
            ]

        return results

    def calculate_brent_wti_spread(self) -> Dict:
        """Calculate the Brent-WTI spread (common trading indicator)."""
        return self.get_spread('BRENT_CRUDE_USD', 'WTI_USD')

    def calculate_crack_spread(self) -> Dict:
        """Calculate 3-2-1 crack spread approximation."""
        commodities = ['WTI_USD', 'GASOLINE_USD', 'HEATING_OIL_USD']
        prices = self._parse_prices(self.get_current_prices(commodities))

        if all(c in prices for c in commodities):
            # Simplified crack spread: 2*gasoline + 1*heating oil - 3*crude
            # Note: Actual calculation requires barrel conversions
            crude = prices['WTI_USD']
            gasoline = prices['GASOLINE_USD'] * 42  # gallons to barrel approximation
            heating_oil = prices['HEATING_OIL_USD'] * 42

            crack_spread = (2 * gasoline + heating_oil) / 3 - crude

            return {
                'crude_price': crude,
                'gasoline_price': prices['GASOLINE_USD'],
                'heating_oil_price': prices['HEATING_OIL_USD'],
                'crack_spread': round(crack_spread, 2),
                'note': 'Simplified calculation - actual spread requires precise barrel conversions'
            }
        return {'error': 'Could not calculate crack spread'}

# Usage example
trading = CommodityTradingAPI('YOUR_API_KEY')

# Get Brent-WTI spread
spread = trading.calculate_brent_wti_spread()
print(f"Brent-WTI spread: ${spread['spread']}")

# Analyze WTI trend
trend = trading.analyze_trend('WTI_USD', days=30)
print(f"WTI momentum: {trend['momentum']}, Z-score: {trend['z_score']}")