Gold price alerts with Python and Slack

export GOLD_THRESHOLD_USD="4500"
export GOLD_ALERT_DB="/absolute/private/directory/gold-alert.sqlite3"
# Set SLACK_WEBHOOK_URL through your scheduler's secret configuration.
# Keep its value out of shell history and source control.

"""Run once per scheduled check. Python 3.10+; standard library only."""
import json
import os
import sqlite3
import sys
from contextlib import closing
from datetime import datetime, timezone
from decimal import Decimal
from urllib.parse import urlparse
from urllib.request import Request, urlopen

PRICE_URL = "https://api.goldprice.dev/v1/prices?symbol=XAU-USD-SPOT"


def fetch_price():
    request = Request(
        PRICE_URL,
        headers={"Accept": "application/json", "User-Agent": "goldprice-alert-tutorial/1.0"},
    )
    with urlopen(request, timeout=15) as response:
        row = json.load(response)["symbols"][0]
    if (row.get("symbol"), row.get("quote_currency"), row.get("unit")) != (
        "XAU", "USD", "troy_ounce"
    ) or row.get("is_stale") is not False:
        raise ValueError("Unexpected or stale quote")
    observed = datetime.fromisoformat(row["computed_at"].replace("Z", "+00:00"))
    age = (datetime.now(timezone.utc) - observed).total_seconds()
    price = Decimal(row["price"])
    if not price.is_finite() or price <= 0 or not 0 <= age <= 900:
        raise ValueError("Invalid price or quote older than 15 minutes")
    return price


def record_crossing(connection, threshold, price):
    """Commit before delivery: at most one attempt per observed upward crossing."""
    connection.execute("CREATE TABLE IF NOT EXISTS alert (id INTEGER PRIMARY KEY CHECK(id=1), threshold TEXT, above INTEGER)")
    previous = connection.execute("SELECT threshold, above FROM alert WHERE id=1").fetchone()
    above = price >= threshold
    crossing = previous is not None and previous[0] == str(threshold) and not previous[1] and above
    connection.execute("INSERT OR REPLACE INTO alert VALUES (1, ?, ?)", (str(threshold), above))
    connection.commit()
    return crossing


def send_slack(webhook, price, threshold):
    payload = json.dumps({"text": f"Gold crossed ${threshold} per troy ounce. Observed XAU/USD: ${price}. Indicative spot price from goldprice.dev."}).encode()
    request = Request(webhook, data=payload, headers={"Content-Type": "application/json"}, method="POST")
    with urlopen(request, timeout=15) as response:
        if response.status != 200 or response.read().strip() != b"ok":
            raise ValueError("Slack did not confirm delivery")


def main():
    # Choose a private, persistent directory; do not share the database between machines.
    os.umask(0o077)
    threshold = Decimal(os.environ["GOLD_THRESHOLD_USD"])
    if not threshold.is_finite() or threshold <= 0:
        raise ValueError("Threshold must be a positive finite number")
    webhook = os.environ["SLACK_WEBHOOK_URL"]
    target = urlparse(webhook)
    if target.scheme != "https" or target.netloc != "hooks.slack.com" or not target.path.startswith("/services/"):
        raise ValueError("Expected a Slack incoming webhook URL")
    database = os.environ.get("GOLD_ALERT_DB", "gold-alert.sqlite3")
    with closing(sqlite3.connect(database, timeout=20)) as connection, connection:
        # Serialize overlapping scheduled runs on this machine, including the quote fetch.
        connection.execute("BEGIN IMMEDIATE")
        price = fetch_price()
        crossing = record_crossing(connection, threshold, price)
    if not crossing:
        print("Baseline recorded or no new upward crossing.")
        return
    try:
        send_slack(webhook, price, threshold)
    except Exception:
        # A timeout can happen AFTER Slack accepts the message. Do not retry blindly.
        print("Alert attempt failed or delivery is uncertain. Check Slack and scheduler logs; this crossing will not be retried automatically.", file=sys.stderr)
        raise SystemExit(1)
    print("Slack confirmed the alert.")


if __name__ == "__main__":
    try:
        main()
    except Exception:
        # Exception URLs may contain the secret webhook; never print them.
        print("Check failed. Check configuration, connectivity and API status; no automatic retry in this run.", file=sys.stderr)
        raise SystemExit(1)