A useful alert compares each newly effective USD/IDR observation with the last one, not each HTTP poll with the last poll. The example suppresses the first observation, ignores stale data and errors, and prints a console alert when the move crosses the configured threshold.
- Use
sources["IDR"]andeffective_at["IDR"]to judge the currency being monitored. - The first observation establishes a baseline and never emits an alert.
- Persist state with an atomic replace and file mode 0600; never log the API key.
- A stale observation, network failure, 401, or 429 retains the previous state and cannot create or re-arm an alert.
- The default five-minute poll is about 8,640 calls in 30 days; a 60-second poll is about 43,200, so account for the shared monthly quota before choosing it.
Model the per-currency observation
The envelope source and data_updated_at fields summarize all requested currencies. An alert for IDR should use sources["IDR"] and effective_at["IDR"] instead, so another currency cannot make the monitored observation look fresher or older than it is.
from decimal import Decimaldef idr_observation(payload: dict) -> dict: return { "rate": Decimal(str(payload["rates"]["IDR"])), "source": payload["sources"]["IDR"], "effective_at": payload["effective_at"]["IDR"], "market_session": payload["market_session"], }Define a crossing policy
The first valid observation is only a baseline. When a new effective observation moves at least the threshold from the previous one, emit one alert and disarm. Do not emit another alert until a later valid observation returns below the threshold; then re-arm. This deduplicates repeated polls of the same crossing.
| Situation | State change | Alert |
|---|---|---|
| First valid observation | Store baseline | No |
| New observation, move below threshold | Set armed=true | No |
| New observation crosses threshold while armed | Set armed=false | One |
| Repeated/stale observation or request error | Keep state | No |
Persist state safely
Write a complete JSON object to a temporary file in the same directory, set its mode to 0600, then replace the state path with os.replace. A process interruption cannot leave a half-written baseline. Store the last rate, effective timestamp, source, market session, and whether the threshold is armed.
import jsonimport osimport tempfilefrom pathlib import Pathdef save_state(path: Path, state: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent, text=True) try: os.fchmod(fd, 0o600) with os.fdopen(fd, "w") as handle: json.dump(state, handle, sort_keys=True) handle.write("\n") os.replace(temporary, path) except Exception: try: os.unlink(temporary) except OSError: pass raiseBound the network and quota
Use a ten-second request timeout and never follow redirects while an Authorization header is present. Keep the key in EXCHANGERATE_API_KEY; it does not belong in a URL or log line. The script defaults to one request every five minutes (about 8,640 calls in 30 days), requiring Basic or above. A 60-second interval uses about 43,200 monthly calls, also above Free’s 5,000-call quota. On 429, wait for Retry-After when present and preserve the last state.
A timeout, 401, 429, 5xx response, malformed JSON response, missing per-currency freshness field, or stale effective timestamp leaves the previous baseline untouched. A closed-market response can still be valid; keep source and market_session beside it.
Run the complete monitor
Download the complete Python alert script, make it executable, and set the threshold, maximum age, and state path outside source control. The default freshness window is one hour; configure a longer FX_ALERT_MAX_AGE_SECONDS if a closed-market or daily-reference observation is intentional. It prints console notifications only; adding email, Slack, or another external destination requires a separate integration and secret policy.
chmod 700 python-exchange-rate-alerts.pyexport EXCHANGERATE_API_KEY="your-key"export FX_ALERT_THRESHOLD_PCT="1.0"export FX_ALERT_POLL_SECONDS="300"export FX_ALERT_STATE_PATH="$HOME/.local/state/fx-alert.json"python3 python-exchange-rate-alerts.pyTest the policy without waiting for the market
Unit-test the state transition with Decimal values: first observation, a move below threshold, a crossing, a repeated effective timestamp, a stale response, a return below threshold, and a second crossing. Mock the network boundary rather than fabricating market data in the script. KeyboardInterrupt stops the loop cleanly and does not delete the baseline.