import io
import json
import os
import shutil
import subprocess
import sys
import tempfile
import unittest
from contextlib import redirect_stdout
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, guard, mt5conn
from sailar.brain import llm
from sailar.broker import Mt5Broker
from sailar.engine import Engine
from sailar.guard import Guard
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)
NOW = pd.Timestamp("2026-10-23 12:00", tz="UTC")


def build(spread=0.0001, params=None):
    mt5conn.reset()
    fake_mt5.reset()
    fake_mt5.S["spread"] = spread
    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 run_until_trade(eng, j, v, stop=1400):
    for i in range(200, stop):
        win = DF.iloc[i - 199: i + 1]
        fake_mt5.set_bid(float(win.close.iloc[-1]))
        eng.step(win, win.index[-1], price=float(win.close.iloc[-1]))
        if j.current_trade(v):
            return i
    return None


def add_closed(j, v, pnl, exit_time, entry_time=None):
    tid = j.add_trade(v, "BUY", entry_time or exit_time, 1.1, 1.0, 1.09, 1.12, 100.0, 70, {}, 0.0, "1")
    j.close_trade(tid, exit_time, 1.1, pnl, 0.0, "SL" if pnl < 0 else "TP", 2)


class Rules(unittest.TestCase):
    def eng_for_rules(self, spread_pips=1.0, pip=0.0001):
        j = Journal(":memory:")
        v = j.new_version(PARAMS, 100000)
        b = NS(spread_pips=lambda s: spread_pips, pip_size=lambda s: pip)
        return NS(j=j, v=v, b=b, symbol="EURUSD"), j, v

    def test_wide_spread_is_refused(self):
        e, j, v = self.eng_for_rules(spread_pips=3.0)
        ok, why = Guard(CFG, PARAMS).entry_check(e, None, 0.0020, NOW)
        self.assertFalse(ok)
        self.assertIn("above the 2.0 pip limit", why)

    def test_stop_must_be_big_enough_for_the_spread(self):
        e, j, v = self.eng_for_rules(spread_pips=1.6)
        ok, why = Guard(CFG, PARAMS).entry_check(e, 0.0008, 0.0008, NOW)       # 8 pip stop, 1.6 pip spread
        self.assertFalse(ok)
        self.assertIn("too tight", why)
        self.assertTrue(Guard(CFG, PARAMS).entry_check(e, 0.0014, 0.0014, NOW)[0])      # 14 pips is fine
        e2, *_ = self.eng_for_rules(spread_pips=0.8)
        self.assertTrue(Guard(CFG, PARAMS).entry_check(e2, 0.0008, 0.0008, NOW)[0])    # same 8 pip stop, normal spread: allowed

    def test_losing_streak_pauses_then_resumes(self):
        e, j, v = self.eng_for_rules()
        for k in range(3):
            add_closed(j, v, -100, NOW - pd.Timedelta(minutes=30 - k))
        g = Guard(CFG, PARAMS)
        ok, why = g.entry_check(e, None, 0.0020, NOW)
        self.assertFalse(ok)
        self.assertIn("3 losing trades in a row", why)
        self.assertTrue(g.entry_check(e, None, 0.0020, NOW + pd.Timedelta(hours=3))[0])      # pause is over

    def test_a_win_breaks_the_streak(self):
        e, j, v = self.eng_for_rules()
        add_closed(j, v, -100, NOW - pd.Timedelta(minutes=40))
        add_closed(j, v, +150, NOW - pd.Timedelta(minutes=30))
        add_closed(j, v, -100, NOW - pd.Timedelta(minutes=20))
        self.assertTrue(Guard(CFG, PARAMS).entry_check(e, None, 0.0020, NOW)[0])

    def test_daily_trade_cap(self):
        e, j, v = self.eng_for_rules()
        for k in range(12):
            add_closed(j, v, +10, NOW - pd.Timedelta(hours=1, minutes=k), NOW - pd.Timedelta(hours=2, minutes=k))
        ok, why = Guard(CFG, PARAMS).entry_check(e, None, 0.0020, NOW)
        self.assertFalse(ok)
        self.assertIn("daily limit", why)

    def test_rules_can_be_switched_off(self):
        e, j, v = self.eng_for_rules(spread_pips=5.0)
        off = {**PARAMS, "max_spread_pips": 0, "min_stop_spread_ratio": 0, "max_consec_losses": 0, "max_trades_per_day": 0}
        self.assertTrue(Guard(CFG, off).entry_check(e, None, 0.0001, NOW)[0])

    def test_break_even_sits_beyond_entry_by_the_spread(self):
        e, j, v = self.eng_for_rules(spread_pips=1.0)
        g = Guard(CFG, PARAMS)
        self.assertAlmostEqual(g.be_level(e, {"side": "BUY", "entry": 1.1000}), 1.10015, places=6)
        self.assertAlmostEqual(g.be_level(e, {"side": "SELL", "entry": 1.1000}), 1.09985, places=6)
        e2 = NS(b=NS(spread_pips=lambda s: None, pip_size=lambda s: None), symbol="X")
        self.assertEqual(g.be_level(e2, {"side": "BUY", "entry": 1.1}), 1.1)             # unknown spread: old behaviour


