Fájl részletek

Előnézet (első 120 sor)

#!/usr/bin/env python3
import csv
import io
import json
import math
import hashlib
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from types import SimpleNamespace
from zoneinfo import ZoneInfo

from flask import Flask, render_template, send_file, jsonify, request, redirect, url_for

import sys
sys.path.insert(0, "/opt/bots/saturnus/app")

from rules_loader import load_rules_doc, load_lab_override_doc
from rules_merge import build_effective_rules
from rule_engine import decide as shared_rule_decide, normalize_state as shared_normalize_state, compute_levels as shared_compute_levels


APP_DIR = Path("/opt/bots/saturnus_monitor/app")
DATA_DIR = Path("/opt/bots/saturnus_monitor/data")

STATE_PATH = Path("/opt/bots/saturnus/state.json")
CONFIG_PATH = Path("/opt/bots/saturnus/freqtrade/user_data/config.json")

PRICES_CSV = DATA_DIR / "prices.csv"
EVENTS_CSV = DATA_DIR / "events.csv"
TRADE_EVENTS_CSV = DATA_DIR / "trade_events.csv"
ANALYSIS_JSON = DATA_DIR / "analysis_summary.json"
ANALYSIS_SCRIPT = Path("/opt/bots/saturnus_monitor/scripts/analyze_monitor_data.py")
FILE_AUDIT_JSON = DATA_DIR / "files_audit_snapshot.json"

app = Flask(__name__, template_folder=str(APP_DIR / "templates"))


# ------------------------------------------------
# alap segédfüggvények
# ------------------------------------------------

def read_json(path: Path):
    if not path.exists():
        return {}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}


def read_csv_rows(path: Path):
    if not path.exists():
        return []
    try:
        with path.open("r", encoding="utf-8", newline="") as f:
            return list(csv.DictReader(f))
    except Exception:
        return []


def to_float(value, default=None):
    try:
        if value is None or value == "":
            return default
        return float(value)
    except Exception:
        return default


def fmt_num(value, digits=4):
    if value is None:
        return "—"
    try:
        return f"{float(value):,.{digits}f}".replace(",", " ")
    except Exception:
        return "—"


def fmt_pct(value, digits=2):
    if value is None:
        return "—"
    try:
        return f"{float(value):.{digits}f}%"
    except Exception:
        return "—"


def safe_tail(rows, n):
    if not rows:
        return []
    return rows[-n:]


def first_nonempty(row, keys, default=""):
    for key in keys:
        val = row.get(key)
        if val not in (None, ""):
            return val
    return default


def parse_bool_arg(name: str, default: bool) -> bool:
    values = request.args.getlist(name)
    if not values:
        return default
    last = values[-1]
    return str(last).strip().lower() in ("1", "true", "yes", "on")


def parse_float_arg(name: str, default: float, min_value=None, max_value=None) -> float:
    value = to_float(request.args.get(name), default)
    if value is None:
        value = default
    if min_value is not None and value < min_value:
        value = min_value
    if max_value is not None and value > max_value:
        value = max_value
    return value

Csak változott diff sorok

Teljes diff