Fájl részletek

Ezen az oldalon egy konkrét fájl aktuális állapotát tudod megnézni.

Vissza a fájltérképhez Csak változott Stratégia-labor Monitor főoldal ← Előző módosult Következő módosult →
Fájl útvonala
/opt/bots/saturnus/app/rule_engine.py
Létezik most?
IGEN
Aktuális státusz
MODIFIED
Méret
43152
Módosítás ideje
1779890409.4975312
Korábbi baseline időpont
1776107087.9101276
SHA256 rövid

Előnézet (első 120 sor)

from __future__ import annotations

from collections import Counter, deque

from dataclasses import dataclass, asdict
from typing import Any, Dict, List, Optional, Tuple
import math


# ============================================================
# Saturnus v2.1 canonical rule engine
# ------------------------------------------------------------
# Main goals:
# - STANDARD / RECOVERY / PANIC / CATASTROPHE decision model
# - Trend-aware PANIC rules
# - Profit invariant for STANDARD and RECOVERY exits
# - Panic context tracking
# - Safe defaults and robust missing-key handling
# - Multiple entry-point aliases for easier drop-in integration
# ============================================================


# -----------------------------
# Helpers
# -----------------------------
def _f(value: Any, default: float = 0.0) -> float:
    try:
        if value is None:
            return float(default)
        return float(value)
    except Exception:
        return float(default)


def _i(value: Any, default: int = 0) -> int:
    try:
        if value is None:
            return int(default)
        return int(value)
    except Exception:
        return int(default)


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


def _s(value: Any, default: str = "") -> str:
    if value is None:
        return default
    return str(value)


def _clean_float(value: float) -> Optional[float]:
    if value is None:
        return None
    if isinstance(value, float) and (math.isnan(value) or math.isinf(value)):
        return None
    return float(value)


def _pct(value: Any, default: float = 0.0) -> float:
    """
    Accepts both:
    - 0.03  => 3%
    - 3     => 3%
    Converts everything to decimal form.
    """
    v = _f(value, default)
    if abs(v) >= 1.0:
        return v / 100.0
    return v


def _first_present(d: Dict[str, Any], keys: List[str], default: Any = None) -> Any:
    for k in keys:
        if k in d and d[k] is not None:
            return d[k]
    return default


def _max_valid(values: List[Optional[float]]) -> Optional[float]:
    vals = [v for v in values if v is not None]
    return max(vals) if vals else None


def _min_valid(values: List[Optional[float]]) -> Optional[float]:
    vals = [v for v in values if v is not None]
    return min(vals) if vals else None


# -----------------------------
# Data models
# -----------------------------
@dataclass
class EngineContext:
    base: float
    last: float
    prev_last: float
    peak: float
    trough: float
    in_position: bool

    ma_short: float
    ma_long: float
    ma_sideways_band_pct: float

    fee_pct_per_side: float
    fee_total_pct: float

Csak változott diff sorok

