提醒应比较每个新的有效 USD/IDR 观测与上一次观测,而不是比较两次轮询时间。第一次只建立基线,旧数据和错误不会改变状态。
- 判断 IDR 时使用
sources["IDR"]和effective_at["IDR"]。全局 source 只是所有请求币种的保守摘要。 - 第一次有效观测只保存基线。跨越阈值后提醒一次并解除武装,只有之后有效观测回到阈值以内才重新武装。
- 先写入同目录临时 JSON,设置 0600,再用
os.replace替换,避免中断留下半个状态文件。 - 请求使用十秒超时,密钥放在
EXCHANGERATE_API_KEY。轮询至少间隔60秒;429时保留状态并遵守Retry-After。 - 下载完整脚本,在版本控制之外配置阈值和状态路径。脚本只打印本地控制台提醒。
建模单币种观测
判断 IDR 时使用 sources["IDR"] 和 effective_at["IDR"]。全局 source 只是所有请求币种的保守摘要。
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"], }定义跨越策略
第一次有效观测只保存基线。跨越阈值后提醒一次并解除武装,只有之后有效观测回到阈值以内才重新武装。
| 情况 | 状态变化 | 提醒 |
|---|---|---|
| 第一次有效观测 | 保存基线 | 否 |
| 低于阈值的新观测 | armed=true | 否 |
| 已武装时跨越阈值 | armed=false | 一次 |
| 重复/旧观测或错误 | 保留状态 | 否 |
安全保存状态
先写入同目录临时 JSON,设置 0600,再用 os.replace 替换,避免中断留下半个状态文件。
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限制网络和配额
请求使用十秒超时,密钥放在 EXCHANGERATE_API_KEY。轮询至少间隔60秒;429时保留状态并遵守 Retry-After。 默认五分钟间隔在30天约调用8,640次;60秒间隔约43,200次,可能超过共享月度配额。
请求使用十秒超时,密钥放在 EXCHANGERATE_API_KEY。轮询至少间隔60秒;429时保留状态并遵守 Retry-After。
运行完整监控
下载完整脚本,在版本控制之外配置阈值和状态路径。脚本只打印本地控制台提醒。 默认最大数据年龄为一小时。如果接受每日参考或市场关闭数据,请设置 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不等待市场也能测试策略
用网络 mock 测试第一次、阈值内移动、跨越、重复时间、旧数据、重新武装和第二次跨越。