Fájl részletek

Előnézet (első 120 sor)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import sys
from pathlib import Path

APP_MONITOR = "/opt/bots/saturnus_monitor/app"
APP_ENGINE = "/opt/bots/saturnus/app"

if APP_MONITOR not in sys.path:
    sys.path.insert(0, APP_MONITOR)
if APP_ENGINE not in sys.path:
    sys.path.insert(0, APP_ENGINE)

from monitor_app import lab_decision_from_shared_engine
from rule_engine import decide as shared_rule_decide


def ok(msg: str) -> None:
    print(f"OK  - {msg}")


def fail(msg: str) -> None:
    print(f"ERR - {msg}")
    raise SystemExit(2)


def assert_equal(actual, expected, msg: str) -> None:
    if actual == expected:
        ok(f"{msg}: {actual}")
    else:
        fail(f"{msg}: actual={actual!r} expected={expected!r}")


def classify_action(action: str | None) -> str:
    a = str(action or "").strip().upper()
    if a.startswith("BUY"):
        return "BUY"
    if a.startswith("SELL"):
        return "SELL"
    return "HOLD"


def run_engine_case(name: str, ctx: dict) -> dict:
    print(f"\n=== ENGINE CASE: {name} ===")
    res = shared_rule_decide(ctx)
    print("ACTION =", res.get("action"))
    print("REASON =", res.get("reason"))
    return res


def run_lab_case(name: str, *, in_position, last, prev_last, base, peak, trough, flags, engine_params, ma_positive, ma_negative, ma_filter_enabled, cycle, ma_fast=None, ma_slow=None) -> dict:
    print(f"\n=== LAB CASE: {name} ===")
    res = lab_decision_from_shared_engine(
        in_position=in_position,
        last=last,
        prev_last=prev_last,
        base=base,
        peak=peak,
        trough=trough,
        flags=flags,
        engine_params=engine_params,
        ma_positive=ma_positive,
        ma_negative=ma_negative,
        ma_filter_enabled=ma_filter_enabled,
        cycle=cycle,
        ma_fast=ma_fast,
        ma_slow=ma_slow,
    )
    print("ACTION =", res.get("action"))
    print("REASON =", res.get("reason"))
    return res


def compare_case(name: str, engine_ctx: dict, lab_kwargs: dict, expected_action: str) -> None:
    engine_res = run_engine_case(name, engine_ctx)
    lab_res = run_lab_case(name, **lab_kwargs)

    assert_equal(engine_res.get("action"), expected_action, f"{name} engine action")
    assert_equal(lab_res.get("action"), expected_action, f"{name} lab action")

    assert_equal(
        classify_action(engine_res.get("action")),
        classify_action(lab_res.get("action")),
        f"{name} engine/lab family parity",
    )


def main() -> int:
    print("SATURNUS ENGINE/LAB PARITY TESTS")

    # -----------------------------------------------------------------
    # CASE 1: BUY_RECOVERY positive
    # -----------------------------------------------------------------
    engine_ctx_1 = {
        "base": 100.0,
        "last": 93.30,
        "prev_last": 93.00,
        "peak": None,
        "trough": 93.0,
        "in_position": False,
        "ma_short": 93.25,
        "ma_long": 93.10,
        "fee_pct_per_side": 0.0005,
        "fee_total_pct": 0.001,
        "fee_buffer_pct": 0.002,
        "std_sell_pct": 0.004,
        "recovery_sell_retrace_pct": 0.002,
        "panic_sell_pct": 0.01,
        "catastrophe_sell_pct": 0.10,
        "std_buy_pct": 0.004,
        "recovery_buy_rebound_pct": 0.003,
        "panic_buy_pct": 0.01,
        "catastrophe_buy_pct": 0.10,
        "sell_reversal_min_pct": 0.0,
        "recovery_context": True,
        "panic_context": "AFTER_SELL_PANIC",
        "last_panic_loss": 0.0,
        "last_panic_sell_price": 97.0,
        "last_panic_buy_price": None,

Csak változott diff sorok

Teljes diff