我如何在周末通过简单的 Python 过滤器修复过时的汇率数据

Source: Dev.to

我们要解决的问题

我的三层过滤逻辑

核心验证函数

def is_valid_trading_data(price, timestamp, last_price, last_timestamp):
    # Timestamp not moving = invalid update
    if timestamp <= last_timestamp:
        return False

    # Not a trading day = skip entirely
    if not is_trading_day():
        return False

    return True

完整的 WebSocket 实现（以 AllTick API 为例）

import json
import websocket

WS_DOMAIN = "wss://apis.alltick.co"
ws_url = f"{WS_DOMAIN}/your_endpoint"

last_price = None
last_ts = None

def is_trading_day():
    # Implement your logic to determine if today is a trading day
    ...

def on_message(ws, message):
    data = json.loads(message)  # assuming JSON payload
    current_price = data.get('price')
    current_ts = data.get('timestamp')

    # Skip on non-trading days
    if not is_trading_day():
        print("Non-trading day — skipped")
        return

    # Skip stale, unchanged prices
    if current_price == last_price:
        print("Price unchanged — filtering stale data")
        return

    # Only process valid data here
    print(f"Valid exchange rate: {current_price}")
    global last_price, last_ts
    last_price = current_price
    last_ts = current_ts

ws = websocket.WebSocketApp(ws_url, on_message=on_message)

生产环境改进（适用于上线的开发者）

关键要点

• API 并不知道你的使用场景。
• 你必须在原始数据和系统之间构建验证层。
• 这个小过滤器彻底消除了误报，清理了数据管道，节省了数小时的调试时间。
• 如果你在使用实时 API、WebSocket 数据流或金融数据，这种模式同样能帮你省事。