Codex
314 lines
13 KiB
Python
314 lines
13 KiB
Python
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()
|