PythagorasGoal/tests/portfolio/test_close_confirm_sl.py

"""EXIT-16 close-confirm SL (2026-06-04): col param `sl_confirm_atr` il wick che
buca lo SL NON stoppa piu' (immune ai wick); lo stop scatta solo se il CLOSE
sfonda sl ∓ buf*ATR14, con uscita al close. TP intrabar invariato.
Senza param il comportamento resta quello storico (regressione)."""
import pandas as pd
from src.live.strategy_worker import StrategyWorker
from src.live.strategy_loader import load_strategy


def _df(last_high, last_low, last_close, n=120, price=100.0):
    c = [price] * n
    h = [price] * n
    l = [price] * n
    h[-1] = last_high
    l[-1] = last_low
    c[-1] = last_close
    ts = (pd.date_range("2024-01-01", periods=n, freq="1h", tz="UTC").astype("int64") // 10**6)
    return pd.DataFrame({"timestamp": ts, "open": [price] * n, "high": h, "low": l,
                         "close": c, "volume": 1.0})


def _long_worker(tmp, confirm=0.5):
    params = {"sl_confirm_atr": confirm} if confirm else {}
    w = StrategyWorker(strategy=load_strategy("MR01_bollinger_fade"), asset="BTC", tf="1h",
                       capital=1000.0, params=params, data_dir=tmp)
    w._notify = lambda *a, **k: None
    w.in_position = True
    w.direction = 1
    w.entry_price = 100.0
    w.tp = 102.0
    w.sl = 98.0
    w.max_bars = 24
    w.bars_held = 1
    w.last_bar_ts = 0
    return w


def test_wick_below_sl_does_not_stop(tmp_path):
    # low 97 buca lo SL(98) ma il CLOSE rientra a 100 -> resta in posizione
    w = _long_worker(tmp_path)
    w.tick(_df(last_high=100.5, last_low=97.0, last_close=100.0))
    assert w.in_position


def test_same_wick_stops_without_param(tmp_path):
    # regressione: senza sl_confirm_atr lo stesso wick stoppa intrabar al livello
    w = _long_worker(tmp_path, confirm=None)
    w.tick(_df(last_high=100.5, last_low=97.0, last_close=100.0))
    assert not w.in_position


def test_close_breach_stops_at_close(tmp_path):
    # close 96.5 sfonda sl-buf (ATR del df flat ~ TR ultimo bar /14 -> buf piccolo) -> stop
    w = _long_worker(tmp_path)
    cap_before = w.capital
    w.tick(_df(last_high=100.5, last_low=96.0, last_close=96.5))
    assert not w.in_position
    assert w.capital < cap_before        # uscito al CLOSE (in perdita), non al livello


def test_tp_intrabar_still_first(tmp_path):
    # high tocca il TP(102) intrabar -> esce al TP anche in modalita' close-confirm
    w = _long_worker(tmp_path)
    w.tick(_df(last_high=102.5, last_low=99.5, last_close=100.0))
    assert not w.in_position
    assert w.capital > 1000.0            # uscito al TP in profitto


def test_small_breach_within_buffer_holds(tmp_path):
    # close appena sotto SL ma DENTRO il buffer ATR -> resta in posizione
    w = _long_worker(tmp_path)
    df = _df(last_high=100.0, last_low=94.0, last_close=97.95)  # TR=6 -> ATR~0.43 -> buf~0.21
    w.tick(df)
    assert w.in_position


# ---------------------------------------------------------------- FIX 2026-06-05
# Il confirm va valutato sul close di barra COMPLETATA (come fade_base.backtest),
# non sul prezzo della barra in formazione: il dip intrabar che rientra NON deve
# stoppare (audit live 2026-06-05: 2 stop su 3 erano wick-stop da barra in corso).

def _df_forming(prev_close, forming_low, forming_close, n=120, price=100.0):
    """Ultima riga = candela IN CORSO (timestamp = ora corrente flooral'ora):
    il worker deve valutare il confirm sulla riga -2 (completata)."""
    c = [price] * n
    h = [price] * n
    l = [price] * n
    c[-2] = prev_close
    l[-1] = forming_low
    c[-1] = forming_close
    h[-1] = max(price, forming_close)
    # ts in MILLISECONDI espliciti (Timestamp.now() ha risoluzione us: l'astype
    # darebbe unita' sbagliate). Ultima barra = ora corrente floor -> in corso.
    end_ms = int(pd.Timestamp.now(tz="UTC").floor("h").timestamp() * 1000)
    ts = [end_ms - (n - 1 - i) * 3_600_000 for i in range(n)]
    return pd.DataFrame({"timestamp": ts, "open": [price] * n, "high": h, "low": l,
                         "close": c, "volume": 1.0})


def test_forming_bar_dip_does_not_stop(tmp_path):
    # barra in corso sprofonda sotto sl-buf (close corrente 96.5) ma la barra
    # COMPLETATA (-2) ha chiuso a 100 -> NIENTE stop (prima del fix stoppava)
    w = _long_worker(tmp_path)
    w.tick(_df_forming(prev_close=100.0, forming_low=96.0, forming_close=96.5))
    assert w.in_position


def test_completed_close_breach_stops_even_if_recovered(tmp_path):
    # la barra COMPLETATA ha chiuso sotto sl-buf; la barra in corso e' rimbalzata
    # sopra -> stop comunque (fill al prezzo corrente, come lag_close_exit)
    w = _long_worker(tmp_path)
    w.tick(_df_forming(prev_close=96.0, forming_low=98.5, forming_close=99.0))
    assert not w.in_position


def test_tp_intrabar_on_forming_bar_unchanged(tmp_path):
    # TP toccato dalla barra in corso -> esce al TP (intrabar invariato)
    w = _long_worker(tmp_path)
    df = _df_forming(prev_close=100.0, forming_low=99.5, forming_close=100.0)
    df.loc[df.index[-1], "high"] = 102.5
    w.tick(df)
    assert not w.in_position
    assert w.capital > 1000.0