Codex
199 lines
7.9 KiB
Python
199 lines
7.9 KiB
Python
import json
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from scripts import run_backtest_batch
|
|
from scripts.run_backtest_batch import (
|
|
build_candidate_model_codes_from_env,
|
|
build_instrument_ids_from_env,
|
|
build_session_groups_from_env,
|
|
build_setup_playbook_ids_from_env,
|
|
build_subject_types_from_env,
|
|
build_timeframe_sets_from_env,
|
|
run_backtest_batch as run_batch,
|
|
)
|
|
from src.services.backtest import BacktestOutcome, BacktestQualityGate
|
|
|
|
|
|
class RunBacktestBatchScriptTests(unittest.TestCase):
|
|
def test_build_batch_dimensions_from_env(self) -> None:
|
|
env = {
|
|
"BACKTEST_BATCH_INSTRUMENT_IDS": "1,2",
|
|
"BACKTEST_BATCH_SUBJECT_TYPES": "plan_replay,signal",
|
|
"BACKTEST_BATCH_TIMEFRAME_SETS": "1h/15m/5m;15m/5m/1m",
|
|
"BACKTEST_BATCH_SESSION_GROUPS": "ALL;LONDON,NY_AM;ASIA",
|
|
"BACKTEST_BATCH_CANDIDATE_MODEL_CODES": "ALL,OSOK",
|
|
"BACKTEST_BATCH_SETUP_PLAYBOOK_IDS": "ALL,ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
|
|
}
|
|
|
|
self.assertEqual(build_instrument_ids_from_env(env), [1, 2])
|
|
self.assertEqual(build_subject_types_from_env(env), ["plan_replay", "signal"])
|
|
self.assertEqual(
|
|
build_timeframe_sets_from_env(env),
|
|
[
|
|
{"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"},
|
|
{"bias_tf": "15m", "setup_tf": "5m", "trigger_tf": "1m"},
|
|
],
|
|
)
|
|
self.assertEqual(build_session_groups_from_env(env), [[], ["LONDON", "NY_AM"], ["ASIA"]])
|
|
self.assertEqual(build_candidate_model_codes_from_env(env), [None, "OSOK"])
|
|
self.assertEqual(build_setup_playbook_ids_from_env(env), [None, "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL"])
|
|
|
|
def test_run_batch_aggregates_quality_gate_and_metadata(self) -> None:
|
|
service = FakeBatchBacktestService(
|
|
{
|
|
"plan_replay": [_win_outcome(signal_id=None, trading_plan_id=10)],
|
|
"signal": [_loss_outcome(signal_id=1)],
|
|
}
|
|
)
|
|
|
|
payload = run_batch(
|
|
service=service,
|
|
batch_id="batch-test",
|
|
instrument_ids=[1],
|
|
limit=25,
|
|
subject_types=["plan_replay", "signal"],
|
|
modes=["replay_no_lookahead"],
|
|
timeframe_sets=[{"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"}],
|
|
session_groups=[[], ["LONDON"]],
|
|
candidate_model_codes=[None],
|
|
setup_playbook_ids=["ICT-CH1-6-LONDON-DISCOUNT-REVERSAL"],
|
|
require_quality_gate=False,
|
|
quality_gate=BacktestQualityGate(min_total_signals=1, min_executed_signals=1, min_profit_factor_r=0.0),
|
|
cost_config=None,
|
|
replay_config=None,
|
|
)
|
|
|
|
self.assertEqual(payload["status"], "completed")
|
|
self.assertEqual(payload["run_count"], 3)
|
|
self.assertEqual(payload["total_result_count"], 3)
|
|
self.assertEqual(payload["aggregate_by_subject"]["plan_replay"]["run_count"], 1)
|
|
self.assertEqual(payload["aggregate_by_subject"]["signal"]["run_count"], 2)
|
|
self.assertIn("ICT-CH1-6-LONDON-DISCOUNT-REVERSAL::complete", payload["aggregate_by_doctrine_cohort"])
|
|
d8_summary = payload["playbook_historical_cohort_summary"]
|
|
self.assertEqual(d8_summary["phase"], "D8_PLAYBOOK_SPECIFIC_HISTORICAL_COHORT_GATE")
|
|
self.assertEqual(d8_summary["run_count"], 3)
|
|
self.assertEqual(d8_summary["cohort_count"], 2)
|
|
self.assertEqual(d8_summary["manual_review_eligible_count"], 0)
|
|
self.assertEqual(d8_summary["study_only_count"], 2)
|
|
self.assertTrue(all(item["status"] == "study_only" for item in d8_summary["cohorts"]))
|
|
self.assertTrue(
|
|
any(
|
|
item["cohort_key"] == "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL::LONDON::1h/15m/5m::london_discount_reversal"
|
|
for item in d8_summary["cohorts"]
|
|
)
|
|
)
|
|
d9_summary = payload["playbook_remediation_queue_summary"]
|
|
self.assertEqual(d9_summary["phase"], "D9_PLAYBOOK_REMEDIATION_QUEUE")
|
|
self.assertEqual(d9_summary["queue_count"], 2)
|
|
self.assertEqual(d9_summary["strategy_backtest_route_count"] + d9_summary["study_notes_route_count"], 2)
|
|
self.assertTrue(all(item["primary_route"] in {"strategy_backtest", "study_notes"} for item in d9_summary["items"]))
|
|
self.assertTrue(d9_summary["operator_summary"].startswith("2 failed cohorts queued"))
|
|
self.assertEqual(service.calls[0]["run_metadata"]["batch_id"], "batch-test")
|
|
self.assertEqual(service.calls[0]["run_metadata"]["setup_playbook_id"], "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL")
|
|
self.assertEqual(service.calls[0]["run_metadata"]["execution_boundary"], "read_only_backtest_no_order_submission")
|
|
self.assertEqual(payload["execution_boundary"], "read_only_backtest_no_order_submission")
|
|
|
|
def test_main_outputs_json_payload(self) -> None:
|
|
service = FakeBatchBacktestService({"plan_replay": [_win_outcome(signal_id=None, trading_plan_id=10)]})
|
|
output = _Capture()
|
|
|
|
original_service = run_backtest_batch.BacktestService
|
|
run_backtest_batch.BacktestService = lambda: service
|
|
try:
|
|
with output:
|
|
exit_code = run_backtest_batch.main(
|
|
env={
|
|
"BACKTEST_BATCH_OUTPUT_JSON": "1",
|
|
"BACKTEST_BATCH_ID": "batch-json",
|
|
"BACKTEST_BATCH_SUBJECT_TYPES": "plan_replay",
|
|
"BACKTEST_BATCH_SESSION_GROUPS": "ALL",
|
|
"BACKTEST_MIN_TOTAL_SIGNALS": "1",
|
|
"BACKTEST_MIN_EXECUTED_SIGNALS": "1",
|
|
"BACKTEST_MIN_PROFIT_FACTOR_R": "0",
|
|
}
|
|
)
|
|
finally:
|
|
run_backtest_batch.BacktestService = original_service
|
|
|
|
payload = json.loads(output.value)
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertEqual(payload["batch_id"], "batch-json")
|
|
self.assertEqual(payload["run_count"], 1)
|
|
|
|
|
|
class FakeBatchBacktestService:
|
|
def __init__(self, outcomes_by_subject: dict[str, list[BacktestOutcome]]) -> None:
|
|
self.outcomes_by_subject = outcomes_by_subject
|
|
self.calls = []
|
|
|
|
def run_backtest(self, **kwargs):
|
|
self.calls.append(kwargs)
|
|
return 100 + len(self.calls), list(self.outcomes_by_subject.get(kwargs["subject_type"]) or [])
|
|
|
|
|
|
class _Capture:
|
|
def __enter__(self):
|
|
import contextlib
|
|
import io
|
|
|
|
self._stream = io.StringIO()
|
|
self._context = contextlib.redirect_stdout(self._stream)
|
|
self._context.__enter__()
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
self._context.__exit__(*args)
|
|
self.value = self._stream.getvalue()
|
|
|
|
|
|
def _win_outcome(*, signal_id, trading_plan_id=None) -> BacktestOutcome:
|
|
return BacktestOutcome(
|
|
signal_id=signal_id,
|
|
outcome="win",
|
|
mfe=2.0,
|
|
mae=0.0,
|
|
tp1_hit=True,
|
|
tp2_hit=True,
|
|
invalidated_before_entry=False,
|
|
error_tags=[],
|
|
trading_plan_id=trading_plan_id,
|
|
meta={
|
|
"signal_side": "buy",
|
|
"signal_symbol": "BTC-USDT",
|
|
"session_codes": ["LONDON"],
|
|
"weekday": "Thursday",
|
|
"mfe_r": 2.0,
|
|
"mae_r": 0.0,
|
|
"r_multiple": 2.0,
|
|
"cost_r": 0.0,
|
|
},
|
|
)
|
|
|
|
|
|
def _loss_outcome(*, signal_id) -> BacktestOutcome:
|
|
return BacktestOutcome(
|
|
signal_id=signal_id,
|
|
outcome="loss",
|
|
mfe=0.2,
|
|
mae=1.0,
|
|
tp1_hit=False,
|
|
tp2_hit=False,
|
|
invalidated_before_entry=False,
|
|
error_tags=["stopped_out"],
|
|
meta={
|
|
"signal_side": "sell",
|
|
"signal_symbol": "BTC-USDT",
|
|
"session_codes": ["NY_AM"],
|
|
"weekday": "Thursday",
|
|
"mfe_r": 0.2,
|
|
"mae_r": 1.0,
|
|
"r_multiple": -1.0,
|
|
"cost_r": 0.0,
|
|
},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|