Build an OPEC+ Production Change Alert System with the Energy Volatility API

pip install requests python-dotenv schedule smtplib

APIVULT_API_KEY=YOUR_API_KEY
SLACK_WEBHOOK_URL=your_slack_webhook
[email protected]
SMTP_HOST=smtp.company.com
SMTP_PASSWORD=your_smtp_password

import os
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv
 
load_dotenv()
 
API_KEY = os.getenv("APIVULT_API_KEY")
BASE_URL = "https://apivult.com/api/energy-volatility"
 
def get_current_prices() -> dict:
    """
    Fetch current prices for key energy benchmarks.
    """
    headers = {"X-RapidAPI-Key": API_KEY}
 
    response = requests.get(
        f"{BASE_URL}/prices/current",
        headers=headers,
        params={
            "commodities": "brent_crude,wti_crude,natural_gas,ttf_gas",
            "include_change": True,
            "include_volume": True
        },
        timeout=10
    )
    response.raise_for_status()
    return response.json()
 
def get_volatility_metrics(commodity: str, lookback_hours: int = 24) -> dict:
    """
    Get volatility metrics for a commodity over a lookback window.
    """
    headers = {"X-RapidAPI-Key": API_KEY}
 
    since = (datetime.utcnow() - timedelta(hours=lookback_hours)).isoformat()
 
    response = requests.get(
        f"{BASE_URL}/volatility",
        headers=headers,
        params={
            "commodity": commodity,
            "since": since,
            "metrics": "realized_vol,price_range,var_95,var_99"
        },
        timeout=10
    )
    response.raise_for_status()
    return response.json()
 
# Get current market snapshot
prices = get_current_prices()
for commodity in prices["data"]:
    name = commodity["commodity"]
    price = commodity["price"]
    change_pct = commodity["change_1h_pct"]
    print(f"{name}: ${price:.2f} ({change_pct:+.2f}% last hour)")

from collections import deque
import statistics
 
class VolatilitySpikDetector:
    """
    Detects volatility spikes relative to a rolling baseline.
    Uses 7-day rolling standard deviation to contextualize current moves.
    """
 
    def __init__(self, commodity: str, spike_threshold_sigma: float = 2.0):
        self.commodity = commodity
        self.threshold_sigma = spike_threshold_sigma
        self.price_history = deque(maxlen=2016)  # 1 week at 5-min intervals
        self.last_alert_time = None
        self.alert_cooldown_minutes = 60  # prevent alert storms
 
    def add_price(self, price: float, timestamp: datetime):
        """Add a new price observation."""
        self.price_history.append({"price": price, "timestamp": timestamp})
 
    def detect_spike(self, current_price: float) -> dict:
        """
        Check if current price represents a volatility spike.
        Returns detection result with severity level.
        """
        if len(self.price_history) < 288:  # Need 24h minimum
            return {"spike_detected": False, "reason": "insufficient_history"}
 
        # Calculate rolling returns
        prices = [p["price"] for p in self.price_history]
        returns = [
            (prices[i] - prices[i-1]) / prices[i-1]
            for i in range(1, len(prices))
        ]
 
        baseline_vol = statistics.stdev(returns)
        last_return = (current_price - prices[-1]) / prices[-1]
 
        # How many standard deviations is the current move?
        if baseline_vol == 0:
            return {"spike_detected": False, "reason": "zero_volatility"}
 
        z_score = abs(last_return) / baseline_vol
 
        # Determine severity
        if z_score >= 4.0:
            severity = "CRITICAL"
        elif z_score >= 3.0:
            severity = "HIGH"
        elif z_score >= self.threshold_sigma:
            severity = "MEDIUM"
        else:
            return {
                "spike_detected": False,
                "z_score": round(z_score, 2),
                "last_return_pct": round(last_return * 100, 3)
            }
 
        # Check cooldown to avoid duplicate alerts
        now = datetime.utcnow()
        if self.last_alert_time:
            minutes_since_alert = (now - self.last_alert_time).total_seconds() / 60
            if minutes_since_alert < self.alert_cooldown_minutes:
                return {"spike_detected": False, "reason": "cooldown_active"}
 
        self.last_alert_time = now
 
        return {
            "spike_detected": True,
            "severity": severity,
            "z_score": round(z_score, 2),
            "current_price": current_price,
            "last_return_pct": round(last_return * 100, 3),
            "baseline_volatility": round(baseline_vol * 100, 4),
            "commodity": self.commodity,
            "detected_at": now.isoformat()
        }

import smtplib
import json
from email.mime.text import MIMEText
 
