Build a US LNG Export Price Tracker with the Energy Volatility API in Python

# data/lng_price_fetcher.py
import httpx
import asyncio
from datetime import datetime, timezone
from typing import Optional
 
ENERGY_VOLATILITY_API_URL = "https://apivult.com/api/energy-volatility/v1"
API_KEY = "YOUR_API_KEY"
 
async def fetch_henry_hub_data(days_back: int = 30) -> dict:
    """Fetch Henry Hub natural gas spot and futures prices."""
    async with httpx.AsyncClient(timeout=20.0) as client:
        response = await client.get(
            f"{ENERGY_VOLATILITY_API_URL}/gas/henry-hub",
            headers={"X-RapidAPI-Key": API_KEY},
            params={
                "include_futures": True,
                "include_volatility": True,
                "days_back": days_back
            }
        )
        response.raise_for_status()
        return response.json()
 
async def fetch_lng_export_data() -> dict:
    """Fetch US LNG export utilization and flow data."""
    async with httpx.AsyncClient(timeout=20.0) as client:
        response = await client.get(
            f"{ENERGY_VOLATILITY_API_URL}/lng/us-exports",
            headers={"X-RapidAPI-Key": API_KEY},
            params={"include_terminal_breakdown": True}
        )
        response.raise_for_status()
        return response.json()
 
async def fetch_european_gas_prices() -> dict:
    """Fetch European TTF gas hub prices for spread calculation."""
    async with httpx.AsyncClient(timeout=20.0) as client:
        response = await client.get(
            f"{ENERGY_VOLATILITY_API_URL}/gas/european-hubs",
            headers={"X-RapidAPI-Key": API_KEY},
            params={"hubs": "TTF,NBP,THE", "days_back": 30}
        )
        response.raise_for_status()
        return response.json()
 
async def fetch_all_lng_data() -> tuple[dict, dict, dict]:
    """Fetch all required data in parallel."""
    hh_task = fetch_henry_hub_data(days_back=60)
    export_task = fetch_lng_export_data()
    eu_task = fetch_european_gas_prices()
    
    henry_hub, exports, eu_prices = await asyncio.gather(hh_task, export_task, eu_task)
    return henry_hub, exports, eu_prices

# analysis/lng_arbitrage_calculator.py
from dataclasses import dataclass
from typing import Optional
 
# LNG export cost components ($/MMBtu, approximate 2026 averages)
LIQUEFACTION_COST = 2.50      # Tolling fee at Sabine Pass, Freeport, etc.
SHIPPING_COST_ATLANTIC = 1.20  # Jones Act-exempt shipping to Europe
REGASIFICATION_COST = 0.30    # European regasification terminal fee
TOTAL_TRANSPORT_COST = LIQUEFACTION_COST + SHIPPING_COST_ATLANTIC + REGASIFICATION_COST
 
@dataclass
class LNGArbitrageSnapshot:
    timestamp: str
    henry_hub_spot: float          # $/MMBtu
    ttf_spot: float                # $/MMBtu (converted from €/MWh)
    nbp_spot: float                # $/MMBtu
    all_in_export_cost: float      # $/MMBtu (HH + transport costs)
    henry_hub_ttf_spread: float    # TTF - all_in_cost
    henry_hub_nbp_spread: float    # NBP - all_in_cost
    export_economics: str          # STRONG / MARGINAL / NEGATIVE
    utilization_rate_pct: float    # US LNG terminal utilization
    daily_exports_bcf: float       # Billion cubic feet per day
 
