Energy Market Risk Management: Integrating Real-Time Volatility Data into Trading Systems

pip install requests pandas numpy scipy matplotlib python-dotenv

import os
import requests
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from dotenv import load_dotenv
 
load_dotenv()
 
ENERGY_API_KEY = os.getenv("ENERGY_VOLATILITY_API_KEY")
BASE_URL = "https://apivult.com/energy-volatility/v1"
 
HEADERS = {
    "X-RapidAPI-Key": ENERGY_API_KEY,
    "Content-Type": "application/json",
}

def get_market_volatility(
    commodity: str,  # 'natural_gas', 'electricity', 'crude_oil'
    hub: str,        # 'henry_hub', 'ercot', 'pjm', 'miso'
    lookback_days: int = 30,
) -> dict:
    """
    Fetch current volatility metrics for a commodity/hub pair.
    """
    response = requests.get(
        f"{BASE_URL}/volatility",
        headers=HEADERS,
        params={
            "commodity": commodity,
            "hub": hub,
            "lookback_days": lookback_days,
            "metrics": "historical,implied,realized,annualized",
        },
        timeout=10,
    )
    response.raise_for_status()
    return response.json()
 
# Example
vol_data = get_market_volatility("natural_gas", "henry_hub", lookback_days=30)
print(vol_data)

{
  "commodity": "natural_gas",
  "hub": "henry_hub",
  "spot_price": 3.847,
  "currency": "USD",
  "unit": "MMBtu",
  "timestamp": "2026-03-31T14:30:00Z",
  "volatility": {
    "historical_30d": 0.342,
    "historical_10d": 0.418,
    "implied": 0.389,
    "realized_daily": 0.021,
    "annualized": 0.342
  },
  "price_range_30d": {
    "min": 3.12,
    "max": 4.98,
    "mean": 3.71
  }
}

from scipy.stats import norm
 
def calculate_var(
    position_value: float,  # USD value of your energy position
    volatility: float,      # annualized volatility (e.g., 0.342 = 34.2%)
    holding_period_days: int = 1,
    confidence_level: float = 0.95,
) -> dict:
    """
    Calculate parametric Value at Risk for an energy position.
    Assumes normally distributed returns (appropriate for short horizons).
    """
    # Convert annualized vol to the holding period
    period_volatility = volatility * np.sqrt(holding_period_days / 252)
 
    # Z-score for confidence level
    z = norm.ppf(confidence_level)
 
    # VaR = position value × z-score × period volatility
    var = position_value * z * period_volatility
 
    return {
        "position_value_usd": position_value,
        "confidence_level": f"{confidence_level * 100:.0f}%",
        "holding_period_days": holding_period_days,
        "annualized_volatility": f"{volatility * 100:.1f}%",
        "var_usd": round(var, 2),
        "var_percent": round((var / position_value) * 100, 2),
        "worst_case_position": round(position_value - var, 2),
    }
 
# Example: $500,000 natural gas position
vol_data = get_market_volatility("natural_gas", "henry_hub")
annual_vol = vol_data["volatility"]["historical_30d"]
 
var_1d = calculate_var(500_000, annual_vol, holding_period_days=1)
var_5d = calculate_var(500_000, annual_vol, holding_period_days=5)
 
print("1-Day 95% VaR:", var_1d)
print("5-Day 95% VaR:", var_5d)

def detect_volatility_regime(hub: str, commodity: str) -> str:
    """
    Compare short-term vs long-term volatility to classify current regime.
    Returns: 'calm', 'elevated', 'stressed'
    """
    vol_10d = get_market_volatility(commodity, hub, lookback_days=10)
    vol_30d = get_market_volatility(commodity, hub, lookback_days=30)
 
    short_term = vol_10d["volatility"]["historical_10d"]
    long_term = vol_30d["volatility"]["historical_30d"]
 
    ratio = short_term / long_term
 
    if ratio < 0.8:
        return "calm"       # short-term vol well below long-term average
    elif ratio < 1.3:
        return "elevated"   # short-term approaching or matching long-term
    else:
        return "stressed"   # short-term significantly above long-term
 
def volatility_regime_report(positions: list[dict]) -> list[dict]:
    """Run regime detection across all positions."""
    report = []
    for pos in positions:
        regime = detect_volatility_regime(pos["hub"], pos["commodity"])
        vol_data = get_market_volatility(pos["commodity"], pos["hub"])
        var = calculate_var(pos["value_usd"], vol_data["volatility"]["historical_30d"])
 
        report.append({
            "position": pos["name"],
            "hub": pos["hub"],
            "regime": regime,
            "var_1d_usd": var["var_usd"],
            "spot_price": vol_data["spot_price"],
            "alert": regime == "stressed",
        })
 
    return report

import smtplib
from email.mime.text import MIMEText
 
RISK_THRESHOLDS = {
    "max_var_percent": 5.0,    # Alert if 1-day VaR > 5% of position
    "max_regime": "elevated",  # Alert if regime reaches 'stressed'
    "vol_spike_threshold": 0.5,  # Alert if annualized vol > 50%
}
 
