實時告警系統數據層設計：去抖與狀態機

import asyncio, json, os, time
import websockets

URL = "wss://api.tiqex.io/api/v1/market/stream"
API_KEY = os.environ["TIQEX_API_KEY"]

DEBOUNCE_SEC = 5.0          # 時間去抖窗口
HYSTERESIS_RATIO = 0.005    # 滯回幅度：0.5%
MIN_NOTIFY_INTERVAL = 60    # 同一規則最短通知間隔（秒）


class AlertRule:
    """帶滯回與時間去抖的告警規則，狀態顯式建模。"""

    def __init__(self, symbol: str, threshold: float, above: bool = True):
        self.symbol = symbol
        self.threshold = threshold
        self.above = above
        # 滯回：恢復閾值比觸發閾值退讓 HYSTERESIS_RATIO
        self.recover = (threshold * (1 - HYSTERESIS_RATIO) if above
                        else threshold * (1 + HYSTERESIS_RATIO))
        self.state = "ARMED"
        self.last_notify = 0.0
        self.last_change = 0.0

    def on_tick(self, price: float) -> str | None:
        now = time.monotonic()

        if self.state == "ARMED":
            crossed = price >= self.threshold if self.above else price <= self.threshold
            if crossed and now - self.last_change >= DEBOUNCE_SEC:
                self.state = "TRIGGERED"
                self.last_change = now
                return self._notify("triggered", price)

        elif self.state == "TRIGGERED":
            recovered = price <= self.recover if self.above else price >= self.recover
            if recovered and now - self.last_change >= DEBOUNCE_SEC:
                self.state = "ARMED"
                self.last_change = now
                return self._notify("recovered", price)

        return None

    def _notify(self, kind: str, price: float) -> str | None:
        now = time.monotonic()
        if now - self.last_notify < MIN_NOTIFY_INTERVAL:
            return None                      # 命中通知預算，靜默丟棄
        self.last_notify = now
        return f"[{kind}] {self.symbol} @ {price}"


RULES = [
    AlertRule("BTCUSDT", 80000, above=True),
    AlertRule("ETHUSDT", 2500, above=False),
]
BY_SYMBOL = {r.symbol: r for r in RULES}


async def main():
    symbols = list(BY_SYMBOL)
    backoff = 1.0
    while True:
        try:
            async with websockets.connect(
                URL, additional_headers={"X-API-Key": API_KEY}, open_timeout=10,
            ) as ws:
                await ws.send(json.dumps({
                    "action": "subscribe", "market": "crypto", "symbols": symbols,
                }))
                backoff = 1.0
                async for raw in ws:
                    frame = json.loads(raw)
                    data = frame.get("data") or {}
                    if data.get("type") != "tick":
                        continue
                    tick = data["tick"]
                    rule = BY_SYMBOL.get(tick["symbol"])
                    if not rule:
                        continue
                    message = rule.on_tick(float(tick["last"]))
                    if message:
                        print(message)      # 這裡換成你的通知渠道
        except Exception as exc:            # noqa: BLE001
            print(f"連接中斷: {exc}")
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 60)
            # 注意：rule.state 保留在內存中，重連後不會重複觸發


asyncio.run(main())