--- baseline +++ current @@ -1,60 +1,44 @@ -# rule_engine.py -# Saturnus v2 – recovery-alapú, 6 szabályos döntési motor -# Kompatibilis wrapperrel a meglévő egyparaméteres decide(raw_state) hívásokhoz - -from typing import Any, Dict, Optional, Tuple - - -# ========================================================= -# Default config -# ========================================================= - -DEFAULT_CFG: Dict[str, Any] = { - "STD_SELL_RETRACE_PCT": 0.010, - "STD_BUY_REBOUND_PCT": 0.010, - "RECOVERY_SELL_RETRACE_PCT": 0.006, - "RECOVERY_BUY_REBOUND_PCT": 0.006, - "PANIC_OFFSET_PCT": 0.020, - "CATASTROPHE_OFFSET_PCT": 0.120, - "MIN_REBOUND_CONFIRM_PCT": 0.002, - "RECOVERY_ARM_TIMEOUT_BARS": 240, - "SPREAD_MAX_PCT": 0.0020, - "STALENESS_MAX_MS": 500, - "PROFIT_BUFFER_PCT": 0.0010, - "COST_GUARD_ENABLED": True, - "PANIC_RECOVERY_EXTRA_GAP_PCT": 0.002, - "SIDEWAYS_MA_GAP_PCT": 0.0015, - "RECOVERY_PROFIT_TARGET_PCT": 0.001, - "RECOVERY_BUY_MAX_ABOVE_BASE_PCT": 0.006, -} - - -SELL_PRIORITY = { - "SELL_PANIC": 1, - "SELL_STANDARD": 2, - "SELL_RECOVERY": 3, -} - -BUY_PRIORITY = { - "BUY_PANIC": 1, - "BUY_STANDARD": 2, - "BUY_RECOVERY": 3, -} - - -# ========================================================= +from collections import Counter, deque + +from dataclasses import dataclass, asdict +from typing import Any, Dict, List, Optional, Tuple +import math + + +# ============================================================ +# Saturnus v2.1 canonical rule engine +# ------------------------------------------------------------ +# Main goals: +# - STANDARD / RECOVERY / PANIC / CATASTROPHE decision model +# - Trend-aware PANIC rules +# - Profit invariant for STANDARD and RECOVERY exits +# - Panic context tracking +# - Safe defaults and robust missing-key handling +# - Multiple entry-point aliases for easier drop-in integration +# ============================================================ + + +# ----------------------------- -# ========================================================= - -def _f(value: Any) -> Optional[float]: +# ----------------------------- +def _f(value: Any, default: float = 0.0) -> float: - return None + return float(default) - return None + return float(default) + + +def _i(value: Any, default: int = 0) -> int: + try: + if value is None: + return int(default) + return int(value) + except Exception: + return int(default) @@ -62,592 +46,1122 @@ - 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 _decision( - *, - action: str, - rule: str, - reason: str, - trigger_level: Optional[float] = None, - gate_reason: str = "", - data: Optional[Dict[str, Any]] = None, + if isinstance(value, (int, float)): + return bool(value) + s = str(value).strip().lower() + if s in ("1", "true", "yes", "y", "on"): + return True + if s in ("0", "false", "no", "n", "off", ""): + return False + return default + + +def _s(value: Any, default: str = "") -> str: + if value is None: + return default + return str(value) + + +def _clean_float(value: float) -> Optional[float]: + if value is None: + return None + if isinstance(value, float) and (math.isnan(value) or math.isinf(value)): + return None + return float(value) + + +def _pct(value: Any, default: float = 0.0) -> float: + """ + Accepts both: + - 0.03 => 3% + - 3 => 3% + Converts everything to decimal form. + """ + v = _f(value, default) + if abs(v) >= 1.0: + return v / 100.0 + return v + + +def _first_present(d: Dict[str, Any], keys: List[str], default: Any = None) -> Any: + for k in keys: + if k in d and d[k] is not None: + return d[k] + return default + + +def _max_valid(values: List[Optional[float]]) -> Optional[float]: + vals = [v for v in values if v is not None] + return max(vals) if vals else None + + +def _min_valid(values: List[Optional[float]]) -> Optional[float]: + vals = [v for v in values if v is not None] + return min(vals) if vals else None + + +# ----------------------------- +# Data models +# ----------------------------- +@dataclass +class EngineContext: + base: float + last: float + prev_last: float + peak: float + trough: float + in_position: bool + + ma_short: float + ma_long: float + ma_sideways_band_pct: float + + fee_pct_per_side: float + fee_total_pct: float + fee_buffer_pct: float + + std_sell_pct: float + recovery_sell_retrace_pct: float + panic_sell_pct: float + catastrophe_sell_pct: float + + std_buy_pct: float + recovery_buy_rebound_pct: float + panic_buy_pct: float

Teljes diff