def check_and_alert(positions: list[dict], smtp_config: dict) -> None:
    """Check all positions against risk thresholds and send alerts."""
    alerts = []
 
    for pos in positions:
        vol_data = get_market_volatility(pos["commodity"], pos["hub"])
        annual_vol = vol_data["volatility"]["annualized"]
        var = calculate_var(pos["value_usd"], annual_vol)
        regime = detect_volatility_regime(pos["hub"], pos["commodity"])
 
        if var["var_percent"] > RISK_THRESHOLDS["max_var_percent"]:
            alerts.append({
                "position": pos["name"],
                "type": "VaR_BREACH",
                "detail": f"1-Day VaR = {var['var_percent']:.1f}% (limit: {RISK_THRESHOLDS['max_var_percent']}%)",
            })
 
        if regime == "stressed":
            alerts.append({
                "position": pos["name"],
                "type": "VOLATILITY_STRESS",
                "detail": f"Volatility regime: STRESSED (10d vol >> 30d vol)",
            })
 
        if annual_vol > RISK_THRESHOLDS["vol_spike_threshold"]:
            alerts.append({
                "position": pos["name"],
                "type": "VOL_SPIKE",
                "detail": f"Annualized volatility: {annual_vol * 100:.1f}% (limit: 50%)",
            })
 
    if alerts:
        send_risk_alert(alerts, smtp_config)
 
def send_risk_alert(alerts: list[dict], smtp_config: dict) -> None:
    body = "⚠️ ENERGY RISK ALERT\n\n"
    for alert in alerts:
        body += f"[{alert['type']}] {alert['position']}\n  {alert['detail']}\n\n"
    body += f"\nGenerated: {datetime.utcnow().isoformat()}Z"
 
    msg = MIMEText(body)
    msg["Subject"] = f"Energy Risk Alert — {len(alerts)} threshold breach(es)"
    msg["From"] = smtp_config["from"]
    msg["To"] = smtp_config["to"]
 
    with smtplib.SMTP(smtp_config["host"], smtp_config["port"]) as server:
        server.starttls()
        server.login(smtp_config["user"], smtp_config["password"])
        server.send_message(msg)

def generate_daily_risk_summary(positions: list[dict]) -> pd.DataFrame:
    """Generate a daily risk summary table for the trading desk."""
    rows = []
 
    for pos in positions:
        vol_data = get_market_volatility(pos["commodity"], pos["hub"])
        annual_vol = vol_data["volatility"]["annualized"]
        var_1d = calculate_var(pos["value_usd"], annual_vol, holding_period_days=1)
        var_5d = calculate_var(pos["value_usd"], annual_vol, holding_period_days=5)
        regime = detect_volatility_regime(pos["hub"], pos["commodity"])
 
        rows.append({
            "Position": pos["name"],
            "Value (USD)": f"${pos['value_usd']:,.0f}",
            "Spot Price": f"${vol_data['spot_price']:.3f}",
            "Ann. Vol": f"{annual_vol * 100:.1f}%",
            "1D VaR (95%)": f"${var_1d['var_usd']:,.0f}",
            "5D VaR (95%)": f"${var_5d['var_usd']:,.0f}",
            "Regime": regime.upper(),
        })
 
    return pd.DataFrame(rows)
 
# Run daily at 7 AM before market open
positions = [
    {"name": "NG Long Q2-2026", "commodity": "natural_gas", "hub": "henry_hub", "value_usd": 500_000},
    {"name": "Power Short ERCOT", "commodity": "electricity", "hub": "ercot", "value_usd": 350_000},
    {"name": "Crude Hedge WTI", "commodity": "crude_oil", "hub": "cushing", "value_usd": 800_000},
]
 
summary = generate_daily_risk_summary(positions)
print(summary.to_string(index=False))

# Adapter pattern for ETRM integration
class EnergyVolatilityAdapter:
    """Adapts the Energy Volatility API to your ETRM system's expected interface."""
 
    def get_mark_to_market_vol(self, position_id: str, etrm_position: dict) -> float:
        """Map ETRM position data to API parameters and return annualized vol."""
        commodity = COMMODITY_MAP.get(etrm_position["product_type"])
        hub = HUB_MAP.get(etrm_position["delivery_point"])
 
        vol_data = get_market_volatility(commodity, hub)
        return vol_data["volatility"]["annualized"]
 
    def get_var_for_book(self, book: list[dict], confidence: float = 0.95) -> float:
        """Calculate portfolio VaR using API volatility data."""
        total_var_squared = 0
        for position in book:
            vol = self.get_mark_to_market_vol(position["id"], position)
            var = calculate_var(position["value_usd"], vol, confidence_level=confidence)
            total_var_squared += var["var_usd"] ** 2
 
        # Simple portfolio VaR (assumes zero correlation — conservative estimate)
        return np.sqrt(total_var_squared)