class WithEngine(unittest.TestCase):
    def test_wide_spread_means_no_trades_normal_spread_trades(self):
        eng, j, v = build(spread=0.0004)                  # 4 pips
        self.assertIsNone(run_until_trade(eng, j, v, stop=800))
        self.assertEqual(j.trades(v), [])
        eng2, j2, v2 = build(spread=0.0001)
        self.assertIsNotNone(run_until_trade(eng2, j2, v2))

    def test_break_even_move_reaches_the_broker_with_a_cushion(self):
        eng, j, v = build()
        i = run_until_trade(eng, j, v)
        t = j.current_trade(v)
        dist = abs(t["entry"] - t["sl"])
        sign = 1 if t["side"] == "BUY" else -1
        price = t["entry"] + sign * 0.9 * dist
        fake_mt5.set_bid(price)
        win = DF.iloc[i - 198: i + 2]
        msgs = eng.step(win, win.index[-1], price=price)
        self.assertTrue(any(m.startswith("BE:") for m in msgs), msgs)
        t2 = j.current_trade(v)
        cushion = 1.5 * 0.0001
        self.assertAlmostEqual(t2["sl"], t["entry"] + sign * cushion, places=5)
        self.assertAlmostEqual(fake_mt5.positions_get(ticket=int(t["ticket"]))[0].sl, round(t["entry"] + sign * cushion, 5), places=5)
        # and it is only moved once
        msgs2 = eng.step(win, win.index[-1], price=price)
        self.assertFalse(any(m.startswith("BE:") for m in msgs2))

    def test_trade_closed_by_hand_is_labelled_external_with_a_warning(self):
        eng, j, v = build()
        i = run_until_trade(eng, j, v)
        t = j.current_trade(v)
        p = fake_mt5.positions_get(ticket=int(t["ticket"]))[0]
        closing_buy = p.type == fake_mt5.POSITION_TYPE_SELL
        fake_mt5.order_send({"action": fake_mt5.TRADE_ACTION_DEAL, "symbol": "EURUSD", "volume": p.volume, "position": p.ticket,
                             "type": fake_mt5.ORDER_TYPE_BUY if closing_buy else fake_mt5.ORDER_TYPE_SELL, "magic": 0})
        win = DF.iloc[i - 198: i + 2]
        eng.step(win, win.index[-1], price=float(win.close.iloc[-1]))
        done = j.trades(v)[0]
        self.assertEqual(done["exit_reason"], "EXTERNAL")
        self.assertTrue(any("OUTSIDE Sailar-x" in e["msg"] for e in j.events(10, "WARN")))

    def test_stop_and_target_closes_keep_their_names(self):
        eng, j, v = build()
        i = run_until_trade(eng, j, v)
        t = j.current_trade(v)
        fake_mt5.trigger_sl(int(t["ticket"]))
        win = DF.iloc[i - 198: i + 2]
        eng.step(win, win.index[-1], price=float(win.close.iloc[-1]))
        self.assertEqual(j.trades(v)[0]["exit_reason"], "SL")


