Alert kursów walut w Pythonie: USD/IDR

Alert porównuje nowe skuteczne obserwacje USD/IDR z poprzednią obserwacją, a nie czas odpytywania. Pierwsza obserwacja tworzy bazę, a stare dane i błędy nie zmieniają stanu.

• Do oceny IDR użyj sources["IDR"] i effective_at["IDR"]; globalne source jest konserwatywnym podsumowaniem.
• Pierwsza poprawna obserwacja to tylko baza. Po przekroczeniu wyślij jeden alert i uzbrój go ponownie dopiero po powrocie poniżej progu.
• Zapisz JSON w pliku tymczasowym, ustaw 0600 i zamień go przez os.replace, aby przerwanie nie uszkodziło stanu.
• Użyj timeoutu 10 sekund, klucza z EXCHANGERATE_API_KEY i odstępu co najmniej 60 sekund. Przy429 zachowaj stan i respektuj Retry-After.
• Pobierz pełny skrypt, a próg i ścieżkę stanu ustaw poza repozytorium. Wynik trafia tylko do konsoli.

Modeluj obserwację waluty

Do oceny IDR użyj sources["IDR"] i effective_at["IDR"]; globalne source jest konserwatywnym podsumowaniem.

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"],    }

Zdefiniuj przekroczenie

Pierwsza poprawna obserwacja to tylko baza. Po przekroczeniu wyślij jeden alert i uzbrój go ponownie dopiero po powrocie poniżej progu.

Bezpiecznie zapisz stan

Zapisz JSON w pliku tymczasowym, ustaw 0600 i zamień go przez os.replace, aby przerwanie nie uszkodziło stanu.

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

Ogranicz sieć i limit

Użyj timeoutu 10 sekund, klucza z EXCHANGERATE_API_KEY i odstępu co najmniej 60 sekund. Przy429 zachowaj stan i respektuj Retry-After. Domyślny interwał pięciu minut to około 8 640 wywołań w 30 dni; 60 sekund to około 43 200 i może przekroczyć wspólny limit.

Uruchom monitor

Pobierz pełny skrypt, a próg i ścieżkę stanu ustaw poza repozytorium. Wynik trafia tylko do konsoli. Domyślny maksymalny wiek danych to godzina. Ustaw FX_ALERT_MAX_AGE_SECONDS dla kursu dziennego lub zamkniętego rynku. 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

Przetestuj politykę bez czekania na rynek

Przetestuj bazę, próg, powtórzony timestamp, stare dane, ponowne uzbrojenie i drugi alert przez mock sieci.