Fájl részletek

Előnézet (első 120 sor)

#!/usr/bin/env python3
"""Saturnus - Freqtrade state.json sync (canonical writer)

Goals (2026-01):
- Write ONLY canonical keys used by the UI / rule-engine.
- Preserve any existing `legacy` block (do not rewrite it every run).
- Never re-introduce legacy top-level keys like:
  last_price, current_rate, open_rate, base_price_current, in_position, etc.

This script is executed by systemd timer: saturnus-state-sync.timer
"""

import json
import os
import time
import urllib.parse
import urllib.request
from base64 import b64encode
from typing import Any, Dict, Optional, Tuple


# =============================================================================
# CONFIG (paths + defaults)
# =============================================================================
FT_CONFIG = os.getenv(
    "SATURNUS_FT_CONFIG",
    "/opt/bots/saturnus/freqtrade/user_data/config.json",
)
STATE_JSON = os.getenv(
    "SATURNUS_STATE_JSON",
    "/opt/bots/saturnus/state.json",
)

STOPLOSS_COOLDOWN_SECONDS = int(os.getenv("SATURNUS_STOPLOSS_COOLDOWN_SECONDS", "300"))  # 5 perc

DEFAULT_FT_BASE_URL = os.getenv("SATURNUS_FT_API_BASE", "http://127.0.0.1:8089").strip()
DEFAULT_FT_USER = os.getenv("SATURNUS_FT_API_USER", "").strip()
DEFAULT_FT_PASS = os.getenv("SATURNUS_FT_API_PASS", "").strip()


# =============================================================================
# TIME + JSON IO
# =============================================================================
def now_utc_iso() -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


def now_epoch() -> int:
    return int(time.time())


def iso_from_epoch(ts: int) -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ts))


def load_json(path: str, default: Any = None) -> Any:
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return default


def atomic_write_json(path: str, data: Dict[str, Any]) -> None:
    tmp = f"{path}.tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2, sort_keys=True)
        f.write("\n")
    os.replace(tmp, path)


# =============================================================================
# HTTP (Basic auth)
# =============================================================================
def http_json(url: str, username: Optional[str], password: Optional[str], timeout: int = 6) -> Any:
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    if username and password:
        token = b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii")
        req.add_header("Authorization", f"Basic {token}")
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        raw = resp.read().decode("utf-8", errors="replace")
    return json.loads(raw)


# =============================================================================
# Helpers
# =============================================================================
def normalize_pair(p: Any) -> Any:
    # state-ben lehet SOL/USDC vagy SOLUSDC; egységesítünk "SOL/USDC"-re
    if not p:
        return p
    p = str(p).strip()
    if "/" in p:
        return p
    if p.endswith("USDC") and len(p) > 4:
        return p[:-4] + "/USDC"
    if p.endswith("USDT") and len(p) > 4:
        return p[:-4] + "/USDT"
    return p


def normalize_base_url(u: str) -> str:
    # ha valahonnan ".../api/v1" jön, levágjuk, mert mi hozzáadjuk az endpointoknál
    u = (u or "").strip()
    if not u:
        return DEFAULT_FT_BASE_URL or "http://127.0.0.1:8089"
    u = u.rstrip("/")
    if u.endswith("/api/v1"):
        u = u[:-7]
    return u.rstrip("/")


def pick_pair(state: Dict[str, Any], ft_cfg: Dict[str, Any], status_item: Optional[Dict[str, Any]]) -> str:
    # 0) status pair
    if isinstance(status_item, dict) and status_item.get("pair"):
        return normalize_pair(status_item["pair"]) or "SOL/USDC"

    # 1) state.json pair
    if isinstance(state, dict) and state.get("pair"):
        return normalize_pair(state["pair"]) or "SOL/USDC"

Csak változott diff sorok

Teljes diff