class OneCopyOnly(unittest.TestCase):
    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp())
        self.orig = guard.other_sailar_processes
        guard.other_sailar_processes = lambda: []

    def tearDown(self):
        guard.other_sailar_processes = self.orig
        shutil.rmtree(self.tmp, ignore_errors=True)

    def test_second_copy_is_refused_stale_lock_is_taken_over(self):
        lock = self.tmp / "sailar.lock"
        lock.write_text(f"{os.getppid()} 1")                      # a live process that is not us
        with self.assertRaises(SystemExit) as cm:
            guard.acquire_lock(lock)
        self.assertIn("ALREADY RUNNING", str(cm.exception))
        lock.write_text("4000000 1")                              # a process that does not exist
        guard.acquire_lock(lock)
        self.assertEqual(lock.read_text().split()[0], str(os.getpid()))
        guard._release(lock)
        self.assertFalse(lock.exists())

    def _fake_ps(self):
        fake_ps = ("  PID COMMAND\n 4242 python C:/other/folder/run.py run --config config.mt5.json\n"
                   " 4343 python something_else.py\n 4444 python run.py dashboard\n"
                   " 4545 python D:/crypto/run.py run --config config.crypto.mt5.json\n")
        real = subprocess.run
        subprocess.run = lambda *a, **k: NS(stdout=fake_ps, returncode=0)
        try:
            return guard.other_sailar_processes()
        finally:
            subprocess.run = real

    def test_copy_in_another_folder_with_the_same_config_is_detected(self):
        guard.other_sailar_processes = self.orig
        found = self._fake_ps()
        self.assertEqual(sorted(p for p, _ in found), [4242, 4545])      # the dashboard and unrelated scripts are not counted
        guard.other_sailar_processes = lambda: found
        old_argv = sys.argv
        sys.argv = ["run.py", "run", "--config", "config.mt5.json"]
        try:
            with self.assertRaises(SystemExit) as cm:
                guard.acquire_lock(self.tmp / "x.lock")
        finally:
            sys.argv = old_argv
        self.assertIn("SAME config", str(cm.exception))
        self.assertIn("4242", str(cm.exception))
        self.assertNotIn("4545", str(cm.exception))

    def test_forex_and_crypto_copies_can_run_side_by_side(self):
        guard.other_sailar_processes = self.orig
        found = self._fake_ps()
        guard.other_sailar_processes = lambda: found
        old_argv = sys.argv
        sys.argv = ["run.py", "run", "--config", "config.crypto.json"]      # a third, different config
        try:
            guard.acquire_lock(self.tmp / "y.lock")                          # no error
        finally:
            sys.argv = old_argv
        self.assertTrue((self.tmp / "y.lock").exists())
        guard._release(self.tmp / "y.lock")

    def test_config_name_parsing(self):
        self.assertEqual(guard._config_name(["run.py", "run", "--config", "C:\\x\\Config.MT5.json"]), "config.mt5.json")
        self.assertEqual(guard._config_name(["run.py", "run", "--config=a/b/c.json"]), "c.json")
        self.assertIsNone(guard._config_name(["run.py", "run"]))


