Fájl részletek

Előnézet (első 120 sor)

# executor.py
# Saturnus – Executor
# Feladata: RuleDecision -> Freqtrade végrehajtás
# NINCS döntési logika itt!

from typing import Optional, Callable, Any
from datetime import datetime, timezone
import importlib

from rule_engine import RuleDecision


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _resolve_callable(mod: Any, candidates: list[str]) -> Callable:
    """
    Visszaadja az első létező callable attribútumot a modulból.
    """
    for name in candidates:
        fn = getattr(mod, name, None)
        if callable(fn):
            return fn
    raise ImportError(
        f"No suitable callable found in module {getattr(mod, '__name__', mod)} "
        f"candidates={candidates}"
    )


def _call_flex(fn: Callable, pair: str) -> Any:
    """
    Meghívás rugalmasan: először pair=..., ha nem megy, akkor pozicionálisan.
    """
    try:
        return fn(pair=pair)
    except TypeError:
        return fn(pair)


def _load_adapter():
    """
    Preferált: freqtrade_adapter
    Fallback: ft_api
    """
    try:
        m = importlib.import_module("freqtrade_adapter")
        src = "freqtrade_adapter"
        return m, src
    except Exception:
        m = importlib.import_module("ft_api")
        src = "ft_api"
        return m, src


# --- Adapter betöltés + függvények feloldása (névfüggetlen) ---
_ADAPTER_MOD, ADAPTER_SOURCE = _load_adapter()

_FT_BUY = _resolve_callable(
    _ADAPTER_MOD,
    [
        "ft_buy",
        "buy",
        "market_buy",
        "open_buy",
        "open_trade",
        "enter_trade",
        "create_buy_order",
    ],
)

_FT_SELL = _resolve_callable(
    _ADAPTER_MOD,
    [
        "ft_sell",
        "sell",
        "market_sell",
        "close_sell",
        "close_trade",
        "exit_trade",
        "create_sell_order",
    ],
)

_FT_IN_POSITION = _resolve_callable(
    _ADAPTER_MOD,
    [
        "ft_in_position",
        "in_position",
        "has_position",
        "is_in_position",
        "position_open",
        "get_in_position",
    ],
)


class SaturnusExecutor:
    """
    Vékonyréteg:
    - fogad egy RuleDecision-t
    - meghívja a megfelelő Freqtrade akciót
    """

    @staticmethod
    def execute(
        *,
        decision: RuleDecision,
        pair: str,
        price: float,
        state: dict,
        log: Callable[[dict], None],
    ) -> Optional[str]:

        if not decision.is_action():
            return None

        action = decision.action
        reason = decision.reason

Csak változott diff sorok

Teljes diff