import json import sys import unittest from pathlib import Path from types import SimpleNamespace as NS 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 import pandas as pd from sailar import config, data, mt5conn from sailar.brain import llm from sailar.brain.brain import Brain from sailar.brain.context import build_context from sailar.brain.watch import clean from sailar.broker import Mt5Broker from sailar.engine import Engine from sailar.guard import Guard, is_forex from sailar.journal import Journal from sailar.risk import Risk from sailar.signals import analyze_mtf from test_brain import FakeNet 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) NOW = pd.Timestamp("2026-10-23 12:00", tz="UTC") class Crypto(unittest.TestCase): def test_which_symbols_are_forex(self): for s in ("EURUSD", "EUR/USD", "USDJPY", "eurusd", "EURUSD.m", "GBPJPY"): self.assertTrue(is_forex(s), s) for s in ("BTCUSD", "BTC/USDT", "ETHUSD", "XAUUSD", "SOL/USDT"): self.assertFalse(is_forex(s), s) def stub(self, spread_pips, symbol="BTCUSD", pip=0.01): j = Journal(":memory:") v = j.new_version(PARAMS, 100000) return NS(j=j, v=v, symbol=symbol, b=NS(spread_pips=lambda s: spread_pips, pip_size=lambda s: pip)) def test_forex_pip_limit_does_not_block_every_crypto_trade(self): # BTC at 70,000, spread $15 = "1500 pips" with pip=0.01. The forex rule (2 pips) would refuse this forever. e = self.stub(1500) ok, why = Guard(CFG, PARAMS).entry_check(e, NS(price=70000.0), 300.0, NOW) self.assertTrue(ok, why) def test_crypto_spread_is_judged_as_percent_of_price(self): g = Guard(CFG, PARAMS) ok, why = g.entry_check(self.stub(20000), NS(price=70000.0), 3000.0, NOW) # $200 spread = 0.29% self.assertFalse(ok) self.assertIn("% of the price", why) def test_stop_vs_spread_ratio_still_protects_crypto(self): ok, why = Guard(CFG, PARAMS).entry_check(self.stub(1500), NS(price=70000.0), 50.0, NOW) # $50 stop, $15 spread self.assertFalse(ok) self.assertIn("too tight", why) def test_forex_rules_unchanged(self): ok, why = Guard(CFG, PARAMS).entry_check(self.stub(3.0, "EURUSD", 0.0001), NS(price=1.1), 0.0020, NOW) self.assertFalse(ok) self.assertIn("pip limit", why) def test_ai_context_speaks_in_percent_for_crypto(self): df = data.synthetic_candles(600, start=70000.0, seed=2) p = {**PARAMS} sig = analyze_mtf(df.iloc[-500:], p) j = Journal(":memory:") v = j.new_version(p, 100000) eng = NS(symbol="BTC/USDT", cfg={"primary_timeframe": "15m"}, p=p, j=j, v=v, day_start=100000, own_equity=lambda: 100000.0, drawdown_pct=lambda: 0.0, b=NS(pip_size=lambda s: 0.01, spread_pips=lambda s: 1500)) ctx = build_context(eng, sig, "buy", sig.atr * 2, 0.1, 10.0, 100000, df, NOW) self.assertIn("crypto", ctx["instrument_type"]) self.assertIn("stop_pct_of_price", ctx["proposal"]) self.assertIn("atr_pct_of_price", ctx["market"]) self.assertAlmostEqual(ctx["market"]["spread_pct_of_price"], 1500 * 0.01 / sig.price * 100, places=3) def verdict(decision="APPROVE"): return json.dumps({"decision": decision, "confidence": 70, "size_multiplier": 1.0, "event_risk": "none", "reasons": [], "risks": [], "summary": ""}) def watch_reply(action="HOLD", to="", conf=80, reasons=("test",)): return json.dumps({"action": action, "tighten_to": to, "confidence": conf, "reasons": list(reasons), "summary": "s"}) def setup(watch_mode="active", reply=None, mode="active", params=None): mt5conn.reset() fake_mt5.reset() net = FakeNet() net.reply = reply if reply is not None else watch_reply() net.groq = lambda body: net.reply if "protecting ONE open" in body["messages"][0]["content"] else verdict() llm._http = net cfg = json.loads(json.dumps(CFG)) cfg["brain"].update(mode=mode, watch=watch_mode, max_calls_per_day=100000) 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) eng.brain = Brain(cfg, j, {"GROQ_API_KEY": "k"}) return eng, j, v, net 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 open_trade(eng, j, v): for i in range(200, 1400): step(eng, i) if j.current_trade(v): return i raise AssertionError("no trade") def at_r(t, r): sign = 1 if t["side"] == "BUY" else -1 R = abs(t["tp"] - t["entry"]) / 2.0 # rr = 2 return t["entry"] + sign * r * R def watch_logs(j): return j.brain_logs("watch", 50) class Watch(unittest.TestCase): def test_validation_blocks_nonsense(self): c = clean({"action": "BUY MORE", "tighten_to": "moon", "confidence": "high"}) self.assertEqual((c["action"], c["tighten_to"], c["confidence"]), ("HOLD", "", 0)) self.assertEqual(clean(None)["action"], "HOLD") def test_shadow_records_but_changes_nothing(self): eng, j, v, net = setup("shadow", watch_reply("TIGHTEN", "breakeven", 90)) i = open_trade(eng, j, v) t = j.current_trade(v) step(eng, i + 1, at_r(t, 0.5)) self.assertEqual(j.current_trade(v)["sl"], t["sl"]) self.assertEqual(fake_mt5.positions_get(ticket=int(t["ticket"]))[0].sl, round(t["sl"], 5)) log = watch_logs(j)[0] self.assertIn("SHADOW: would tighten to breakeven", log["applied"]) self.assertTrue(any("AI watch (shadow)" in e["msg"] for e in j.events(20, "INFO"))) def test_off_never_calls_the_ai(self): eng, j, v, net = setup("off", watch_reply("EXIT", "", 99)) i = open_trade(eng, j, v) calls = len(net.calls) step(eng, i + 1, at_r(j.current_trade(v), 0.5)) self.assertEqual(len(net.calls), calls) self.assertEqual(watch_logs(j), []) def test_once_per_candle(self): eng, j, v, net = setup("shadow", watch_reply("HOLD")) i = open_trade(eng, j, v) t = j.current_trade(v) step(eng, i + 1, at_r(t, 0.1)); step(eng, i + 1, at_r(t, 0.1)); step(eng, i + 1, at_r(t, 0.1)) self.assertEqual(len(watch_logs(j)), 1) step(eng, i + 2, at_r(t, 0.1)) self.assertEqual(len(watch_logs(j)), 2) def test_active_tighten_reaches_the_broker(self): eng, j, v, net = setup("active", watch_reply("TIGHTEN", "breakeven", 80, ("news in 10 minutes",))) i = open_trade(eng, j, v) t = j.current_trade(v) sign = 1 if t["side"] == "BUY" else -1 msgs = step(eng, i + 1, at_r(t, 0.5)) self.assertTrue(any("AI watch: stop tightened" in m for m in msgs), msgs) new = j.current_trade(v)["sl"] self.assertGreater(sign * (new - t["sl"]), 0) # safer than before self.assertAlmostEqual(new, t["entry"] + sign * 1.5 * 0.0001, places=5) # break-even + the spread cushion self.assertEqual(fake_mt5.positions_get(ticket=int(t["ticket"]))[0].sl, round(new, 5)) def test_it_can_never_loosen_a_stop(self): eng, j, v, net = setup("active", watch_reply("TIGHTEN", "breakeven", 80)) i = open_trade(eng, j, v) t = j.current_trade(v) step(eng, i + 1, at_r(t, 0.5)) be = j.current_trade(v)["sl"] net.reply = watch_reply("TIGHTEN", "half_risk", 95) # half_risk is LOOSER than break-even step(eng, i + 2, at_r(t, 0.5)) self.assertEqual(j.current_trade(v)["sl"], be) self.assertIn("not allowed", watch_logs(j)[0]["applied"]) def test_break_even_needs_real_profit(self): eng, j, v, net = setup("active", watch_reply("TIGHTEN", "breakeven", 90)) i = open_trade(eng, j, v) t = j.current_trade(v) step(eng, i + 1, at_r(t, 0.1)) self.assertEqual(j.current_trade(v)["sl"], t["sl"]) net.reply = watch_reply("TIGHTEN", "lock_half_r", 90) step(eng, i + 2, at_r(t, 0.6)) # lock needs 1R of profit self.assertEqual(j.current_trade(v)["sl"], t["sl"]) def test_half_risk_is_allowed_when_the_trade_is_under_water(self): eng, j, v, net = setup("active", watch_reply("TIGHTEN", "half_risk", 80)) i = open_trade(eng, j, v) t = j.current_trade(v) sign = 1 if t["side"] == "BUY" else -1 step(eng, i + 1, at_r(t, -0.2)) new = j.current_trade(v)["sl"] self.assertGreater(sign * (new - t["sl"]), 0) R = abs(t["tp"] - t["entry"]) / 2 self.assertAlmostEqual(new, t["entry"] - sign * 0.5 * R, places=5) def test_stop_must_keep_room_from_price(self): eng, j, v, net = setup("active", watch_reply("TIGHTEN", "half_risk", 80)) i = open_trade(eng, j, v) t = j.current_trade(v) step(eng, i + 1, at_r(t, -0.4)) # price is only 0.1R from where half_risk would sit self.assertEqual(j.current_trade(v)["sl"], t["sl"]) def test_exit_needs_high_confidence(self): eng, j, v, net = setup("active", watch_reply("EXIT", "", 60)) i = open_trade(eng, j, v) step(eng, i + 1, at_r(j.current_trade(v), -0.2)) self.assertIsNotNone(j.current_trade(v)) self.assertEqual(len(fake_mt5.positions_get(symbol="EURUSD")), 1) def test_exit_with_high_confidence_closes_the_trade(self): eng, j, v, net = setup("active", watch_reply("EXIT", "", 90, ("momentum reversed",))) i = open_trade(eng, j, v) t = j.current_trade(v) msgs = step(eng, i + 1, at_r(t, -0.2)) self.assertIsNone(j.current_trade(v)) self.assertEqual(j.trades(v)[0]["exit_reason"], "AI_EXIT") self.assertEqual(len(fake_mt5.positions_get(symbol="EURUSD")), 0) self.assertTrue(any("AI exit" in m for m in msgs)) def test_cannot_exit_a_trade_younger_than_one_candle(self): eng, j, v, net = setup("active", watch_reply("EXIT", "", 99)) i = open_trade(eng, j, v) t = j.current_trade(v) win = DF.iloc[i - 198: i + 2] young = pd.Timestamp(t["entry_time"]) + pd.Timedelta(minutes=5) self.assertFalse(eng.guard.watch(eng, t, win, at_r(t, -0.2), young, [])) self.assertIsNotNone(j.current_trade(v)) def test_garbage_or_outage_changes_nothing(self): eng, j, v, net = setup("active", "I think you should sell everything!") i = open_trade(eng, j, v) t = j.current_trade(v) step(eng, i + 1, at_r(t, 0.5)) self.assertEqual(j.current_trade(v)["sl"], t["sl"]) net.groq = lambda body: (_ for _ in ()).throw(llm.LLMError("HTTP 503")) step(eng, i + 2, at_r(t, 0.5)) self.assertIsNotNone(j.current_trade(v)) self.assertEqual(j.count_events("CRASH"), 0) def test_limited_number_of_tightens_per_trade(self): eng, j, v, net = setup("active", watch_reply("TIGHTEN", "half_risk", 80)) i = open_trade(eng, j, v) t = j.current_trade(v) sign = 1 if t["side"] == "BUY" else -1 R = abs(t["tp"] - t["entry"]) / 2 applied = 0 for k, r in enumerate((-0.3, 0.9, 1.1, 1.3, 1.5), start=1): net.reply = watch_reply("TIGHTEN", "half_risk" if k == 1 else "lock_half_r", 80) before = j.current_trade(v)["sl"] step(eng, i + k, at_r(t, r)) now = j.current_trade(v) if now is None: break applied += 1 if now["sl"] != before else 0 self.assertLessEqual(applied, 3) if __name__ == "__main__": unittest.main(verbosity=2)