import os import contextlib import io import json import tempfile import unittest from pathlib import Path from unittest.mock import patch os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test") from src.services.backtest import detect_confirmed_swings_for_replay, iter_historical_replay, load_replay_candles, validate_replay_candles from scripts import run_replay_file class HistoricalReplayTests(unittest.TestCase): def test_frames_only_expose_known_candles(self) -> None: candles = [_candle(idx, high=100 + idx, low=90 - idx) for idx in range(4)] frames = list(iter_historical_replay(candles, left_bars=1, right_bars=1)) self.assertEqual(len(frames), 4) for frame in frames: self.assertEqual(len(frame.known_candles), frame.index + 1) self.assertEqual(frame.known_candles[-1], frame.candle) def test_swing_confirms_only_after_right_window_exists(self) -> None: candles = [ _candle(0, high=100, low=90), _candle(1, high=105, low=91), _candle(2, high=101, low=92), ] frames = list(iter_historical_replay(candles, left_bars=1, right_bars=1)) self.assertEqual(frames[1].new_swings, ()) self.assertEqual(len(frames[2].new_swings), 1) swing = frames[2].new_swings[0] self.assertEqual(swing.kind, "swing_high") self.assertEqual(swing.source_index, 1) self.assertEqual(swing.confirmed_at_index, 2) def test_detects_confirmed_highs_and_lows_without_future_leakage(self) -> None: candles = [ _candle(0, high=100, low=95), _candle(1, high=105, low=96), _candle(2, high=101, low=94), _candle(3, high=102, low=90), _candle(4, high=103, low=93), ] swings = detect_confirmed_swings_for_replay(candles, left_bars=1, right_bars=1) self.assertEqual([(s.kind, s.source_index, s.confirmed_at_index) for s in swings], [("swing_high", 1, 2), ("swing_low", 3, 4)]) def test_validates_replay_candle_integrity(self) -> None: candles = [ _candle(0, high=100, low=95), _candle(0, high=101, low=96), {"ts_open": 2, "open": 110, "high": 105, "low": 100, "close": 104}, {"ts_open": 3, "open": 100, "high": 100, "low": 101, "close": 106}, {"ts_open": 4, "open": None, "high": 105, "low": 101, "close": 103}, ] result = validate_replay_candles(candles) self.assertFalse(result.is_valid) self.assertIn("duplicate_ts_open:1", result.errors) self.assertIn("non_increasing_ts_open:1", result.errors) self.assertIn("open_outside_range:2", result.errors) self.assertIn("high_below_low:3", result.errors) self.assertIn("close_outside_range:3", result.errors) self.assertIn("missing_open:4", result.errors) def test_validates_replay_price_quality(self) -> None: candles = [ {"ts_open": 1, "open": "bad", "high": 101, "low": 99, "close": 100}, {"ts_open": 2, "open": 100, "high": float("inf"), "low": 99, "close": 100}, {"ts_open": 3, "open": 100, "high": 101, "low": -1, "close": 100}, ] result = validate_replay_candles(candles) self.assertFalse(result.is_valid) self.assertIn("non_numeric_open:0", result.errors) self.assertIn("non_finite_high:1", result.errors) self.assertIn("non_positive_low:2", result.errors) def test_load_replay_candles_reports_non_numeric_prices(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "candles.jsonl" path.write_text('{"ts_open":1,"open":"bad","high":101,"low":99,"close":100}\n', encoding="utf-8") with self.assertRaisesRegex(ValueError, "non_numeric_open:0"): load_replay_candles(path) def test_replay_rejects_invalid_candles_before_iterating(self) -> None: candles = [{"ts_open": 1, "open": 100, "high": 99, "low": 101, "close": 100}] with self.assertRaisesRegex(ValueError, "invalid_replay_candles"): list(iter_historical_replay(candles)) def test_loads_replay_candles_from_json_items_payload(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "candles.json" path.write_text('{"items":[{"ts_open":"1","open":"100","high":"101","low":"99","close":"100.5"}]}', encoding="utf-8") candles = load_replay_candles(path) self.assertEqual(candles[0]["ts_open"], 1) self.assertEqual(candles[0]["open"], 100.0) def test_loads_replay_candles_from_jsonl(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "candles.jsonl" path.write_text( '{"ts_open":1,"open":100,"high":101,"low":99,"close":100}\n' '{"ts_open":2,"open":101,"high":102,"low":100,"close":101}\n', encoding="utf-8", ) candles = load_replay_candles(path) self.assertEqual(len(candles), 2) self.assertEqual(candles[1]["high"], 102.0) def test_loads_replay_candles_from_csv(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "candles.csv" path.write_text("ts_open,open,high,low,close\n1,100,101,99,100\n2,101,102,100,101\n", encoding="utf-8") candles = load_replay_candles(path) self.assertEqual(len(candles), 2) self.assertEqual(candles[0]["low"], 99.0) def test_fixture_replay_sample_has_two_confirmed_swings(self) -> None: fixture = Path(__file__).resolve().parents[1] / "fixtures" / "offline_replay_sample.jsonl" candles = load_replay_candles(fixture) swings = detect_confirmed_swings_for_replay(candles, left_bars=1, right_bars=1) self.assertEqual(len(candles), 5) self.assertEqual([(s.kind, s.source_index, s.confirmed_at_index) for s in swings], [("swing_high", 1, 2), ("swing_low", 3, 4)]) def test_p5_replay_fixture_has_confirmed_swings_with_default_window(self) -> None: fixture = Path(__file__).resolve().parents[2] / "data" / "replay" / "research_council_confirmed_swings_v1.jsonl" candles = load_replay_candles(fixture) swings = detect_confirmed_swings_for_replay(candles) self.assertEqual(len(candles), 9) self.assertEqual([(s.kind, s.source_index, s.confirmed_at_index) for s in swings], [("swing_high", 2, 4), ("swing_low", 6, 8)]) def test_load_replay_candles_rejects_unsupported_format(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "candles.txt" path.write_text("", encoding="utf-8") with self.assertRaisesRegex(ValueError, "unsupported_replay_candle_file"): load_replay_candles(path) def test_run_replay_file_reports_summary(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "candles.jsonl" path.write_text( '{"ts_open":1,"open":97,"high":100,"low":95,"close":98}\n' '{"ts_open":2,"open":100,"high":105,"low":96,"close":102}\n' '{"ts_open":3,"open":99,"high":101,"low":94,"close":96}\n', encoding="utf-8", ) output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = run_replay_file.main([str(path)], env={"REPLAY_LEFT_BARS": "1", "REPLAY_RIGHT_BARS": "1"}) self.assertEqual(exit_code, 0) self.assertIn("Replay candles: 3", output.getvalue()) self.assertIn("Replay sample: first_ts=1 last_ts=3", output.getvalue()) self.assertIn("Replay confirmed swings: 1", output.getvalue()) self.assertIn("Replay quality gate passed: True", output.getvalue()) def test_run_replay_file_can_emit_json_summary(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "candles.jsonl" path.write_text( '{"ts_open":1,"open":97,"high":100,"low":95,"close":98}\n' '{"ts_open":2,"open":100,"high":105,"low":96,"close":102}\n' '{"ts_open":3,"open":99,"high":101,"low":94,"close":96}\n', encoding="utf-8", ) output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = run_replay_file.main( [str(path)], env={ "REPLAY_LEFT_BARS": "1", "REPLAY_RIGHT_BARS": "1", "REPLAY_OUTPUT_JSON": "1", }, ) summary = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertEqual(summary["candle_count"], 3) self.assertEqual(summary["sample"]["first_ts_open"], 1) self.assertEqual(summary["sample"]["last_ts_open"], 3) self.assertEqual(summary["sample"]["highest_high"], 105.0) self.assertEqual(summary["sample"]["lowest_low"], 94.0) self.assertAlmostEqual(summary["sample"]["max_gap_minutes"], 1 / 60) self.assertEqual(summary["swing_counts"]["swing_high"], 1) self.assertTrue(summary["quality_gate"]["passed"]) def test_replay_sample_summary_handles_empty_samples(self) -> None: summary = run_replay_file.build_replay_sample_summary([]) self.assertIsNone(summary["first_ts_open"]) self.assertIsNone(summary["close_return_pct"]) self.assertIsNone(summary["max_gap_minutes"]) def test_run_replay_file_blocks_when_gap_gate_fails(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "candles.jsonl" path.write_text( '{"ts_open":"2026-04-29T08:00:00+00:00","open":97,"high":100,"low":95,"close":98}\n' '{"ts_open":"2026-04-29T08:01:00+00:00","open":100,"high":105,"low":96,"close":102}\n' '{"ts_open":"2026-04-29T08:10:00+00:00","open":99,"high":101,"low":94,"close":96}\n', encoding="utf-8", ) output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = run_replay_file.main( [str(path)], env={ "REPLAY_LEFT_BARS": "1", "REPLAY_RIGHT_BARS": "1", "REPLAY_MAX_GAP_MINUTES": "2", }, ) self.assertEqual(exit_code, 2) self.assertIn("Replay max gap minutes: 9.0", output.getvalue()) self.assertIn("candle_gap_too_large", output.getvalue()) def test_run_replay_file_blocks_when_replay_gate_fails(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "candles.jsonl" path.write_text( '{"ts_open":1,"open":97,"high":100,"low":95,"close":98}\n' '{"ts_open":2,"open":100,"high":105,"low":96,"close":102}\n' '{"ts_open":3,"open":99,"high":101,"low":94,"close":96}\n', encoding="utf-8", ) output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = run_replay_file.main( [str(path)], env={ "REPLAY_LEFT_BARS": "1", "REPLAY_RIGHT_BARS": "1", "REPLAY_MIN_CANDLES": "10", "REPLAY_MIN_SWINGS": "2", }, ) self.assertEqual(exit_code, 2) self.assertIn("candle_sample_too_small", output.getvalue()) self.assertIn("swing_sample_too_small", output.getvalue()) def test_run_replay_file_rejects_missing_path(self) -> None: output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = run_replay_file.main([]) self.assertEqual(exit_code, 2) self.assertIn("Usage:", output.getvalue()) def test_run_replay_file_empty_env_ignores_ambient_json_setting(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "candles.jsonl" path.write_text( '{"ts_open":1,"open":97,"high":100,"low":95,"close":98}\n' '{"ts_open":2,"open":100,"high":105,"low":96,"close":102}\n' '{"ts_open":3,"open":99,"high":101,"low":94,"close":96}\n', encoding="utf-8", ) output = io.StringIO() with patch.dict(os.environ, {"REPLAY_OUTPUT_JSON": "1"}, clear=False), contextlib.redirect_stdout(output): exit_code = run_replay_file.main([str(path)], env={}) self.assertEqual(exit_code, 0) self.assertIn("Replay candles: 3", output.getvalue()) self.assertNotIn('"candle_count"', output.getvalue()) def _candle(idx: int, high: float, low: float) -> dict: return { "ts_open": idx, "ts_close": idx + 1, "open": (high + low) / 2, "high": high, "low": low, "close": (high + low) / 2, } if __name__ == "__main__": unittest.main()