주말에 오래된 환율 데이터를 간단한 Python 필터로 해결한 방법

Source: Dev.to

우리가 해결하고자 하는 문제

나의 3‑Layer 필터링 로직

핵심 검증 함수

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 피드, 혹은 금융 데이터를 다루고 있다면 — 이 패턴이 여러분에게도 큰 도움이 될 것입니다.