def calculate_arbitrage_spreads(
    henry_hub_data: dict,
    eu_data: dict,
    export_data: dict
) -> LNGArbitrageSnapshot:
    """Calculate current LNG export arbitrage economics."""
    
    hh_spot = henry_hub_data["spot_price_usd_mmbtu"]
    
    # TTF is quoted in €/MWh — convert to $/MMBtu
    # 1 MWh = 3.412 MMBtu; use daily EUR/USD rate
    ttf_eur_mwh = eu_data["hubs"]["TTF"]["spot_price"]
    eur_usd_rate = eu_data.get("eur_usd_rate", 1.08)
    ttf_usd_mmbtu = (ttf_eur_mwh * eur_usd_rate) / 3.412
    
    nbp_spot = eu_data["hubs"].get("NBP", {}).get("spot_price_usd_mmbtu", 0)
    
    all_in_cost = hh_spot + TOTAL_TRANSPORT_COST
    ttf_spread = ttf_usd_mmbtu - all_in_cost
    nbp_spread = nbp_spot - all_in_cost if nbp_spot else 0
 
    # Classify export economics
    if ttf_spread > 4.0:
        economics = "STRONG"
    elif ttf_spread > 1.5:
        economics = "MARGINAL"
    else:
        economics = "NEGATIVE"
 
    return LNGArbitrageSnapshot(
        timestamp=henry_hub_data.get("timestamp", ""),
        henry_hub_spot=hh_spot,
        ttf_spot=ttf_usd_mmbtu,
        nbp_spot=nbp_spot,
        all_in_export_cost=all_in_cost,
        henry_hub_ttf_spread=round(ttf_spread, 3),
        henry_hub_nbp_spread=round(nbp_spread, 3),
        export_economics=economics,
        utilization_rate_pct=export_data.get("utilization_rate_pct", 0),
        daily_exports_bcf=export_data.get("daily_exports_bcf", 0)
    )

# dashboard/lng_dashboard.py
import asyncio
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.columns import Columns
from data.lng_price_fetcher import fetch_all_lng_data
from analysis.lng_arbitrage_calculator import calculate_arbitrage_spreads
 
console = Console()
 
def render_economics_indicator(economics: str) -> str:
    colors = {"STRONG": "green", "MARGINAL": "yellow", "NEGATIVE": "red"}
    icons = {"STRONG": "▲", "MARGINAL": "►", "NEGATIVE": "▼"}
    color = colors.get(economics, "white")
    icon = icons.get(economics, "?")
    return f"[{color}]{icon} {economics}[/{color}]"
 
async def render_lng_dashboard():
    """Render real-time LNG export economics dashboard."""
    henry_hub, exports, eu_prices = await fetch_all_lng_data()
    snapshot = calculate_arbitrage_spreads(henry_hub, eu_prices, exports)
 
    # Price table
    price_table = Table(title="Gas Hub Prices ($/MMBtu)", border_style="blue")
    price_table.add_column("Hub", style="cyan")
    price_table.add_column("Spot", justify="right")
    price_table.add_column("Change", justify="right")
    
    price_table.add_row("Henry Hub (US)", f"${snapshot.henry_hub_spot:.3f}", 
                        henry_hub.get("change_formatted", ""))
    price_table.add_row("TTF (Europe)", f"${snapshot.ttf_spot:.3f}", 
                        eu_prices["hubs"]["TTF"].get("change_formatted", ""))
    price_table.add_row("NBP (UK)", f"${snapshot.nbp_spot:.3f}", 
                        eu_prices["hubs"].get("NBP", {}).get("change_formatted", ""))
 
    # Arbitrage table
    arb_table = Table(title="LNG Export Economics", border_style="green")
    arb_table.add_column("Component", style="cyan")
    arb_table.add_column("Value", justify="right")
    
    arb_table.add_row("Henry Hub Spot", f"${snapshot.henry_hub_spot:.3f}/MMBtu")
    arb_table.add_row("+ Liquefaction", f"${TOTAL_TRANSPORT_COST:.2f}/MMBtu")
    arb_table.add_row("= All-In Export Cost", f"${snapshot.all_in_export_cost:.3f}/MMBtu")
    arb_table.add_row("──────────────────", "──────────────")
    arb_table.add_row("HH-TTF Spread", f"${snapshot.henry_hub_ttf_spread:.3f}/MMBtu")
    arb_table.add_row("HH-NBP Spread", f"${snapshot.henry_hub_nbp_spread:.3f}/MMBtu")
    arb_table.add_row("Export Economics", render_economics_indicator(snapshot.export_economics))
 
    # Export volume table
    volume_table = Table(title="US LNG Export Activity", border_style="yellow")
    volume_table.add_column("Metric", style="cyan")
    volume_table.add_column("Value", justify="right")
    
    volume_table.add_row("Daily Exports", f"{snapshot.daily_exports_bcf:.1f} Bcf/day")
    volume_table.add_row("Utilization Rate", f"{snapshot.utilization_rate_pct:.1f}%")
 
    # Terminal breakdown
    terminal_table = Table(title="Terminal Utilization", border_style="magenta")
    terminal_table.add_column("Terminal", style="cyan")
    terminal_table.add_column("Capacity (Bcf/d)", justify="right")
    terminal_table.add_column("Utilization", justify="right")
    
    for terminal in exports.get("terminals", []):
        terminal_table.add_row(
            terminal["name"],
            f"{terminal['capacity_bcf_day']:.1f}",
            f"{terminal['utilization_pct']:.0f}%"
        )
 
    console.print(Panel(f"[bold]US LNG Export Price Tracker[/bold] — {snapshot.timestamp}", 
                        style="bold blue"))
    console.print(Columns([price_table, arb_table]))
    console.print(Columns([volume_table, terminal_table]))
 
