"""Fix gap intrabar: lo StrategyWorker esce su TP/SL toccati INTRABAR (high/low della barra), al livello, come il backtest — non solo quando il close supera il livello.""" 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}) def _long_worker(tmp): w = StrategyWorker(strategy=load_strategy("MR01_bollinger_fade"), asset="BTC", tf="1h", capital=1000.0, 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_long_exits_at_tp_on_intrabar_high(tmp_path): w = _long_worker(tmp_path) # close=100 (dentro la banda) ma high tocca 102.5 -> con il fix esce a TP w.tick(_df(last_high=102.5, last_low=99.5)) assert not w.in_position def test_long_exits_at_sl_on_intrabar_low(tmp_path): w = _long_worker(tmp_path) w.tick(_df(last_high=100.5, last_low=97.0)) # low sotto SL -> stop assert not w.in_position def test_long_holds_when_bar_within_band(tmp_path): w = _long_worker(tmp_path) w.tick(_df(last_high=101.0, last_low=99.0)) # né TP né SL toccati -> resta in posizione assert w.in_position def test_sl_has_priority_over_tp_same_bar(tmp_path): w = _long_worker(tmp_path) # barra che tocca SIA tp(102) SIA sl(98): conservativo -> SL prima w.tick(_df(last_high=103.0, last_low=97.0)) assert not w.in_position # uscito (allo stop, ramo SL valutato per primo)