Fájl részletek

Előnézet (első 120 sor)

from __future__ import annotations

from dataclasses import dataclass, asdict
from typing import Any, Dict, Optional


@dataclass
class Decision:
    action: str
    rule: str
    reason: str
    levels: Dict[str, Optional[float]]
    meta: Dict[str, Any]

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


def _to_float(value: Any, default: float = 0.0) -> float:
    try:
        if value is None:
            return default
        return float(value)
    except (TypeError, ValueError):
        return default


def _to_bool(value: Any, default: bool = False) -> bool:
    if isinstance(value, bool):
        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 build_default_state() -> Dict[str, Any]:
    return {
        "base": 0.0,
        "peak": 0.0,
        "trough": 0.0,
        "last": 0.0,
        "prev_last": 0.0,
        "in_position": False,
        "ma_positive": False,
        "flags": {
            "allow_profitless_buy": False,
            "allow_profitless_sell": False,
        },
        "params": {
            "panic_offset": 0.015,
            "fee": 0.002,
            "catastrophe_offset": 0.10,
            "std_sell_retrace": 0.03,
            "std_buy_rebound": 0.03,
            "profitless_sell_swing": 0.005,
            "profitless_buy_swing": 0.005,
        },
    }


def normalize_state(raw_state: Dict[str, Any]) -> Dict[str, Any]:
    default = build_default_state()

    state = {
        "base": _to_float(raw_state.get("base", default["base"])),
        "peak": _to_float(raw_state.get("peak", default["peak"])),
        "trough": _to_float(raw_state.get("trough", default["trough"])),
        "last": _to_float(raw_state.get("last", default["last"])),
        "prev_last": _to_float(raw_state.get("prev_last", default["prev_last"])),
        "in_position": _to_bool(raw_state.get("in_position", default["in_position"])),
        "ma_positive": _to_bool(raw_state.get("ma_positive", default["ma_positive"])),
        "flags": {},
        "params": {},
    }

    raw_flags = raw_state.get("flags", {}) or {}
    raw_params = raw_state.get("params", {}) or {}

    state["flags"] = {
        "allow_profitless_buy": _to_bool(
            raw_flags.get("allow_profitless_buy", default["flags"]["allow_profitless_buy"])
        ),
        "allow_profitless_sell": _to_bool(
            raw_flags.get("allow_profitless_sell", default["flags"]["allow_profitless_sell"])
        ),
    }

    state["params"] = {
        "panic_offset": _to_float(raw_params.get("panic_offset", default["params"]["panic_offset"])),
        "fee": _to_float(raw_params.get("fee", default["params"]["fee"])),
        "catastrophe_offset": _to_float(raw_params.get("catastrophe_offset", default["params"]["catastrophe_offset"])),
        "std_sell_retrace": _to_float(raw_params.get("std_sell_retrace", default["params"]["std_sell_retrace"])),
        "std_buy_rebound": _to_float(raw_params.get("std_buy_rebound", default["params"]["std_buy_rebound"])),
        "profitless_sell_swing": _to_float(raw_params.get("profitless_sell_swing", default["params"]["profitless_sell_swing"])),
        "profitless_buy_swing": _to_float(raw_params.get("profitless_buy_swing", default["params"]["profitless_buy_swing"])),
    }

    return state


def compute_levels(state: Dict[str, Any]) -> Dict[str, Optional[float]]:
    B = _to_float(state["base"])
    P = _to_float(state["peak"])
    T = _to_float(state["trough"])

    params = state["params"]
    p = _to_float(params["panic_offset"])
    f = _to_float(params["fee"])
    c = _to_float(params["catastrophe_offset"])
    r_std_sell = _to_float(params["std_sell_retrace"])
    r_std_buy = _to_float(params["std_buy_rebound"])

    panic_buy = B * (1 + p)
    panic_sell = B * (1 - p)

Csak változott diff sorok

Teljes diff