class Patcher(unittest.TestCase):
    def test_idempotent_and_safe_when_a_spot_was_edited(self):
        tmp = Path(tempfile.mkdtemp())
        try:
            shutil.copytree(ROOT / "sailar", tmp / "sailar")
            r1 = subprocess.run([sys.executable, str(ROOT / "apply_guard_patch.py"), str(tmp)], capture_output=True, text=True)
            r2 = subprocess.run([sys.executable, str(ROOT / "apply_guard_patch.py"), str(tmp)], capture_output=True, text=True)
            self.assertIn("0 applied", r2.stdout)
            self.assertEqual(r2.returncode, 0, r2.stderr)
            # the user edited a spot: that one is skipped, nothing crashes, the rest still applies
            mini = tmp / "mini"
            (mini / "sailar").mkdir(parents=True)
            shutil.copy(ROOT / "sailar" / "guard.py", mini / "sailar" / "guard.py")
            (mini / "sailar" / "runner.py").write_text('from .brain import make_brain\n# user rewrote the Journal line\n')
            r3 = subprocess.run([sys.executable, str(ROOT / "apply_guard_patch.py"), str(mini)], capture_output=True, text=True)
            self.assertIn("DONE  sailar/runner.py: import the lock", r3.stdout)
            self.assertIn("SKIP  sailar/runner.py: one copy only", r3.stdout)
            self.assertEqual(r3.returncode, 0, r3.stderr)
            self.assertTrue(any((mini / "backup_before_guard").iterdir()))
        finally:
            shutil.rmtree(tmp, ignore_errors=True)


class BrainFixes(unittest.TestCase):
    def test_groq_json_failure_is_retried_without_json_mode(self):
        calls = []

        def net(method, url, headers, body, timeout):
            calls.append(json.loads(json.dumps(body)) if body else body)     # snapshot: the retry edits the same dict
            if url.endswith("/models"):
                return {"data": [{"id": "openai/gpt-oss-120b"}]}
            if body.get("response_format"):
                raise llm.LLMError('HTTP 400: {"error":{"code":"json_validate_failed","message":"Failed to generate JSON"}}')
            return {"choices": [{"message": {"content": '{"decision":"APPROVE"}'}}]}
        old = llm._http
        llm._http = net
        try:
            g = llm.Groq("k", ["openai/gpt-oss-120b"])
            text, model, _ = g.chat("s", "u", True, 8, 500, 0.2)
        finally:
            llm._http = old
        self.assertEqual(text, '{"decision":"APPROVE"}')
        post = [b for b in calls if b and "messages" in b]
        self.assertIn("response_format", post[0])
        self.assertNotIn("response_format", post[1])
        self.assertGreaterEqual(post[1]["max_tokens"], 1500)

    def test_failed_review_is_logged_as_error_not_approve(self):
        from sailar.brain.brain import Brain
        cfg = json.loads(json.dumps(CFG))
        cfg["brain"]["mode"] = "shadow"
        j = Journal(":memory:")
        j.new_version(PARAMS, 100000)

        def boom(method, url, headers, body, timeout):
            raise llm.LLMError("HTTP 503")
        old = llm._http
        llm._http = boom
        try:
            b = Brain(cfg, j, {"GROQ_API_KEY": "x"})
            vd = b.review({"symbol": "EURUSD"}, NOW, 1)
        finally:
            llm._http = old
        self.assertFalse(vd.ok)
        self.assertEqual(j.brain_logs("review")[0]["response"]["decision"], "ERROR")


class LossReport(unittest.TestCase):
    def test_report_explains_external_closes_and_mt5_audit(self):
        import loss_report
        eng, j, v = build()
        i = run_until_trade(eng, j, v)
        t = j.current_trade(v)
        p = fake_mt5.positions_get(ticket=int(t["ticket"]))[0]
        fake_mt5.order_send({"action": fake_mt5.TRADE_ACTION_DEAL, "symbol": "EURUSD", "volume": p.volume, "position": p.ticket,
                             "type": fake_mt5.ORDER_TYPE_BUY if p.type == fake_mt5.POSITION_TYPE_SELL else fake_mt5.ORDER_TYPE_SELL, "magic": 0})
        win = DF.iloc[i - 198: i + 2]
        eng.step(win, win.index[-1], price=float(win.close.iloc[-1]))
        notes, buf = [], io.StringIO()
        with redirect_stdout(buf):
            loss_report.journal_report(j, v, notes)
            loss_report.mt5_report(7, notes)
        out = buf.getvalue()
        self.assertIn("EXTERNAL", out)
        self.assertIn("by hand", out)
        self.assertTrue(any("outside Sailar-x" in n for n in notes))
        self.assertTrue(any("BY HAND" in n for n in notes))


if __name__ == "__main__":
    unittest.main(verbosity=2)