"""REAL-TRUTH (2026-06-10): col flag attivo il ledger `capital` dei worker eseguiti si aggiorna col PnL dei FILL REALI (fee reali incluse); il sim resta diagnostica nel log CLOSE (pnl_source/sim_pnl/real_pnl). Fallback dichiarato al sim solo se il trade reale non e' mai esistito/fillato. Executor finti, nessuna rete.""" import json from types import SimpleNamespace from src.live.execution import Fill, PairFill from src.live.strategy_worker import StrategyWorker from src.live.pairs_worker import PairsWorker # ---------------- helpers ---------------- class FakeExec: """Single-leg: open filla a `open_px`, close a `close_px` (fee fisse).""" verify_polls = 1 verify_sleep = 0.0 disaster_sl_pct = None def __init__(self, open_px=100.0, close_px=100.0, close_verified=True): self.open_px = open_px self.close_px = close_px self.close_verified = close_verified def open(self, instrument, side, notional, label=None): amt = round(notional / self.open_px, 6) return Fill(instrument, side, notional, amt, self.open_px, 0.0, 0.05, "oid-open", "filled", True) def place_tp_limit(self, *a, **k): return Fill("x", "sell", 0.0, 0.0, None, 0.0, 0.0, None, None, False) def place_disaster_sl(self, *a, **k): return Fill("x", "sell", 0.0, 0.0, None, 0.0, 0.0, None, None, False) def cancel_order(self, oid): return {"state": "cancelled"} def resting_fills(self, instrument, oid): return 0.0, None, 0.0 def close_amount(self, instrument, side, amount, label=None): return Fill(instrument, "sell" if side == "buy" else "buy", 0.0, amount, self.close_px, 0.0, 0.05, "oid-close", "filled", self.close_verified) def _sw(tmp_path, fake, real_truth=True): w = StrategyWorker(strategy=SimpleNamespace(name="FAKE", fee_rt=0.001), asset="BTC", tf="1h", capital=100.0, position_size=0.5, leverage=2.0, data_dir=tmp_path, executor=fake, exec_instrument="BTC_USDC-PERPETUAL", real_truth=real_truth) w._notify = lambda *a, **k: None return w def _last_close(w): rows = [json.loads(l) for l in w.trades_path.read_text().strip().splitlines()] return [r for r in rows if r.get("event") == "CLOSE"][-1] def _open_sim(w, price=100.0): from src.strategies.base import Signal w._open_position(Signal(idx=0, direction=1, entry_price=price, metadata={"max_bars": 12}), price) # ---------------- StrategyWorker ---------------- def test_capital_updates_from_real_fills(tmp_path): """Fill reali 100->105: capital += real_pnl (gross sul notional reale - fee reali), NON il sim pnl.""" fake = FakeExec(open_px=100.0, close_px=105.0) w = _sw(tmp_path, fake) _open_sim(w, 100.0) assert w.real_in_position notional = w.real_entry_notional # = capital*ps*lev = 100 w._close_position(105.0, "time_limit") real_pnl = 0.05 * notional - 0.10 # +5% su notional - fee 2x0.05 assert abs(w.capital - (100.0 + real_pnl)) < 1e-6 c = _last_close(w) assert c["pnl_source"] == "real" assert abs(c["real_pnl"] - real_pnl) < 1e-3 assert c["win"] is True and w.total_wins == 1 def test_real_loss_counts_as_loss_even_if_sim_wins(tmp_path): """Divergenza sim/reale: il sim vede +5% (exit sim 105) ma il fill reale e' 99 -> in real-truth il trade e' LOSS e il capitale scende.""" fake = FakeExec(open_px=100.0, close_px=99.0) w = _sw(tmp_path, fake) _open_sim(w, 100.0) w._close_position(105.0, "take_profit") # sim esce a 105 (win sim) c = _last_close(w) assert c["pnl_source"] == "real" assert c["real_pnl"] < 0 and c["win"] is False assert w.capital < 100.0 assert c["sim_pnl"] > 0 # il sim avrebbe bookato un win def test_fallback_sim_when_real_open_failed(tmp_path): """REAL_OPEN_FAIL (nessuna posizione reale): il ledger usa il sim, con flag.""" class FailOpen(FakeExec): def open(self, instrument, side, notional, label=None): return Fill(instrument, side, notional, 0.0, None, 0.0, 0.0, None, "error", False, notes="boom") w = _sw(tmp_path, FailOpen()) _open_sim(w, 100.0) assert not w.real_in_position w._close_position(105.0, "time_limit") c = _last_close(w) assert c["pnl_source"] == "sim_fallback" assert c["pnl"] == c["sim_pnl"] assert w.capital > 100.0 # sim pnl applicato def test_shadow_mode_unchanged_without_flag(tmp_path): """real_truth=False (default storico): capital segue il SIM, real_capital il reale.""" fake = FakeExec(open_px=100.0, close_px=99.0) w = _sw(tmp_path, fake, real_truth=False) _open_sim(w, 100.0) w._close_position(105.0, "take_profit") c = _last_close(w) assert "pnl_source" not in c assert w.capital > 100.0 # sim win applicato al ledger assert w.real_capital < 100.0 # reale in perdita, separato # ---------------- PairsWorker ---------------- class FakePairsExec: def __init__(self, open_a=100.0, open_b=50.0, close_a=102.0, close_b=50.0): self.px = dict(open_a=open_a, open_b=open_b, close_a=close_a, close_b=close_b) def _fill(self, inst, side, amount, px): return Fill(inst, side, 0.0, amount, px, 0.0, 0.02, "oid", "filled", True) def open_pair(self, inst_a, inst_b, direction, notional, label=None): sa = "buy" if direction == 1 else "sell" sb = "sell" if direction == 1 else "buy" return PairFill(True, self._fill(inst_a, sa, notional / self.px["open_a"], self.px["open_a"]), self._fill(inst_b, sb, notional / self.px["open_b"], self.px["open_b"])) def close_pair(self, inst_a, inst_b, side_a, side_b, amount_a, amount_b, label=None): return PairFill(True, self._fill(inst_a, "sell", amount_a, self.px["close_a"]), self._fill(inst_b, "buy", amount_b, self.px["close_b"])) def _pw(tmp_path, fake, real_truth=True): w = PairsWorker(asset_a="ETH", asset_b="BTC", tf="1h", capital=100.0, position_size=0.2, leverage=2.0, data_dir=tmp_path, executor=fake, exec_instruments={"ETH": "ETH_USDC-PERPETUAL", "BTC": "BTC_USDC-PERPETUAL"}, real_truth=real_truth) w._notify = lambda *a, **k: None return w def test_pairs_capital_updates_from_real_fills(tmp_path): """Long ratio, gamba A +2% reale: capital += real_pnl (2 gambe, 4 fee da 0.02).""" fake = FakePairsExec(open_a=100.0, close_a=102.0) w = _pw(tmp_path, fake) w._open(1, 100.0, 50.0, -2.1) assert w.real_in_position na = w.real_notional_a w._close(102.0, 50.0, 0.1, "mean_revert") real_pnl = 0.02 * na - 4 * 0.02 # +2% gamba A - fee 4 lati assert abs(w.capital - (100.0 + real_pnl)) < 1e-6 c = json.loads([l for l in w.trades_path.read_text().splitlines() if '"CLOSE"' in l][-1]) assert c["pnl_source"] == "real" assert w.total_trades == 1 def test_pairs_fallback_sim_on_leg_fail(tmp_path): """Apertura reale fallita (leg-risk unwound): chiusura su ledger sim con flag.""" class FailOpen(FakePairsExec): def open_pair(self, *a, **k): bad = Fill("x", "buy", 0.0, 0.0, None, 0.0, 0.0, None, "error", False) return PairFill(False, bad, bad, unwound=False, notes="leg-fail") w = _pw(tmp_path, FailOpen()) w._open(1, 100.0, 50.0, -2.1) assert not w.real_in_position w._close(102.0, 50.0, 0.1, "mean_revert") c = json.loads([l for l in w.trades_path.read_text().splitlines() if '"CLOSE"' in l][-1]) assert c["pnl_source"] == "sim_fallback" assert c["pnl"] == c["sim_pnl"]