def send_slack_alert(webhook_url: str, spike_data: dict):
    """Send formatted Slack alert for volatility spike."""
    severity_emoji = {
        "CRITICAL": "🚨",
        "HIGH": "⚠️",
        "MEDIUM": "📊"
    }
 
    emoji = severity_emoji.get(spike_data["severity"], "📊")
    change_str = f"{spike_data['last_return_pct']:+.2f}%"
 
    message = {
        "text": f"{emoji} Energy Volatility Alert",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": (
                        f"*{emoji} {spike_data['severity']} VOLATILITY SPIKE*\n"
                        f"*Commodity:* {spike_data['commodity'].replace('_', ' ').title()}\n"
                        f"*Price:* ${spike_data['current_price']:.2f} "
                        f"({change_str} last interval)\n"
                        f"*Z-Score:* {spike_data['z_score']}σ above baseline\n"
                        f"*Time:* {spike_data['detected_at']}"
                    )
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": (
                        "*Recommended Actions:*\n"
                        "• Review open positions for this commodity\n"
                        "• Check OPEC+ news feeds for announcement context\n"
                        "• Consider activating hedging procedures if >3σ"
                    )
                }
            }
        ]
    }
 
    response = requests.post(webhook_url, json=message, timeout=10)
    return response.status_code == 200
 
def send_email_alert(spike_data: dict):
    """Send email alert for critical volatility events."""
    smtp_host = os.getenv("SMTP_HOST")
    smtp_pass = os.getenv("SMTP_PASSWORD")
    recipient = os.getenv("ALERT_EMAIL")
 
    subject = (
        f"[{spike_data['severity']}] Energy Volatility Alert: "
        f"{spike_data['commodity']} moved {spike_data['last_return_pct']:+.2f}%"
    )
 
    body = f"""
Energy Volatility Alert
=======================
 
Severity: {spike_data['severity']}
Commodity: {spike_data['commodity']}
Current Price: ${spike_data['current_price']:.2f}
Price Change: {spike_data['last_return_pct']:+.2f}%
Statistical Significance: {spike_data['z_score']} standard deviations
 
This move is {spike_data['z_score']}x larger than the typical 5-minute
price change over the past 7 days.
 
Detected at: {spike_data['detected_at']} UTC
 
---
Powered by Energy Volatility API | apivult.com
    """
 
    msg = MIMEText(body)
    msg["Subject"] = subject
    msg["From"] = "[email protected]"
    msg["To"] = recipient
 
    with smtplib.SMTP(smtp_host, 587) as server:
        server.starttls()
        server.login("[email protected]", smtp_pass)
        server.send_message(msg)

import schedule
import time
 
# Initialize detectors for each commodity
detectors = {
    "brent_crude": VolatilitySpikDetector("brent_crude", spike_threshold_sigma=2.5),
    "wti_crude": VolatilitySpikDetector("wti_crude", spike_threshold_sigma=2.5),
    "natural_gas": VolatilitySpikDetector("natural_gas", spike_threshold_sigma=3.0)
}
 
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK_URL")
 
def monitoring_tick():
    """Run every 5 minutes during market hours."""
    try:
        prices = get_current_prices()
 
        for commodity_data in prices["data"]:
            commodity = commodity_data["commodity"]
            price = commodity_data["price"]
 
            if commodity not in detectors:
                continue
 
            detector = detectors[commodity]
            detector.add_price(price, datetime.utcnow())
 
            spike = detector.detect_spike(price)
 
            if spike.get("spike_detected"):
                print(f"SPIKE DETECTED: {spike}")
 
                # Route alerts based on severity
                send_slack_alert(SLACK_WEBHOOK, spike)
 
                if spike["severity"] == "CRITICAL":
                    send_email_alert(spike)
 
    except Exception as e:
        print(f"Monitoring tick error: {e}")
 
# Schedule polling every 5 minutes
schedule.every(5).minutes.do(monitoring_tick)
 
print("Energy volatility monitoring started...")
while True:
    schedule.run_pending()
    time.sleep(30)

OPEC_MEETINGS_2026 = [
    {"date": "2026-06-01", "type": "ministerial", "impact": "high"},
    {"date": "2026-11-30", "type": "ministerial", "impact": "high"},
    # Add from official OPEC calendar
]
 
def check_opec_calendar_alert():
    """Alert the team 24 hours before scheduled OPEC meetings."""
    tomorrow = (datetime.utcnow() + timedelta(days=1)).date()
 
    for meeting in OPEC_MEETINGS_2026:
        meeting_date = datetime.fromisoformat(meeting["date"]).date()
 
        if meeting_date == tomorrow:
            send_slack_alert(SLACK_WEBHOOK, {
                "severity": "HIGH",
                "commodity": "brent_crude",
                "current_price": 0,
                "last_return_pct": 0,
                "z_score": 0,
                "detected_at": datetime.utcnow().isoformat(),
                "custom_message": f"⏰ OPEC+ {meeting['type'].title()} Meeting tomorrow. "
                                  f"Expect elevated volatility. Pre-position hedges if needed."
            })