QuantEngineByItz/src/quant_engine/auto_fill_velocity_v1.py

#!/usr/bin/env python3
"""
WBS-9.3: Velocity 자동 충전 절차

조건: velocity_1d IS NULL AND (close_price AND previous_close_price) IS NOT NULL
"""

def calculate_velocity_1d(close_price: float, previous_close: float) -> float:
    """
    Calculate 1-day velocity (percentage change).

    입력: close_price, previous_close
    출력: velocity_1d (%, e.g., 2.5 = +2.5%)
    """
    if previous_close is None or previous_close == 0:
        return None

    velocity = ((close_price - previous_close) / previous_close) * 100
    return round(velocity, 2)


def calculate_velocity_5d(closes: list[float]) -> float:
    """
    Calculate 5-day velocity (5-period momentum).

    입력: close 가격 리스트 (최소 5개)
    출력: velocity_5d (%, e.g., 5.2 = +5.2%)
    """
    if not closes or len(closes) < 5:
        return None

    current = closes[-1]
    five_days_ago = closes[-5]

    if five_days_ago is None or five_days_ago == 0:
        return None

    velocity = ((current - five_days_ago) / five_days_ago) * 100
    return round(velocity, 2)


def auto_fill_velocity_1d(row: dict) -> dict:
    """
    자동 충전: velocity_1d 필드

    입력:
      row: 현재 행 (velocity_1d = None)
      필수: close_price, previous_close_price

    출력:
      row (velocity_1d 채워짐) 또는 원본 (실패시)
    """
    if row.get("velocity_1d") is not None:
        return row

    close = row.get("close_price")
    prev_close = row.get("previous_close_price")

    if close is None or prev_close is None:
        return row

    velocity_1d = calculate_velocity_1d(close, prev_close)
    if velocity_1d is not None:
        row["velocity_1d"] = velocity_1d
        row["_fill_source"] = "auto_fill_velocity_v1"

    return row


def auto_fill_velocity_5d(row: dict, historical_closes: list[float] = None) -> dict:
    """
    자동 충전: velocity_5d 필드

    입력:
      row: 현재 행
      historical_closes: 과거 close 가격 (최소 5일)

    출력:
      row (velocity_5d 채워짐) 또는 원본 (실패시)
    """
    if row.get("velocity_5d") is not None:
        return row

    if not historical_closes:
        return row

    velocity_5d = calculate_velocity_5d(historical_closes)
    if velocity_5d is not None:
        row["velocity_5d"] = velocity_5d
        row["_fill_source"] = "auto_fill_velocity_v1"

    return row