Fájl részletek
Ezen az oldalon egy konkrét fájl aktuális állapotát tudod megnézni.
Fájl útvonala
/opt/bots/saturnus/app/rule_engine.py
Aktuális státusz
MODIFIED
Módosítás ideje
1779890409.4975312
Korábbi baseline időpont
1776107087.9101276
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