--- baseline +++ current @@ -1,60 +1,44 @@ -# rule_engine.py -# Saturnus v2 – recovery-alapú, 6 szabályos döntési motor -# Kompatibilis wrapperrel a meglévő egyparaméteres decide(raw_state) hívásokhoz - from __future__ import annotations -from typing import Any, Dict, Optional, Tuple - - -# ========================================================= -# Default config -# ========================================================= - -DEFAULT_CFG: Dict[str, Any] = { - "STD_SELL_RETRACE_PCT": 0.010, - "STD_BUY_REBOUND_PCT": 0.010, - "RECOVERY_SELL_RETRACE_PCT": 0.006, - "RECOVERY_BUY_REBOUND_PCT": 0.006, - "PANIC_OFFSET_PCT": 0.020, - "CATASTROPHE_OFFSET_PCT": 0.120, - "MIN_REBOUND_CONFIRM_PCT": 0.002, - "RECOVERY_ARM_TIMEOUT_BARS": 240, - "SPREAD_MAX_PCT": 0.0020, - "STALENESS_MAX_MS": 500, - "PROFIT_BUFFER_PCT": 0.0010, - "COST_GUARD_ENABLED": True, - "PANIC_RECOVERY_EXTRA_GAP_PCT": 0.002, - "SIDEWAYS_MA_GAP_PCT": 0.0015, - "RECOVERY_PROFIT_TARGET_PCT": 0.001, - "RECOVERY_BUY_MAX_ABOVE_BASE_PCT": 0.006, -} - - -SELL_PRIORITY = { - "SELL_PANIC": 1, - "SELL_STANDARD": 2, - "SELL_RECOVERY": 3, -} - -BUY_PRIORITY = { - "BUY_PANIC": 1, - "BUY_STANDARD": 2, - "BUY_RECOVERY": 3, -} - - -# ========================================================= +from collections import Counter, deque + +from dataclasses import dataclass, asdict +from typing import Any, Dict, List, Optional, Tuple +import math + + +# ============================================================ +# Saturnus v2.1 canonical rule engine +# ------------------------------------------------------------ +# Main goals: +# - STANDARD / RECOVERY / PANIC / CATASTROPHE decision model +# - Trend-aware PANIC rules +# - Profit invariant for STANDARD and RECOVERY exits +# - Panic context tracking +# - Safe defaults and robust missing-key handling +# - Multiple entry-point aliases for easier drop-in integration +# ============================================================ + + +# ----------------------------- # Helpers -# ========================================================= - -def _f(value: Any) -> Optional[float]: +# ----------------------------- +def _f(value: Any, default: float = 0.0) -> float: try: if value is None: - return None + return float(default) return float(value) except Exception: - return None + return float(default) + + +def _i(value: Any, default: int = 0) -> int: + try: + if value is None: + return int(default) + return int(value) + except Exception: + return int(default) def _b(value: Any, default: bool = False) -> bool: @@ -62,592 +46,1122 @@ 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 _decision( - *, - action: str, - rule: str, - reason: str, - trigger_level: Optional[float] = None, - gate_reason: str = "", - data: Optional[Dict[str, Any]] = None, + if isinstance(value, (int, float)): + return bool(value) + s = str(value).strip().lower() + if s in ("1", "true", "yes", "y", "on"): + return True + if s in ("0", "false", "no", "n", "off", ""): + return False + return default + + +def _s(value: Any, default: str = "") -> str: + if value is None: + return default + return str(value) + + +def _clean_float(value: float) -> Optional[float]: + if value is None: + return None + if isinstance(value, float) and (math.isnan(value) or math.isinf(value)): + return None + return float(value) + + +def _pct(value: Any, default: float = 0.0) -> float: + """ + Accepts both: + - 0.03 => 3% + - 3 => 3% + Converts everything to decimal form. + """ + v = _f(value, default) + if abs(v) >= 1.0: + return v / 100.0 + return v + + +def _first_present(d: Dict[str, Any], keys: List[str], default: Any = None) -> Any: + for k in keys: + if k in d and d[k] is not None: + return d[k] + return default + + +def _max_valid(values: List[Optional[float]]) -> Optional[float]: + vals = [v for v in values if v is not None] + return max(vals) if vals else None + + +def _min_valid(values: List[Optional[float]]) -> Optional[float]: + vals = [v for v in values if v is not None] + return min(vals) if vals else None + + +# ----------------------------- +# Data models +# ----------------------------- +@dataclass +class EngineContext: + base: float + last: float + prev_last: float + peak: float + trough: float + in_position: bool + + ma_short: float + ma_long: float + ma_sideways_band_pct: float + + fee_pct_per_side: float + fee_total_pct: float + fee_buffer_pct: float + + std_sell_pct: float + recovery_sell_retrace_pct: float + panic_sell_pct: float + catastrophe_sell_pct: float + + std_buy_pct: float + recovery_buy_rebound_pct: float + panic_buy_pct: float ... [DIFF LEVÁGVA] további sorok: 1600