สร้างการแจ้งเตือนอัตรา USD/IDR ด้วย Python

การแจ้งเตือนควรเทียบ observation ของ USD/IDR ที่มีผลใหม่กับ observation ก่อนหน้า ไม่ใช่เทียบเวลาที่ polling ครั้งก่อน ครั้งแรกสร้าง baseline ส่วนข้อมูลเก่าหรือ error ไม่เปลี่ยน state

สร้างโมเดล observation

ใช้ sources["IDR"] และ effective_at["IDR"] เพื่อประเมิน IDR ส่วน source รวมเป็นเพียงสรุปแบบ conservative

from decimal import Decimal

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

กำหนดนโยบายการข้าม threshold

observation แรกเป็น baseline เท่านั้น หลังข้าม threshold ให้แจ้งครั้งเดียว และ arm ใหม่เมื่อ observation ที่ถูกต้องกลับมาต่ำกว่า threshold

บันทึก state อย่างปลอดภัย

เขียน JSON ลงไฟล์ชั่วคราว ตั้ง mode0600 แล้วใช้ os.replace เพื่อป้องกัน state เสียหาย

import json
import os
import tempfile
from pathlib import Path

def 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

จำกัด network และ quota

ใช้ timeout 10 วินาที เก็บคีย์ใน EXCHANGERATE_API_KEY และ polling ห่างอย่างน้อย60วินาที เมื่อ429 ให้เก็บ state และทำตาม Retry-After ค่าเริ่มต้นทุกห้านาทีใช้ประมาณ 8,640 calls ใน30วัน ส่วนทุก60วินาทีใช้ประมาณ43,200และอาจเกิน quota ร่วมรายเดือน

รัน monitor

ดาวน์โหลดสคริปต์เต็ม ตั้งค่า threshold และ state path นอก source control และแสดงผลใน console เท่านั้น ค่าเริ่มต้นอายุข้อมูลสูงสุดคือหนึ่งชั่วโมง ตั้ง FX_ALERT_MAX_AGE_SECONDS หากยอมรับข้อมูลอ้างอิงรายวันหรือตลาดปิด python-exchange-rate-alerts.py

chmod 700 python-exchange-rate-alerts.py
export 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

ทดสอบนโยบายโดยไม่ต้องรอตลาด

ทดสอบ baseline, crossing, timestamp ซ้ำ, ข้อมูลเก่า, rearm และ crossing ครั้งที่สองด้วย network mock