Fájl részletek

Előnézet (első 120 sor)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Saturnus tick_runner V2
- fetch last candles via Freqtrade REST API
- sync position from Freqtrade
- compute V2 levels (STANDARD / RECOVERY / PANIC + catastrophe zones)
- persist recovery arming state
- call rule_engine.decide(ctx)
- optional executor wiring
"""

from __future__ import annotations

import os
import json
import time
import base64
from datetime import datetime, timezone
from typing import Any, Tuple
from urllib.request import Request, urlopen


APP_DIR = os.path.dirname(os.path.abspath(__file__))

STATE_PATH = os.getenv("STATE_PATH", "/opt/bots/saturnus/state.json")
FT_URL = os.getenv("FT_URL", "http://127.0.0.1:8089").rstrip("/")
FT_TIMEOUT = float(os.getenv("FT_TIMEOUT", "8"))

TICK_SECONDS = float(os.getenv("TICK_SECONDS", os.getenv("SATURNUS_TICK_SECONDS", "10")))
PAIR = os.getenv("PAIR", "SOL/USDC")
TIMEFRAME = os.getenv("TIMEFRAME", "1m")

# =========================================================
# HTF (Higher Timeframe) debug-only Phase-1
# =========================================================
HTF_TIMEFRAME = os.getenv("HTF_TIMEFRAME", "1h")
HTF_ENABLED = os.getenv("HTF_ENABLED", "1").strip().lower() in ("1", "true", "yes", "on")
HTF_LIMIT = int(os.getenv("HTF_LIMIT", "120"))

LIMIT = int(os.getenv("LIMIT", "2"))
MA_SHORT_PERIOD = int(os.getenv("MA_SHORT_PERIOD", "5"))
MA_LONG_PERIOD = int(os.getenv("MA_LONG_PERIOD", "20"))

FEE_PCT = float(os.getenv("FEE_PCT", "0.1"))
SLIPPAGE_PCT = float(os.getenv("SLIPPAGE_PCT", "0.0"))

FT_USERNAME = os.getenv("FT_USERNAME", "").strip()
FT_PASSWORD = os.getenv("FT_PASSWORD", "").strip()

EXECUTION_ENABLED = os.getenv("EXECUTION_ENABLED", "0") == "1"
EXECUTION_CONFIRM = os.getenv("EXECUTION_CONFIRM", "1") == "1"
EXECUTION_LOG_ONLY = os.getenv("EXECUTION_LOG_ONLY", "1") == "1"

STD_SELL_PCT = float(os.getenv("STD_SELL_PCT", "0.01"))
RECOVERY_SELL_RETRACE_PCT = float(os.getenv("RECOVERY_SELL_RETRACE_PCT", "0.006"))
PANIC_SELL_PCT = float(os.getenv("PANIC_SELL_PCT", "0.01"))
CATASTROPHE_SELL_PCT = float(os.getenv("CATASTROPHE_SELL_PCT", "0.10"))

STD_BUY_PCT = float(os.getenv("STD_BUY_PCT", "0.01"))
RECOVERY_BUY_REBOUND_PCT = float(os.getenv("RECOVERY_BUY_REBOUND_PCT", "0.006"))
PANIC_BUY_PCT = float(os.getenv("PANIC_BUY_PCT", "0.01"))
CATASTROPHE_BUY_PCT = float(os.getenv("CATASTROPHE_BUY_PCT", "0.10"))

RECOVERY_ARM_TIMEOUT_BARS = int(os.getenv("RECOVERY_ARM_TIMEOUT_BARS", "240"))
MIN_REBOUND_CONFIRM_PCT = float(os.getenv("MIN_REBOUND_CONFIRM_PCT", "0.003"))
SPREAD_MAX_PCT = float(os.getenv("SPREAD_MAX_PCT", "0.002"))
STALENESS_MAX_MS = int(os.getenv("STALENESS_MAX_MS", "500"))
PROFIT_BUFFER_PCT = float(os.getenv("PROFIT_BUFFER_PCT", "0.001"))
COST_GUARD_ENABLED = os.getenv("COST_GUARD_ENABLED", "1").strip().lower() in ("1", "true", "yes", "on")


def _runtime_guardrail_flags() -> tuple[bool, int, float]:
    kill_switch = str(os.getenv("KILL_SWITCH", "0")).strip().lower() in ("1", "true", "yes", "on")
    max_trades = int(os.getenv("MAX_TRADES_PER_DAY", "6"))
    daily_loss_cap_pct = float(os.getenv("DAILY_LOSS_CAP_PCT", "2.0"))
    return kill_switch, max_trades, daily_loss_cap_pct


def now_utc_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def log(msg: str) -> None:
    ts = now_utc_iso()
    print(f"{ts} [tick_runner] {msg}", flush=True)


def _env_bool(name: str, default: bool = False) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return bool(default)
    return str(raw).strip().lower() in ("1", "true", "yes", "on")


def _env_int(name: str, default: int) -> int:
    raw = os.getenv(name)
    if raw is None:
        return int(default)
    try:
        return int(str(raw).strip())
    except Exception:
        return int(default)
def _env_float(name: str, default: float = 0.0) -> float:
    try:
        val = os.getenv(name)
        if val is None or str(val).strip() == "":
            return float(default)
        return float(val)
    except Exception:
        return float(default)

def _as_float(x) -> float | None:
    try:
        return float(x) if x is not None else None
    except Exception:
        return None


def _runtime_exec_flags() -> tuple[bool, bool, bool]:

Csak változott diff sorok

Teljes diff