Crear una alerta de tipo de cambio en Python con USD/IDR

Una alerta debe comparar cada observación efectiva nueva con la anterior, no cada consulta HTTP con la consulta anterior. La primera observación crea una base, los datos antiguos y los errores no cambian el estado, y solo un cruce válido produce un aviso.

• Usa sources["IDR"] y effective_at["IDR"] para juzgar el dato de IDR. Los campos globales son un resumen conservador de todas las monedas solicitadas.
• La primera observación válida solo crea una base. Tras un cruce, desarma la alerta y vuelve a armarla únicamente cuando una observación válida posterior regrese por debajo del umbral.
• Escribe el JSON en un archivo temporal, usa modo 0600 y reemplaza el archivo con os.replace. Así una interrupción no deja una base incompleta.
• Usa un timeout de diez segundos, conserva la clave en EXCHANGERATE_API_KEY y espera al menos 60 segundos entre consultas. En 429 conserva el estado y respeta Retry-After.
• Descarga el script completo, configura el umbral y la ruta de estado fuera del control de versiones. Solo imprime en la consola; no envía mensajes externos.

Modela la observación por moneda

Usa sources["IDR"] y effective_at["IDR"] para juzgar el dato de IDR. Los campos globales son un resumen conservador de todas las monedas solicitadas.

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 la política de cruce

La primera observación válida solo crea una base. Tras un cruce, desarma la alerta y vuelve a armarla únicamente cuando una observación válida posterior regrese por debajo del umbral.

Guarda el estado de forma segura

Escribe el JSON en un archivo temporal, usa modo 0600 y reemplaza el archivo con os.replace. Así una interrupción no deja una base incompleta.

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        raise

Limita red y cuota

Usa un timeout de diez segundos, conserva la clave en EXCHANGERATE_API_KEY y espera al menos 60 segundos entre consultas. En 429 conserva el estado y respeta Retry-After. El valor predeterminado de cinco minutos consume unas 8.640 llamadas en 30 días; un intervalo de 60 segundos consume unas 43.200 y puede superar la cuota compartida.

Ejecuta el monitor completo

Descarga el script completo, configura el umbral y la ruta de estado fuera del control de versiones. Solo imprime en la consola; no envía mensajes externos. La edad máxima predeterminada es una hora. Configura FX_ALERT_MAX_AGE_SECONDS si aceptas una referencia diaria o un mercado cerrado. python-exchange-rate-alerts.py

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.py

Prueba la política sin esperar al mercado

Prueba la primera observación, un movimiento menor, un cruce, una marca temporal repetida, datos antiguos, el rearme y un segundo cruce mediante un mock de red.