import sys import unittest from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "tests")) import fake_mt5 sys.modules["MetaTrader5"] = fake_mt5 from sailar import config, data, mt5conn from sailar.broker import Mt5Broker, OrderRejected from sailar.engine import Engine from sailar.journal import Journal from sailar.risk import Risk CFG = config.load(str(ROOT / "config.mt5.json")) PARAMS = {**CFG["strategy"], "min_confidence": 20, "mtf_confirm": False, "trend_exit": False} DF = data.synthetic_candles(1500, start=1.10, seed=3) def build(params=None, pre=None): mt5conn.reset() fake_mt5.reset() if pre: pre() j = Journal(":memory:") p = {**PARAMS, **(params or {})} v = j.new_version(p, 100000) eng = Engine(CFG, Mt5Broker(CFG, v, "demo"), j, Risk(CFG), v, p) return eng, j, v def open_extra(magic, volume=0.01): fake_mt5.order_send({"action": fake_mt5.TRADE_ACTION_DEAL, "symbol": "EURUSD", "volume": volume, "type": fake_mt5.ORDER_TYPE_SELL, "magic": magic, "comment": "x"}) return fake_mt5.S["positions"][-1].ticket def step(eng, i, price=None): win = DF.iloc[i - 199: i + 1] price = float(win.close.iloc[-1]) if price is None else price fake_mt5.set_bid(price) return eng.step(win, win.index[-1], price=price) def run_until_trade(eng, j, v, start=200, stop=1400): for i in range(start, stop): step(eng, i) if j.current_trade(v): return i return None def positions(): return {p.ticket: p for p in fake_mt5.positions_get(symbol="EURUSD")} def warns(j, text): return [e for e in j.events(50, "WARN") if text in e["msg"]] class TwoTrades(unittest.TestCase): def test_the_second_trade_goes_the_managed_one_stays(self): eng, j, v = build() i = run_until_trade(eng, j, v) t = j.current_trade(v) extra = open_extra(eng.b.magic) # a second Sailar-tagged position appears self.assertEqual(len(positions()), 2) step(eng, i + 1) self.assertEqual(set(positions()), {int(t["ticket"])}) # the extra is gone, the managed trade is untouched self.assertEqual(j.current_trade(v)["ticket"], t["ticket"]) self.assertTrue(warns(j, "EXTRA position of Sailar-x")) def test_extra_position_already_there_before_entry_is_removed_first(self): eng, j, v = build() extra = open_extra(eng.b.magic) # leftover from a crash / second copy i = run_until_trade(eng, j, v) self.assertEqual(len(positions()), 1) # never two at once self.assertNotIn(extra, positions()) def test_foreign_position_default_is_warn_once_never_touched(self): eng, j, v = build() foreign = open_extra(0) # opened by hand: magic 0 i = run_until_trade(eng, j, v) for k in range(1, 4): step(eng, i + k) self.assertIn(foreign, positions()) # not touched self.assertEqual(len(warns(j, "NOT opened by Sailar-x")), 1) # warned once, not every loop def test_foreign_position_block_policy(self): eng, j, v = build({"foreign_positions": "block"}) foreign = open_extra(0) self.assertIsNone(run_until_trade(eng, j, v, stop=700)) # no trades while it exists self.assertEqual(j.trades(v), []) fake_mt5.order_send({"action": fake_mt5.TRADE_ACTION_DEAL, "symbol": "EURUSD", "volume": 0.01, "position": foreign, "type": fake_mt5.ORDER_TYPE_BUY, "magic": 0}) # you close it by hand self.assertIsNotNone(run_until_trade(eng, j, v, start=700)) # trading resumes def test_foreign_position_close_policy(self): eng, j, v = build({"foreign_positions": "close"}) foreign = open_extra(0) step(eng, 300) self.assertNotIn(foreign, positions()) self.assertTrue(warns(j, "position NOT opened by Sailar-x")) def test_if_the_extra_cannot_be_closed_no_new_trade_is_opened(self): eng, j, v = build() extra = open_extra(eng.b.magic) fake_mt5.S["reject_deal"] = 10018 # broker refuses orders for i in range(200, 500): step(eng, i) self.assertEqual(j.trades(v), []) self.assertEqual(len(warns(j, "cannot close the extra position")), 1) fake_mt5.S["reject_deal"] = None self.assertIsNotNone(run_until_trade(eng, j, v, start=500)) self.assertNotIn(extra, positions()) def test_wrong_ticket_in_the_journal_is_repaired_not_closed(self): eng, j, v = build() i = run_until_trade(eng, j, v) t = j.current_trade(v) j.db.execute("UPDATE trades SET ticket='999' WHERE id=?", (t["id"],)) j.db.commit() step(eng, i + 1) self.assertIn(int(t["ticket"]), positions()) # the real trade was NOT closed as an "extra" self.assertEqual(j.current_trade(v)["ticket"], t["ticket"]) # and the journal is right again self.assertTrue(warns(j, "repaired the position number")) class RightPositionTargeted(unittest.TestCase): """The bug: with two positions open, the old code closed / protected 'the first one in the list', usually the OLDER one.""" def setUp(self): self.eng, self.j, self.v = build({"one_position_rule": False}) # switch the new rule off to show the targeting itself self.stray = open_extra(self.eng.b.magic) # older, Sailar-tagged self.i = run_until_trade(self.eng, self.j, self.v) self.t = self.j.current_trade(self.v) self.assertEqual(len(positions()), 2) self.assertNotEqual(int(self.t["ticket"]), self.stray) def test_break_even_moves_the_managed_trades_stop_not_the_strays(self): t = self.t sign = 1 if t["side"] == "BUY" else -1 price = t["entry"] + sign * 0.9 * abs(t["entry"] - t["sl"]) msgs = step(self.eng, self.i + 1, price=price) self.assertTrue(any(m.startswith("BE:") for m in msgs), msgs) ps = positions() self.assertEqual(ps[self.stray].sl, 0.0) # the older position's stop was NOT touched self.assertAlmostEqual(ps[int(t["ticket"])].sl, round(t["entry"] + sign * 1.5 * 0.0001, 5), places=5) def test_closing_closes_the_managed_trade_not_the_first_in_the_list(self): self.eng._close(self.t, 1.1, "TIME", self.eng.last_bar or DF.index[300]) ps = positions() self.assertIn(self.stray, ps) # the older one is still open self.assertNotIn(int(self.t["ticket"]), ps) # the managed one was closed def test_broker_refuses_to_guess(self): b = self.eng.b with self.assertRaises(OrderRejected): b.protect("EURUSD", "424242", 1.2, 1.0) # unknown ticket, two positions: no guessing with self.assertRaises(OrderRejected): b.market_order("EURUSD", "buy", 1.0, reduce_only=True, ticket="424242") fake_mt5.S["positions"] = [p for p in fake_mt5.S["positions"] if p.ticket == self.stray] b.protect("EURUSD", "424242", 1.2, 1.0) # exactly one position: it is unambiguous self.assertEqual(positions()[self.stray].sl, 1.2) if __name__ == "__main__": unittest.main(verbosity=2)