from analysis.lng_arbitrage_calculator import TOTAL_TRANSPORT_COST
asyncio.run(render_lng_dashboard())

# alerts/lng_spread_alerts.py
import asyncio
import smtplib
from email.mime.text import MIMEText
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class SpreadAlertConfig:
    strong_threshold: float = 4.0   # Alert when spread opens above this (export rush)
    marginal_threshold: float = 1.5  # Alert when spread drops below this (exports slow)
    extreme_threshold: float = 8.0   # Alert on extreme spread (market dislocation)
 
async def check_and_alert(
    current_spread: float,
    previous_spread: Optional[float],
    config: SpreadAlertConfig
) -> Optional[str]:
    """
    Check if the spread has crossed an alert threshold.
    Returns alert message if threshold crossed, None otherwise.
    """
    if previous_spread is None:
        return None
 
    # Check for threshold crossings (not just current level)
    alerts = []
 
    if current_spread > config.extreme_threshold:
        alerts.append(
            f"EXTREME SPREAD ALERT: HH-TTF spread at ${current_spread:.2f}/MMBtu "
            f"— market dislocation may indicate supply disruption or price spike"
        )
    elif current_spread > config.strong_threshold and previous_spread <= config.strong_threshold:
        alerts.append(
            f"SPREAD OPENED: HH-TTF spread crossed ${config.strong_threshold}/MMBtu "
            f"(now ${current_spread:.2f}/MMBtu) — export economics STRONG, expect utilization increase"
        )
    elif current_spread < config.marginal_threshold and previous_spread >= config.marginal_threshold:
        alerts.append(
            f"SPREAD COMPRESSED: HH-TTF spread dropped below ${config.marginal_threshold}/MMBtu "
            f"(now ${current_spread:.2f}/MMBtu) — export economics deteriorating"
        )
 
    return "\n".join(alerts) if alerts else None
 
class LNGSpreadMonitor:
    """Continuous LNG spread monitoring with configurable alerts."""
    
    def __init__(self, check_interval_minutes: int = 15):
        self.check_interval = check_interval_minutes * 60
        self.previous_spread: Optional[float] = None
        self.alert_config = SpreadAlertConfig()
 
    async def run(self):
        from data.lng_price_fetcher import fetch_all_lng_data
        from analysis.lng_arbitrage_calculator import calculate_arbitrage_spreads
 
        print(f"Starting LNG spread monitor (checking every {self.check_interval // 60} minutes)...")
        
        while True:
            try:
                henry_hub, exports, eu_prices = await fetch_all_lng_data()
                snapshot = calculate_arbitrage_spreads(henry_hub, eu_prices, exports)
                
                alert = await check_and_alert(
                    snapshot.henry_hub_ttf_spread,
                    self.previous_spread,
                    self.alert_config
                )
                
                if alert:
                    print(f"\n⚠ ALERT: {alert}")
                    # Send to your alerting system (Slack, email, PagerDuty)
 
                print(f"[{snapshot.timestamp}] HH: ${snapshot.henry_hub_spot:.3f} | "
                      f"TTF: ${snapshot.ttf_spot:.3f} | "
                      f"Spread: ${snapshot.henry_hub_ttf_spread:.3f} | "
                      f"{snapshot.export_economics}")
 
                self.previous_spread = snapshot.henry_hub_ttf_spread
 
            except Exception as e:
                print(f"Error fetching data: {e}")
 
            await asyncio.sleep(self.check_interval)
 
# Run the monitor
monitor = LNGSpreadMonitor(check_interval_minutes=15)
asyncio.run(monitor.run())