PythagorasGoal/Old/tests/portfolio/test_tp_phantom.py

"""TP_PHANTOM (2026-06-11): il tocco TP che esiste solo nel feed (spike-print testnet)
NON deve chiudere — il LIMIT resting sul book reale e' l'oracolo: zero fill + prezzo
lontano dal livello = wick fantasma -> exit soppressa. Con fill reale (anche parziale),
o prezzo realmente oltre il livello, o worker senza esecuzione -> comportamento storico."""
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, n=120, price=100.0):
    c = [price] * n
    h = [price] * n
    l = [price] * n
    h[-1] = last_high
    l[-1] = last_low
    ts = (pd.date_range("2024-01-01", periods=n, freq="1h", tz="UTC").astype("int64") // 10**6)
    return pd.DataFrame({"timestamp": ts, "open": c, "high": h, "low": l, "close": c, "volume": 1.0})


class _FakeExecutor:
    """Solo cio' che serve al gate + alla _real_close di un eventuale exit."""
    def __init__(self, tp_filled_amount=0.0):
        self.tp_filled_amount = tp_filled_amount
        self.verify_polls = 1
        self.verify_sleep = 0.0

    def resting_fills(self, instrument, order_id):
        return self.tp_filled_amount, 102.0 if self.tp_filled_amount else None, 0.0

    def cancel_order(self, order_id):
        return {"state": "cancelled"}

    def close_amount(self, instrument, side, amount, label=""):
        class F:
            verified = True
            amount = 0.01
            filled_amount = 0.01
            fill_price = 100.0
            fee_usd = 0.01
            order_id = "X"
        return F()


def _long_worker(tmp, executor=None):
    w = StrategyWorker(strategy=load_strategy("MR01_bollinger_fade"), asset="BTC", tf="1h",
                       capital=1000.0, data_dir=tmp,
                       executor=executor,
                       exec_instrument="BTC_USDC-PERPETUAL" if executor else None)
    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
    if executor:
        w.real_in_position = True
        w.real_side = "buy"
        w.real_amount = 0.01
        w.real_entry_price = 100.0
        w.real_entry_notional = 1.0
        w.real_tp_order_id = "TP-1"
    return w


def test_phantom_touch_suppressed(tmp_path):
    # wick a 102.5 nel feed, resting zero-fill, close 100 lontano dal TP -> NON chiude
    w = _long_worker(tmp_path, executor=_FakeExecutor(tp_filled_amount=0.0))
    w.tick(_df(last_high=102.5, last_low=99.5))
    assert w.in_position
    assert w.real_in_position


def test_real_fill_closes(tmp_path):
    # stesso wick ma il resting HA fillato -> tocco reale -> chiude
    w = _long_worker(tmp_path, executor=_FakeExecutor(tp_filled_amount=0.01))
    w.tick(_df(last_high=102.5, last_low=99.5))
    assert not w.in_position


def test_price_beyond_tp_closes_without_fill(tmp_path):
    # close corrente OLTRE il TP (gap fra poll): tocco genuino anche a zero fill
    w = _long_worker(tmp_path, executor=_FakeExecutor(tp_filled_amount=0.0))
    df = _df(last_high=103.0, last_low=99.5)
    df.loc[df.index[-1], "close"] = 102.6
    w.tick(df)
    assert not w.in_position


def test_no_executor_keeps_legacy_behavior(tmp_path):
    # worker senza esecuzione reale: il gate non si applica (parita' storica)
    w = _long_worker(tmp_path, executor=None)
    w.tick(_df(last_high=102.5, last_low=99.5))
    assert not w.in_position


def test_phantom_does_not_block_other_exits(tmp_path):
    # tocco TP fantasma E max_bars raggiunto nello stesso tick -> esce a time_limit
    w = _long_worker(tmp_path, executor=_FakeExecutor(tp_filled_amount=0.0))
    w.bars_held = 24
    w.tick(_df(last_high=102.5, last_low=99.5))
    assert not w.in_position