การแจ้งเตือนควรเทียบ observation ของ USD/IDR ที่มีผลใหม่กับ observation ก่อนหน้า ไม่ใช่เทียบเวลาที่ polling ครั้งก่อน ครั้งแรกสร้าง baseline ส่วนข้อมูลเก่าหรือ error ไม่เปลี่ยน state
- ใช้
sources["IDR"]และeffective_at["IDR"]เพื่อประเมิน IDR ส่วน source รวมเป็นเพียงสรุปแบบ conservative - observation แรกเป็น baseline เท่านั้น หลังข้าม threshold ให้แจ้งครั้งเดียว และ arm ใหม่เมื่อ observation ที่ถูกต้องกลับมาต่ำกว่า threshold
- เขียน JSON ลงไฟล์ชั่วคราว ตั้ง mode0600 แล้วใช้
os.replaceเพื่อป้องกัน state เสียหาย - ใช้ timeout 10 วินาที เก็บคีย์ใน
EXCHANGERATE_API_KEYและ polling ห่างอย่างน้อย60วินาที เมื่อ429 ให้เก็บ state และทำตามRetry-After - ดาวน์โหลดสคริปต์เต็ม ตั้งค่า threshold และ state path นอก source control และแสดงผลใน console เท่านั้น
สร้างโมเดล observation
ใช้ sources["IDR"] และ effective_at["IDR"] เพื่อประเมิน IDR ส่วน source รวมเป็นเพียงสรุปแบบ conservative
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"], }กำหนดนโยบายการข้าม threshold
observation แรกเป็น baseline เท่านั้น หลังข้าม threshold ให้แจ้งครั้งเดียว และ arm ใหม่เมื่อ observation ที่ถูกต้องกลับมาต่ำกว่า threshold
| สถานการณ์ | การเปลี่ยน state | การแจ้งเตือน |
|---|---|---|
| observation ที่ถูกต้องครั้งแรก | บันทึก baseline | ไม่ |
| observation ต่ำกว่า threshold | armed=true | ไม่ |
| ข้าม threshold ขณะ armed | armed=false | หนึ่งครั้ง |
| ข้อมูลซ้ำ/เก่าหรือ error | คง state | ไม่ |
บันทึก state อย่างปลอดภัย
เขียน JSON ลงไฟล์ชั่วคราว ตั้ง mode0600 แล้วใช้ os.replace เพื่อป้องกัน state เสียหาย
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จำกัด network และ quota
ใช้ timeout 10 วินาที เก็บคีย์ใน EXCHANGERATE_API_KEY และ polling ห่างอย่างน้อย60วินาที เมื่อ429 ให้เก็บ state และทำตาม Retry-After ค่าเริ่มต้นทุกห้านาทีใช้ประมาณ 8,640 calls ใน30วัน ส่วนทุก60วินาทีใช้ประมาณ43,200และอาจเกิน quota ร่วมรายเดือน
ใช้ timeout 10 วินาที เก็บคีย์ใน EXCHANGERATE_API_KEY และ polling ห่างอย่างน้อย60วินาที เมื่อ429 ให้เก็บ state และทำตาม Retry-After
รัน monitor
ดาวน์โหลดสคริปต์เต็ม ตั้งค่า threshold และ state path นอก source control และแสดงผลใน console เท่านั้น ค่าเริ่มต้นอายุข้อมูลสูงสุดคือหนึ่งชั่วโมง ตั้ง FX_ALERT_MAX_AGE_SECONDS หากยอมรับข้อมูลอ้างอิงรายวันหรือตลาดปิด 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ทดสอบนโยบายโดยไม่ต้องรอตลาด
ทดสอบ baseline, crossing, timestamp ซ้ำ, ข้อมูลเก่า, rearm และ crossing ครั้งที่สองด้วย network mock