okx-codex-trader/okx_codex_trader/strategy.py

from typing import Mapping

from okx_codex_trader.models import Candle, TradeSignal


def simple_moving_average(candles: list[Candle], window: int) -> list[float | None]:
    averages: list[float | None] = []
    running_total = 0.0

    for index, candle in enumerate(candles):
        running_total += candle.close
        if index >= window:
            running_total -= candles[index - window].close
        if index + 1 < window:
            averages.append(None)
            continue
        averages.append(running_total / window)

    return averages


def validate_signal(payload: Mapping[str, object]) -> TradeSignal:
    required_keys = {
        "action",
        "confidence",
        "leverage",
        "entry_price",
        "take_profit_price",
        "stop_loss_price",
        "reason",
    }
    if set(payload) != required_keys:
        raise ValueError("signal shape is invalid")

    action = payload["action"]
    if action not in {"long", "short", "flat"}:
        raise ValueError("signal action is invalid")

    confidence = payload["confidence"]
    if isinstance(confidence, bool):
        raise ValueError("signal confidence is invalid")
    if not isinstance(confidence, int | float) or not 0 <= float(confidence) <= 1:
        raise ValueError("signal confidence is invalid")

    leverage = payload["leverage"]
    if isinstance(leverage, bool):
        raise ValueError("signal leverage is invalid")
    if not isinstance(leverage, int) or not 1 <= leverage <= 3:
        raise ValueError("signal leverage is invalid")

    entry_price = payload["entry_price"]
    if entry_price is not None:
        if isinstance(entry_price, bool):
            raise ValueError("signal entry_price is invalid")
        if not isinstance(entry_price, int | float):
            raise ValueError("signal entry_price is invalid")
        entry_price = float(entry_price)

    take_profit_price = payload["take_profit_price"]
    if take_profit_price is not None:
        if isinstance(take_profit_price, bool):
            raise ValueError("signal take_profit_price is invalid")
        if not isinstance(take_profit_price, int | float):
            raise ValueError("signal take_profit_price is invalid")
        take_profit_price = float(take_profit_price)

    stop_loss_price = payload["stop_loss_price"]
    if stop_loss_price is not None:
        if isinstance(stop_loss_price, bool):
            raise ValueError("signal stop_loss_price is invalid")
        if not isinstance(stop_loss_price, int | float):
            raise ValueError("signal stop_loss_price is invalid")
        stop_loss_price = float(stop_loss_price)

    reason = payload["reason"]
    if not isinstance(reason, str) or not reason:
        raise ValueError("signal reason is invalid")

    return TradeSignal(
        action=action,
        confidence=float(confidence),
        leverage=leverage,
        entry_price=entry_price,
        take_profit_price=take_profit_price,
        stop_loss_price=stop_loss_price,
        reason=reason,
    )