Fájl részletek

Előnézet (első 120 sor)

# rule_engine.py
# Saturnus – kanonikus rule engine
# Fő elvek:
# - normál trade-pár ne legyen veszteséges, ha ezt a piac megengedi
# - panic trade után recovery mód kötelező
# - recovery BUY csak akkor engedhető, ha a következő SELL oldalon
#   a korábbi veszteség visszahozható
# - recovery SELL csak akkor engedhető, ha ténylegesen visszahozza
#   a korábbi veszteséget
# - panic szabályokra kötelező az MA + iránykényszer
# - catastrophe szabályok külön végső védelmi ágként maradnak

from __future__ import annotations

from typing import Any, Dict, List, Optional, Tuple


DEFAULT_CFG = {
    "STD_SELL_RETRACE_PCT": 0.006,
    "STD_BUY_REBOUND_PCT": 0.006,
    "RECOVERY_SELL_RETRACE_PCT": 0.004,
    "RECOVERY_BUY_REBOUND_PCT": 0.012,
    "PANIC_OFFSET_PCT": 0.03,
    "CATASTROPHE_OFFSET_PCT": 0.10,
    "MIN_REBOUND_CONFIRM_PCT": 0.003,
    "RECOVERY_ARM_TIMEOUT_BARS": 240,
    "SPREAD_MAX_PCT": 0.05,
    "STALENESS_MAX_MS": 999999999,
    "PROFIT_BUFFER_PCT": 0.001,
    "COST_GUARD_ENABLED": True,
    "PANIC_RECOVERY_EXTRA_GAP_PCT": 0.002,
    "SIDEWAYS_MA_GAP_PCT": 0.0015,
    "RECOVERY_PROFIT_TARGET_PCT": 0.003,
    "RECOVERY_BUY_MAX_ABOVE_BASE_PCT": 0.0,
}


def _f(value: Any) -> Optional[float]:
    try:
        if value is None:
            return None
        return float(value)
    except Exception:
        return None


def _b(value: Any, default: bool = False) -> bool:
    if isinstance(value, bool):
        return value
    if value is None:
        return default
    if isinstance(value, str):
        v = value.strip().lower()
        if v in ("1", "true", "yes", "on"):
            return True
        if v in ("0", "false", "no", "off"):
            return False
    return bool(value)


def _cfg_value(params: Dict[str, Any], *keys: str, default: Any, cast=float):
    for key in keys:
        if key in params and params.get(key) is not None:
            try:
                return cast(params.get(key))
            except Exception:
                pass
    return default


def _hold(reason: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    payload = {
        "action": "HOLD",
        "rule": "HOLD",
        "reason": reason,
        "trigger_level": None,
        "gate_reason": "",
        "data": data or {},
        "decision_action": "HOLD",
        "decision_rule_id": "HOLD",
        "decision_level": None,
        "decision_reason": reason,
    }
    return payload


def _no_trade(reason: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    return _hold(reason, data or {})


def _sell(rule_id: str, level: float, reason: str, data: Dict[str, Any]) -> Dict[str, Any]:
    payload = {
        "action": "SELL",
        "rule": rule_id,
        "reason": reason,
        "trigger_level": level,
        "gate_reason": "",
        "data": data,
        "decision_action": "SELL",
        "decision_rule_id": rule_id,
        "decision_level": level,
        "decision_reason": reason,
    }
    return payload


def _buy(rule_id: str, level: float, reason: str, data: Dict[str, Any]) -> Dict[str, Any]:
    payload = {
        "action": "BUY",
        "rule": rule_id,
        "reason": reason,
        "trigger_level": level,
        "gate_reason": "",
        "data": data,
        "decision_action": "BUY",
        "decision_rule_id": rule_id,
        "decision_level": level,
        "decision_reason": reason,
    }
    return payload

Csak változott diff sorok

Teljes diff